Wake a link's thread only when it sleeps, once

Every queued record wrote the link's eventfd, a system call each, and woke
the thread to send a packet of a few records. A record queued now wakes
the thread only when it sleeps and nobody woke it yet; one queued while it
sends marks the link, and the thread looks again before it sleeps. With
records queued one at a time as fast as a thread can, 200,000 small ones
went at 400k a second, and now at 730k; queuing one costs 0.43 us, not
1.0.

research/experiments/019-idle-links/transport.c measures the transport
alone: a link and a listener in one process, no app, no store. A backlog
of 200,000 records of 61 bytes drains at 3 to 4 million a second, 0.24 us
a record on each side.

1 files changed, 13 insertions, 3 deletions

diff --git a/serve/link.c b/serve/link.c
index 016f306..1dde29d 100644
--- a/serve/link.c
+++ b/serve/link.c
@@ -195,6 +195,9 @@ struct PureLink {
     size_t head, count, capacity;
     uint64_t queued, stored, told; /* the last queued; saved by the receiver; told the owner */
     bool stopping, started;
+    /* The thread sleeps (in poll), and has been woken since; records were
+     * queued since it last sent. A queue wakes it only when it sleeps, once. */
+    bool sleeping, signaled, queued_new;
     int fd, wake;
     pthread_t thread;
     void (*notify)(void *);
@@ -316,9 +319,13 @@ bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) {
         } else
             ok = false;
     }
+    bool signal = false;
     if (ok) {
         *record_at(l, l->count++) = r;
         l->queued = seq;
+        l->queued_new = true;
+        signal = l->sleeping && !l->signaled;
+        l->signaled = l->signaled || signal;
     }
     pthread_mutex_unlock(&l->lock);
     if (!ok) {
@@ -326,7 +333,7 @@ bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) {
         free(r.piece);
         return false;
     }
-    wake(l->wake);
+    if (signal) wake(l->wake);
     return true;
 }
 
@@ -648,6 +655,7 @@ static bool transmit(PureLink *l, Flight *f, const unsigned char *body, size_t s
 
 static void send_packets(PureLink *l, uint64_t now) {
     unsigned char body[BODY];
+    l->queued_new = false; /* all queued is looked at now */
     for (unsigned n = 0; n < 128; ++n) {
         bool probe = l->probes > 0;
         if (!probe && l->in_flight + DATAGRAM > l->window) break;
@@ -721,10 +729,11 @@ static void *run_link(void *context) {
     for (;;) {
         pthread_mutex_lock(&l->lock);
         uint64_t now = now_ns(), due = next_timer(l);
-        bool stopping = l->stopping;
+        bool stopping = l->stopping, more = l->queued_new; /* queued while it sent: no sleep */
+        l->sleeping = !more;
         pthread_mutex_unlock(&l->lock);
         if (stopping) break;
-        int timeout = due ? (int)((due > now ? due - now : 0) / MS) + 1 : -1;
+        int timeout = more ? 0 : due ? (int)((due > now ? due - now : 0) / MS) + 1 : -1;
         struct pollfd waits[2] = {{l->fd, POLLIN, 0}, {l->wake, POLLIN, 0}};
         if (poll(waits, 2, timeout) < 0 && errno != EINTR) break;
         if (waits[1].revents & POLLIN) {
@@ -732,6 +741,7 @@ static void *run_link(void *context) {
             if (read(l->wake, &signals, sizeof(signals)) < 0) { /* cleared already */ }
         }
         pthread_mutex_lock(&l->lock);
+        l->sleeping = l->signaled = false;
         now = now_ns();
         if (waits[0].revents) take_answers(l, now);
         run_timers(l, now);