Take the sending loop out of the links, for the wires to join it (017, 2b)

link/sending.c is the process's sending loop: a sender joins it with its
round and the socket it waits on, and is rounded when its socket is
ready, when pushed, and when its time comes. Links are its first
senders; nothing they do changed. Events of a sender that left while the
loop waited are let go (the epoll is level-triggered: the others' come
again).

1 files changed, 20 insertions, 120 deletions

diff --git a/link/link.c b/link/link.c
index 8b57ba6..a9270fe 100644
--- a/link/link.c
+++ b/link/link.c
@@ -2,6 +2,7 @@
 /* Links over UDP (link/link.h): link/link-protocol.h's two sides with
  * sockets and threads. */
 #include "link.h"
+#include "sending.h"
 #include <errno.h>
 #include <netdb.h>
 #include <netinet/in.h>
@@ -11,7 +12,6 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <stdatomic.h>
 #include <sys/epoll.h>
 #include <sys/eventfd.h>
 #include <sys/socket.h>
@@ -113,6 +113,7 @@ static int take_datagrams(int fd, Taking *t) {
 /* ---- The sending side */
 
 struct PureLink {
+    PureSending sending;    /* its place in the sending loop (link/sending.h) */
     PureLinkSender *sender; /* under the lock */
     char host[256], port[8], input[PURE_LINK_NAME + 1], source[PURE_LINK_NAME + 1];
     bool peered, keyed; /* given the receiver's key; the keys made from it */
@@ -122,13 +123,8 @@ struct PureLink {
     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) */
+    /* Its rounds' own, on the sending loop. */
+    unsigned char *burst; /* packets sealed and not yet sent (BURST datagrams) */
     size_t burst_size;
     unsigned burst_count;
     uint64_t burst_pn[BURST];
@@ -139,6 +135,8 @@ struct PureLink {
     unsigned refusal; /* why the receiver refused the stream when it was last said (0: it takes it) */
 };
 
+static uint64_t link_round(PureSending *, uint32_t ready);
+
 PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own,
                          const unsigned char own_public[PURE_SEAL_KEY], char *error, size_t size) {
     if (!own) {
@@ -164,6 +162,7 @@ PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, co
     strcpy(l->input, slash + 1);
     memcpy(l->source, source, length);
     l->fd = -1;
+    pure_sending_init(&l->sending, link_round);
     if (!(l->sender = pure_link_sender(source, l->input, shape, own, own_public))) {
         pthread_mutex_destroy(&l->lock);
         free(l);
@@ -185,24 +184,8 @@ static void wake(int fd) {
     if (fd >= 0 && write(fd, &one, sizeof(one)) != sizeof(one)) { /* full: it is readable already */ }
 }
 
-/* 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 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;
-    int epoll, wake;
-    pthread_t thread;
-    bool running, stopping;
-    uint64_t ids;
-    Taking *taking; /* answers, received at once, any link's */
-    _Atomic bool queued, sleeping, signaled;
-} loop = {.lock = PTHREAD_MUTEX_INITIALIZER, .epoll = -1, .wake = -1};
+/* Answers received at once, any link's: the sending loop's. */
+static Taking answers;
 
 bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) {
     pthread_mutex_lock(&l->lock);
@@ -211,11 +194,7 @@ bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size
     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);
-}
+void pure_link_push(PureLink *l) { pure_sending_push(&l->sending); }
 
 uint64_t pure_link_queued(PureLink *l) {
     pthread_mutex_lock(&l->lock);
@@ -247,9 +226,9 @@ size_t pure_link_held(PureLink *l) {
 
 static void take_answers(PureLink *l, uint64_t now) {
     for (unsigned rounds = 0, n = TAKE; rounds < 4 && n == TAKE; ++rounds) {
-        n = (unsigned)take_datagrams(l->fd, loop.taking);
+        n = (unsigned)take_datagrams(l->fd, &answers);
         for (unsigned k = 0; k < n; ++k)
-            pure_link_sender_take(l->sender, loop.taking->bytes[k], loop.taking->messages[k].msg_len, now);
+            pure_link_sender_take(l->sender, answers.bytes[k], answers.messages[k].msg_len, now);
     }
 }
 
@@ -355,6 +334,7 @@ static bool look(PureLink *l, char *error, size_t size) {
     int segment = DATAGRAM, none = 0;
     l->gso = !l->no_gso && !setsockopt(fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment)) &&
              !setsockopt(fd, SOL_UDP, UDP_SEGMENT, &none, sizeof(none));
+    pure_sending_watch(&l->sending, fd, EPOLLIN); /* answers come to it */
     if (l->fd >= 0) close(l->fd);
     l->fd = fd;
     return true;
@@ -369,22 +349,17 @@ enum { LOOK_PROBES = 3 };
  * 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: when told). */
-static uint64_t link_round(PureLink *l) {
+static uint64_t link_round(PureSending *s, uint32_t ready) {
+    PureLink *l = (PureLink *)s;
     pthread_mutex_lock(&l->lock);
     uint64_t now = now_ns();
-    if (l->readable) take_answers(l, now);
-    l->readable = false;
+    if (ready) take_answers(l, now);
     pure_link_sender_tick(l->sender, now);
     unsigned probes = pure_link_sender_probes(l->sender);
     if ((probes >= LOOK_PROBES && probes != l->looked) || l->nowhere) { /* before the next goes, so its answer comes to the new socket */
         l->nowhere = false;
         pthread_mutex_unlock(&l->lock); /* a name looked up may take its time: the owner queues meanwhile */
-        int before = l->fd;
         look(l, NULL, 0);
-        if (l->fd != before && l->fd >= 0) { /* its new socket in the epoll (the old one left it closed) */
-            struct epoll_event event = {.events = EPOLLIN, .data.u64 = l->id};
-            epoll_ctl(loop.epoll, EPOLL_CTL_ADD, l->fd, &event);
-        }
         pthread_mutex_lock(&l->lock);
     }
     l->looked = probes;
@@ -406,43 +381,6 @@ static uint64_t link_round(PureLink *l) {
     return due;
 }
 
-static void *run_sending(void *context) {
-    (void)context;
-    struct epoll_event events[64];
-    int timeout = 0;
-    for (;;) {
-        atomic_store(&loop.sleeping, true);
-        if (atomic_load(&loop.queued)) timeout = 0; /* queued since: no sleep */
-        int ready = epoll_wait(loop.epoll, events, 64, timeout);
-        atomic_store(&loop.sleeping, false);
-        atomic_store(&loop.signaled, false);
-        atomic_store(&loop.queued, false);
-        pthread_mutex_lock(&loop.lock);
-        if (loop.stopping) {
-            pthread_mutex_unlock(&loop.lock);
-            break;
-        }
-        for (int e = 0; e < ready; ++e) {
-            if (events[e].data.u64 == 0) { /* the wake */
-                uint64_t signals;
-                if (read(loop.wake, &signals, sizeof(signals)) < 0) { /* cleared already */ }
-                continue;
-            }
-            for (PureLink *l = loop.first; l; l = l->next) /* a link closed since is gone from the list */
-                if (l->id == events[e].data.u64) l->readable = true;
-        }
-        uint64_t soonest = 0, now = now_ns();
-        for (PureLink *l = loop.first; l; l = l->next) {
-            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;
-    }
-    return NULL;
-}
-
 bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *error, size_t size) {
     if (!l->peered) {
         snprintf(error, size, "pure://%s:%s/%s: no receiver's key (--seal)", l->host, l->port, l->input);
@@ -456,34 +394,11 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e
         snprintf(error, size, "pure://%s:%s/%s: cannot start (the keys, memory)", l->host, l->port, l->input);
         return false;
     }
-    pthread_mutex_lock(&loop.lock);
-    bool ok = loop.running;
-    if (!ok) { /* the first link: the loop */
-        loop.stopping = false;
-        struct epoll_event event = {.events = EPOLLIN, .data.u64 = 0};
-        ok = (loop.taking || (loop.taking = malloc(sizeof(*loop.taking)))) &&
-             (loop.epoll >= 0 || (loop.epoll = epoll_create1(EPOLL_CLOEXEC)) >= 0) &&
-             (loop.wake >= 0 || ((loop.wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) >= 0 &&
-                                 !epoll_ctl(loop.epoll, EPOLL_CTL_ADD, loop.wake, &event))) &&
-             !pthread_create(&loop.thread, NULL, run_sending, NULL);
-        if (ok) pthread_setname_np(loop.thread, "pure-send"); /* the sends' CPU, seen apart */
-        loop.running = ok;
-    }
-    struct epoll_event event = {.events = EPOLLIN, .data.u64 = l->id = ++loop.ids};
-    ok = ok && !epoll_ctl(loop.epoll, EPOLL_CTL_ADD, l->fd, &event);
-    if (ok) {
-        l->next = loop.first;
-        loop.first = l;
-        l->started = true;
-        atomic_store(&l->pushed, true); /* a first round for it */
-        atomic_store(&loop.queued, true);
-    }
-    pthread_mutex_unlock(&loop.lock);
-    if (!ok) {
-        snprintf(error, size, "pure://%s:%s/%s: cannot start (the sending loop)", l->host, l->port, l->input);
+    char why[128];
+    if (!(l->started = pure_sending_join(&l->sending, why, sizeof(why)))) {
+        snprintf(error, size, "pure://%s:%s/%s: cannot start (%s)", l->host, l->port, l->input, why);
         return false;
     }
-    wake(loop.wake);
     return true;
 }
 
@@ -495,22 +410,7 @@ void pure_link_counts(PureLink *l, PureLinkCounts *out) {
 
 void pure_link_close(PureLink *l) {
     if (!l) return;
-    if (l->started) { /* out of the loop: the loop holds its lock while it goes over the links */
-        pthread_mutex_lock(&loop.lock);
-        for (PureLink **at = &loop.first; *at; at = &(*at)->next)
-            if (*at == l) {
-                *at = l->next;
-                break;
-            }
-        if (l->fd >= 0) epoll_ctl(loop.epoll, EPOLL_CTL_DEL, l->fd, NULL);
-        bool last = !loop.first && loop.running;
-        if (last) loop.stopping = true, loop.running = false;
-        pthread_mutex_unlock(&loop.lock);
-        if (last) { /* the last link: the loop stops */
-            wake(loop.wake);
-            pthread_join(loop.thread, NULL);
-        }
-    }
+    pure_sending_leave(&l->sending); /* no round of it after */
     pure_link_sender_free(l->sender);
     free(l->burst);
     if (l->fd >= 0) close(l->fd);