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);