Push a turn's sends to a link at once, and round only the links that want one (017, 2b-1)

The sending loop was woken by every record queued while it slept: the
owner queues one by one, the loop caught up and slept between them, and
every link took a round on every wake. Four links sent 64k packets of
12 records where threads sent 40k of 20, and ran at 257k sends/s
against 581k. The owner now queues a turn's sends and pushes each link
once; the loop rounds a link only when its socket has answers, it was
pushed, or its time came.

2 files changed, 5 insertions, 3 deletions

diff --git a/research/experiments/019-idle-links/run.py b/research/experiments/019-idle-links/run.py
index 0d18c1d..bfe5b92 100644
--- a/research/experiments/019-idle-links/run.py
+++ b/research/experiments/019-idle-links/run.py
@@ -490,7 +490,7 @@ def drain_pure(work, n, size):
     b, _ = start([ROOT / 'build/pure-serve', '--module', receiver_module, '--store', work / 'drain-b', '--port', port_b,
                   '--threads', '2', '--key', work / 'b.key', '--link-port', link_port, '--senders', work / 'drain-senders'])
     cpu = {name: (thread_cpu(pid, None), thread_cpu(pid, thread)) for name, pid, thread in
-           (('sender', a.pid, 'pure-link'), ('receiver', b.pid, 'pure-listen'))}
+           (('sender', a.pid, 'pure-send'), ('receiver', b.pid, 'pure-listen'))}
     import http.client
     connection = http.client.HTTPConnection('127.0.0.1', port_b, timeout=30)
     first = None
@@ -507,7 +507,7 @@ def drain_pure(work, n, size):
     seconds = now - first[0]
     row = {'messages': n, 'item_bytes': size, 'first_seen': first[1], 'seconds': round(seconds, 4),
            'per_second': round((n - first[1]) / seconds) if seconds else None}
-    for name, pid, thread in (('sender', a.pid, 'pure-link'), ('receiver', b.pid, 'pure-listen')):
+    for name, pid, thread in (('sender', a.pid, 'pure-send'), ('receiver', b.pid, 'pure-listen')):
         total, link = thread_cpu(pid, None) - cpu[name][0], thread_cpu(pid, thread) - cpu[name][1]
         row[f'{name}_cpu_us_per_message'] = round(total * 1000 / n, 2)
         row[f'{name}_link_cpu_us_per_message'] = round(link * 1000 / n, 2)
diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c
index 3a531ef..4560850 100644
--- a/research/experiments/019-idle-links/transport.c
+++ b/research/experiments/019-idle-links/transport.c
@@ -191,6 +191,7 @@ int main(int argc, char **argv) {
     if (!window) /* a backlog: all queued before the link starts */
         for (; first <= count; ++first)
             if (!pure_link_queue(link, first, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1;
+    if (!window) pure_link_push(link);
     began = now_ns();
     uint64_t cpu_link = 0, cpu_listen = thread_cpu("pure-listen"), cpu_all = thread_cpu(NULL);
     uint64_t end = limit > 0 ? began + (uint64_t)(limit * 1e9) : UINT64_MAX;
@@ -204,13 +205,14 @@ int main(int argc, char **argv) {
         if (now_ns() >= end) break;
         if (queued_at) queued_at[seq] = now_ns();
         if (!pure_link_queue(link, seq, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1;
+        pure_link_push(link);
     }
     while (pure_link_delivered(link) < count && now_ns() < end) {
         struct timespec pause = {0, 50000};
         nanosleep(&pause, NULL);
     }
     double seconds = (double)((end < now_ns() ? end : now_ns()) - began) / 1e9;
-    uint64_t link_ns = thread_cpu("pure-link") - cpu_link, listen_ns = thread_cpu("pure-listen") - cpu_listen,
+    uint64_t link_ns = thread_cpu("pure-send") - cpu_link, listen_ns = thread_cpu("pure-listen") - cpu_listen,
              all_ns = thread_cpu(NULL) - cpu_all;
     PureLinkCounts sent, got;
     pure_link_counts(link, &sent);