Log and plot the times ingestion starts and ends

App/net.cpp

@@ -264,6 +264,22 @@ void NetIO::ing_receive_msgbundle(tcp::socket* csocket, clientid_t c_simid)
             return;
         }
 
+#ifdef TRACE_SOCKIO
+        struct timeval now;
+        gettimeofday(&now, NULL);
+        long elapsedus = (now.tv_sec - last_ing.tv_sec) * 1000000
+            + (now.tv_usec - last_ing.tv_usec);
+        if (num_ing > 0 && elapsedus > 500000) {
+            printf("%lu.%06lu: End ingestion of %lu messages\n",
+                last_ing.tv_sec, last_ing.tv_usec, num_ing);
+            num_ing = 0;
+        }
+        if (num_ing == 0) {
+            printf("%lu.%06lu: Begin ingestion\n", now.tv_sec,
+                now.tv_usec);
+        }
+#endif
+
         bool ret;
         //Ingest the message_bundle
         if(conf.private_routing) {
@@ -273,6 +289,11 @@ void NetIO::ing_receive_msgbundle(tcp::socket* csocket, clientid_t c_simid)
         }
         free(msgbundle);
 
+#ifdef TRACE_SOCKIO
+        gettimeofday(&last_ing, NULL);
+        ++num_ing;
+#endif
+
         // Continue to async receive client message bundles
         if(ret) {
             ing_receive_msgbundle(csocket, c_simid);
@@ -486,6 +507,11 @@ NetIO::NetIO(boost::asio::io_context &io_context, const Config &config)
     nodeios.resize(num_nodes);
     me = conf.my_node_num;
 
+#ifdef TRACE_SOCKIO
+    last_ing = {0, 0};
+    num_ing = 0;
+#endif
+
     // Node number n will accept connections from nodes 0, ..., n-1 and
     // make connections to nodes n+1, ..., num_nodes-1.  This is all
     // single threaded, but it doesn't deadlock because node 0 isn't

App/net.hpp

@@ -180,6 +180,12 @@ public:
     void close();
 
     uint64_t reset_bytes_sent();
+
+#ifdef TRACE_SOCKIO
+private:
+    struct timeval last_ing;
+    size_t num_ing;
+#endif
 };
 
 extern NetIO *g_netio;

plot_traces

@@ -104,6 +104,22 @@ for nodelog in nodelogs:
                 labels[node].append((tsf, typ))
                 if max_ts is None or max_ts < tsf:
                     max_ts = tsf
+            elif matches := re.match(
+                    r'(\d+\.\d+): Begin ingestion',
+                    logline):
+                [ts] = matches.groups()
+                tsf = float(ts)
+                labels[node].append((tsf, 'i'))
+                if max_ts is None or max_ts < tsf:
+                    max_ts = tsf
+            elif matches := re.match(
+                    r'(\d+\.\d+): End ingestion of \d+ messages',
+                    logline):
+                [ts] = matches.groups()
+                tsf = float(ts)
+                labels[node].append((tsf, 'I'))
+                if max_ts is None or max_ts < tsf:
+                    max_ts = tsf
 
 # Pass 2: For each sender and receiver, note the receive start time and
 # receive end time for each message in the queued_messages list