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, 30 insertions, 20 deletions

diff --git a/link/link.c b/link/link.c
index 3ec138b..8b57ba6 100644
--- a/link/link.c
+++ b/link/link.c
@@ -118,15 +118,16 @@ struct PureLink {
     bool peered, keyed; /* given the receiver's key; the keys made from it */
     pthread_mutex_t lock;
     bool started;
-    bool queued_new; /* records were queued since it last sent */
-    uint64_t told;   /* the last stored the owner was told of */
+    uint64_t told; /* the last stored the owner was told of */
     int fd;
     void (*notify)(void *);
     void *context;
     /* The sending loop's own (below). */
     struct PureLink *next; /* in the loop */
     uint64_t id;           /* its epoll events' */
+    _Atomic bool pushed;   /* records were queued and pushed since its last round */
     bool readable;         /* its socket said so this round */
+    uint64_t due;          /* when it wants its next round (0: when told) */
     unsigned char *burst;  /* packets sealed and not yet sent (BURST datagrams) */
     size_t burst_size;
     unsigned burst_count;
@@ -187,9 +188,11 @@ static void wake(int fd) {
 /* The sending loop: every link of the process sends on one thread
  * (docs/design/017-apps-on-one-host.md, 2b), an epoll over their sockets
  * and one wake, sleeping until the soonest link's next time. Made with the
- * first link started, stopped with the last closed. A queue wakes it only
- * when it sleeps, once: queued tells it records came (it then does not
- * sleep), sleeping that it is about to, signaled that it was woken. */
+ * first link started, stopped with the last closed. A round goes to the
+ * links that want one: answers on its socket, records pushed, or its time
+ * come. A push wakes it only when it sleeps, once: queued tells it records
+ * came (it then does not sleep), sleeping that it is about to, signaled
+ * that it was woken. */
 static struct {
     pthread_mutex_t lock; /* the links, and its making and stopping */
     PureLink *first;
@@ -204,15 +207,16 @@ static struct {
 bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) {
     pthread_mutex_lock(&l->lock);
     bool ok = pure_link_sender_queue(l->sender, seq, bytes, size);
-    if (ok) l->queued_new = true;
     pthread_mutex_unlock(&l->lock);
-    if (ok) {
-        atomic_store(&loop.queued, true);
-        if (atomic_load(&loop.sleeping) && !atomic_exchange(&loop.signaled, true)) wake(loop.wake);
-    }
     return ok;
 }
 
+void pure_link_push(PureLink *l) {
+    atomic_store(&l->pushed, true);
+    atomic_store(&loop.queued, true);
+    if (atomic_load(&loop.sleeping) && !atomic_exchange(&loop.signaled, true)) wake(loop.wake);
+}
+
 uint64_t pure_link_queued(PureLink *l) {
     pthread_mutex_lock(&l->lock);
     uint64_t queued = pure_link_sender_queued(l->sender);
@@ -297,7 +301,6 @@ static void flush(PureLink *l) {
 /* Packets while the sender has them (up to 128 a turn), into bursts: a
  * short packet ends one, and a probe of a larger datagram goes alone. */
 static void send_packets(PureLink *l, uint64_t now) {
-    l->queued_new = false; /* all queued is looked at now */
     l->segment = pure_link_sender_datagram(l->sender); /* larger once the path is found to carry it */
     unsigned most = BURST_BYTES / l->segment < BURST ? (unsigned)(BURST_BYTES / l->segment) : BURST;
     for (unsigned n = 0; n < 128; ++n) {
@@ -365,7 +368,7 @@ enum { LOOK_PROBES = 3 };
 /* A link's round in the loop (its lock not held): answers taken when its
  * socket had them, its times kept, the receiver looked for again when
  * probes go unanswered, packets sent, and the owner told of what was
- * stored. When it wants its next round (ns on now's clock; 0: none). */
+ * stored. When it wants its next round (ns on now's clock; 0: when told). */
 static uint64_t link_round(PureLink *l) {
     pthread_mutex_lock(&l->lock);
     uint64_t now = now_ns();
@@ -386,7 +389,7 @@ static uint64_t link_round(PureLink *l) {
     }
     l->looked = probes;
     send_packets(l, now);
-    uint64_t stored = pure_link_sender_stored(l->sender), due = l->queued_new ? now : pure_link_sender_due(l->sender);
+    uint64_t stored = pure_link_sender_stored(l->sender), due = pure_link_sender_due(l->sender);
     bool told = stored != l->told;
     l->told = stored;
     unsigned refusal = pure_link_sender_refusal(l->sender), before = l->refusal;
@@ -430,8 +433,9 @@ static void *run_sending(void *context) {
         }
         uint64_t soonest = 0, now = now_ns();
         for (PureLink *l = loop.first; l; l = l->next) {
-            uint64_t due = link_round(l);
-            if (due && (!soonest || due < soonest)) soonest = due;
+            bool pushed = atomic_exchange(&l->pushed, false); /* pushed after: the next round's */
+            if (l->readable || pushed || (l->due && l->due <= now)) l->due = link_round(l);
+            if (l->due && (!soonest || l->due < soonest)) soonest = l->due;
         }
         pthread_mutex_unlock(&loop.lock);
         timeout = !soonest ? -1 : soonest <= now ? 0 : (int)((soonest - now) / MS) + 1;
@@ -471,7 +475,8 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e
         l->next = loop.first;
         loop.first = l;
         l->started = true;
-        atomic_store(&loop.queued, true); /* a first round for it */
+        atomic_store(&l->pushed, true); /* a first round for it */
+        atomic_store(&loop.queued, true);
     }
     pthread_mutex_unlock(&loop.lock);
     if (!ok) {
diff --git a/link/link.h b/link/link.h
index c8d8635..9846af4 100644
--- a/link/link.h
+++ b/link/link.h
@@ -3,9 +3,9 @@
 #include "link-protocol.h"
 
 /* Links over UDP: the two sides of link/link-protocol.h (what goes on the
- * wire, and what each side keeps) with sockets and threads. A sender sends
- * on a thread of its own, packets in bursts with UDP GSO; a listener takes
- * a receiver's packets on a port, with UDP GRO. */
+ * wire, and what each side keeps) with sockets and threads. The process's
+ * senders send on one thread, packets in bursts with UDP GSO; a listener
+ * takes a receiver's packets on a port, with UDP GRO. */
 
 /* The sending side: one output's saved sends to url, pure://host:port/INPUT,
  * as source (a Pure receiver's name for it), their messages of shape (the
@@ -22,7 +22,10 @@ void pure_link_peer(PureLink *, const unsigned char peer[PURE_SEAL_KEY]);
  * pure_module_fields): a receiver whose input takes another shape may take
  * them by name (docs/design/014-fields-by-name.md). */
 bool pure_link_fields(PureLink *, const unsigned char *fields, size_t size);
+/* Sends are queued, then pushed: what was queued goes out once pushed, so
+ * the owner's sends of a turn go together, in full packets. */
 bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_t size);
+void pure_link_push(PureLink *);
 uint64_t pure_link_queued(PureLink *);
 /* The last number the receiver saved; any thread. */
 uint64_t pure_link_delivered(PureLink *);
@@ -32,7 +35,9 @@ size_t pure_link_held(PureLink *);
  * off, each in a call of its own: for a path whose queues take a burst as
  * one packet (netem does). Before pure_link_start. */
 void pure_link_gso(PureLink *, bool on);
-/* Send on a thread of the link's own, calling notify when more was saved. */
+/* Send on the process's sending loop (its thread made with the first link
+ * started, stopped with the last closed), calling notify (on that thread)
+ * when more was saved. */
 bool pure_link_start(PureLink *, void (*notify)(void *), void *context, char *error, size_t size);
 void pure_link_counts(PureLink *, PureLinkCounts *);
 void pure_link_close(PureLink *);