Added warning when dpnode buffers don't empty.

If the dpnode is receiving too many points, the worker socket buffers will never be empty since the event loop cannot grab data fast enough from the socket buffer. Now, whenever the worker event loop buffer is empty, it will check how much data is in the socket buffer for each worker and if it has not been empty in the last 30 seconds, it will print a warning message.

dpnode.cc

@@ -6,6 +6,7 @@ extern "C" {
 }
 
 #include <sys/socket.h>
+#include <sys/ioctl.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 
@@ -17,6 +18,7 @@ extern "C" {
 #include <stdlib.h>
 #include <string.h>
 #include <errno.h>
+#include <time.h>
 
 #include <sstream>
 #include <fstream>
@@ -31,6 +33,10 @@ extern "C" {
 static ofstream dp_file_stream;
 #endif
 
+time_t last_time_buffer_empty = 0;
+time_t last_time_buffer_warning = 0;
+// for the buffer warning messages
+
 typedef unordered_map<std::string, pair<ZZ,ZZ> > DTable;
 
 typedef enum {
@@ -401,8 +407,38 @@ static void controllerconn_event_cb(struct bufferevent *bev, short events,
     }
 }
 
+void ev_loop_empty_cb()
+{
+    time_t raw_time;
+    time(&raw_time);
+
+    for (auto worker : dpctrlstate.workers) {
+	int bytes = 0;
+	if (ioctl(bufferevent_getfd(worker), FIONREAD, &bytes) == 0) {
+	    if (bytes == 0) {
+		last_time_buffer_empty = raw_time;
+		// last time any of the socket buffers were empty
+	    }
+	}
+    }
+
+    if (dpctrlstate.workers.size() == 0) {
+	// there are no buffers
+	last_time_buffer_empty = raw_time;
+    }
+
+    if (raw_time-last_time_buffer_warning > 60 && raw_time-last_time_buffer_empty > 30) {
+	// if we haven't shown a message in the last minute, and haven't had an empty buffer in the last 30 seconds
+	// the dpnode probably can't keep up with the data from one of the workers
+	char time_str[100];
+	strftime(time_str, sizeof(time_str), "%c", std::localtime(&raw_time));
+	cerr << "No worker data buffers have been empty in " << raw_time-last_time_buffer_empty << " seconds... (" << time_str << ")\n" << flush;
+	last_time_buffer_warning = raw_time;
+    }
+}
+
 int dpnode_main(const char *controller_host, unsigned short controller_port)
 {
     return controller_client(controller_host, controller_port,
-				controllerconn_event_cb, false);
+				controllerconn_event_cb, false, &ev_loop_empty_cb);
 }

evutils.cc

@@ -107,7 +107,7 @@ struct bufferevent *client_create(struct event_base *evbase,
 // when the program is finished.
 int controller_client(const char *controller_host,
     unsigned short controller_port, bufferevent_event_cb event_handler,
-    bool multithread)
+    bool multithread, void (*ev_loop_empty_cb)())
 {
     struct event_base *evbase = event_base_new();
 
@@ -133,7 +133,17 @@ int controller_client(const char *controller_host,
 
     bufferevent_setcb(controller_bev, NULL, NULL, event_handler, NULL);
 
-    event_base_dispatch(evbase);
+    if (ev_loop_empty_cb == NULL) {
+	event_base_dispatch(evbase);
+    } else {
+	while (true) {
+	    event_base_loop(evbase, EVLOOP_ONCE);
+	    ev_loop_empty_cb();
+	    if (event_base_got_exit(evbase) || event_base_got_break(evbase)) {
+		break;
+	    }
+	}
+    }
 
     return 0;
 }

evutils.h

@@ -28,6 +28,6 @@ struct bufferevent *client_create(struct event_base *evbase,
 // when the program is finished.
 int controller_client(const char *controller_host,
     unsigned short controller_port, bufferevent_event_cb event_handler,
-    bool multithread);
+    bool multithread, void (*ev_loop_empty_cb)());
 
 #endif

worker.cc

@@ -347,5 +347,5 @@ int worker_main(const char *controller_host, unsigned short controller_port, int
 	cuda_device_id = gpu_id;
 
     return controller_client(controller_host, controller_port,
-				controllerconn_event_cb, true);
+				controllerconn_event_cb, true, NULL);
 }