From 50f8ea3c249009fcbe649561e7aba32580bab454 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Thu, 1 Oct 2026 09:00:45 +0100 Subject: 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). --- link/link.c | 140 ++++++++------------------------------------------ link/sending.c | 160 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ link/sending.h | 46 +++++++++++++++++ 3 files changed, 226 insertions(+), 120 deletions(-) create mode 100644 link/sending.c create mode 100644 link/sending.h (limited to 'link') 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 #include #include @@ -11,7 +12,6 @@ #include #include #include -#include #include #include #include @@ -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); diff --git a/link/sending.c b/link/sending.c new file mode 100644 index 0000000..f14dfc2 --- /dev/null +++ b/link/sending.c @@ -0,0 +1,160 @@ +#define _GNU_SOURCE +#include "sending.h" +#include +#include +#include +#include +#include +#include +#include + +#define MS UINT64_C(1000000) + +uint64_t pure_sending_now(void) { + struct timespec t; + clock_gettime(CLOCK_MONOTONIC, &t); + return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec; +} + +/* The loop: its senders under the lock, which it holds over its rounds. A + * push wakes it only when it sleeps, once: pushed tells it a sender was + * (it then does not sleep), sleeping that it is about to, signaled that it + * was woken. A sender that left while it waited may be in the events it + * got: those are let go (left counts the leaves), and come again. */ +static struct { + pthread_mutex_t lock; + PureSending *first; + int epoll, wake; + pthread_t thread; + bool running, stopping; + uint64_t left; + _Atomic bool pushed, sleeping, signaled; +} loop = {.lock = PTHREAD_MUTEX_INITIALIZER, .epoll = -1, .wake = -1}; + +static void wake(void) { + uint64_t one = 1; + if (loop.wake >= 0 && write(loop.wake, &one, sizeof(one)) != sizeof(one)) { /* full: it is readable already */ } +} + +void pure_sending_init(PureSending *s, uint64_t (*round)(PureSending *, uint32_t ready)) { + memset(s, 0, sizeof(*s)); + s->round = round; + s->fd = -1; +} + +void pure_sending_push(PureSending *s) { + atomic_store(&s->pushed, true); + atomic_store(&loop.pushed, true); + if (atomic_load(&loop.sleeping) && !atomic_exchange(&loop.signaled, true)) wake(); +} + +static void watch_now(PureSending *s, int fd, uint32_t events) { + if (s->watched && (fd != s->fd || !events)) { + epoll_ctl(loop.epoll, EPOLL_CTL_DEL, s->fd, NULL); + s->watched = false; + } + if (fd >= 0 && events) { + struct epoll_event event = {.events = events, .data.ptr = s}; + s->watched = !epoll_ctl(loop.epoll, s->watched ? EPOLL_CTL_MOD : EPOLL_CTL_ADD, fd, &event); + } +} + +void pure_sending_watch(PureSending *s, int fd, uint32_t events) { + if (s->joined && (fd != s->fd || events != s->events || (fd >= 0 && events && !s->watched))) watch_now(s, fd, events); + s->fd = fd; + s->events = events; +} + +static void *run(void *context) { + (void)context; + struct epoll_event events[64]; + int timeout = 0; + pthread_mutex_lock(&loop.lock); + uint64_t left = loop.left; + pthread_mutex_unlock(&loop.lock); + for (;;) { + atomic_store(&loop.sleeping, true); + if (atomic_load(&loop.pushed)) timeout = 0; /* pushed 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.pushed, false); + pthread_mutex_lock(&loop.lock); + if (loop.stopping) { + pthread_mutex_unlock(&loop.lock); + break; + } + bool again = ready > 0 && left != loop.left; /* one may be gone: its events let go, the others' come again */ + for (int e = 0; e < ready; ++e) { + PureSending *s = events[e].data.ptr; + if (!s) { + uint64_t signals; + if (read(loop.wake, &signals, sizeof(signals)) < 0) { /* cleared already */ } + } else if (!again) s->ready |= events[e].events; + } + uint64_t soonest = 0, now = pure_sending_now(); + for (PureSending *s = loop.first; s; s = s->next) { + bool pushed = atomic_exchange(&s->pushed, false); /* pushed after: the next round's */ + if (s->ready || pushed || (s->due && s->due <= now)) { + uint32_t was = s->ready; + s->ready = 0; + s->due = s->round(s, was); + } + if (s->due && (!soonest || s->due < soonest)) soonest = s->due; + } + left = loop.left; + pthread_mutex_unlock(&loop.lock); + timeout = again ? 0 : !soonest ? -1 : soonest <= now ? 0 : (int)((soonest - now) / MS) + 1; + } + return NULL; +} + +bool pure_sending_join(PureSending *s, char *error, size_t size) { + pthread_mutex_lock(&loop.lock); + bool ok = loop.running; + if (!ok) { /* the first: the loop */ + loop.stopping = false; + struct epoll_event event = {.events = EPOLLIN, .data.ptr = NULL}; + ok = (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, NULL); + if (ok) pthread_setname_np(loop.thread, "pure-send"); /* the sends' CPU, seen apart */ + loop.running = ok; + } + if (ok) { + s->joined = true; + if (s->fd >= 0 && s->events) watch_now(s, s->fd, s->events); + s->next = loop.first; + loop.first = s; + atomic_store(&s->pushed, true); /* a first round */ + atomic_store(&loop.pushed, true); + } + pthread_mutex_unlock(&loop.lock); + if (!ok) { + snprintf(error, size, "the sending loop cannot be made"); + return false; + } + wake(); + return true; +} + +void pure_sending_leave(PureSending *s) { + if (!s->joined) return; + pthread_mutex_lock(&loop.lock); + for (PureSending **at = &loop.first; *at; at = &(*at)->next) + if (*at == s) { + *at = s->next; + break; + } + if (s->watched) epoll_ctl(loop.epoll, EPOLL_CTL_DEL, s->fd, NULL); + s->watched = s->joined = false; + ++loop.left; + bool last = !loop.first && loop.running; + if (last) loop.stopping = true, loop.running = false; + pthread_mutex_unlock(&loop.lock); + if (last) { /* the last: the loop stops */ + wake(); + pthread_join(loop.thread, NULL); + } +} diff --git a/link/sending.h b/link/sending.h new file mode 100644 index 0000000..f6b6227 --- /dev/null +++ b/link/sending.h @@ -0,0 +1,46 @@ +#ifndef PURE_SENDING_H +#define PURE_SENDING_H +#include +#include +#include +#include + +/* The process's sending loop (docs/design/017-apps-on-one-host.md, 2b): one + * thread, pure-send, sends for every link and wire, an epoll over their + * sockets. A sender joins it, and has rounds on that thread: when its socket + * is ready, when it was pushed (from any thread: there is more to send), and + * when the time it last wanted comes. The thread is made with the first + * sender joined and stopped with the last that leaves. A sender is a + * PureSending in its struct, with its round. */ +typedef struct PureSending PureSending; +struct PureSending { + /* Its round, with what its socket was ready for (EPOLLIN, EPOLLOUT, + * EPOLLERR, EPOLLHUP; 0: pushed, or its time came): when it wants the + * next (CLOCK_MONOTONIC ns; 0: when pushed or its socket is ready). */ + uint64_t (*round)(PureSending *, uint32_t ready); + /* The loop's own. */ + PureSending *next; + int fd; /* the socket it waits on, or -1 */ + uint32_t events; /* for what */ + bool joined, watched; + uint32_t ready; + uint64_t due; + _Atomic bool pushed; +}; + +/* Before it joins: no socket. */ +void pure_sending_init(PureSending *, uint64_t (*round)(PureSending *, uint32_t ready)); +/* Its first round comes at once. False with why in error when the loop + * cannot be made. */ +bool pure_sending_join(PureSending *, char *error, size_t size); +/* Out of the loop: no round of it runs once this returns (one under way is + * waited for). Not from a round. */ +void pure_sending_leave(PureSending *); +/* A round for it soon (any thread); the loop woken when it sleeps. */ +void pure_sending_push(PureSending *); +/* The socket it waits on and for what (EPOLLIN, EPOLLOUT; fd -1: none), in + * its round or before it joins; before the socket it waited on is closed. */ +void pure_sending_watch(PureSending *, int fd, uint32_t events); +/* Now on CLOCK_MONOTONIC, ns. */ +uint64_t pure_sending_now(void); +#endif -- cgit v1.2.3