diff options
| -rw-r--r-- | Makefile | 2 | ||||
| -rw-r--r-- | link/link.c | 140 | ||||
| -rw-r--r-- | link/sending.c | 160 | ||||
| -rw-r--r-- | link/sending.h | 46 |
4 files changed, 227 insertions, 121 deletions
@@ -25,7 +25,7 @@ BUILD ?= build INCLUDES = -Icore MODEL = core/module.c core/module-list.c core/module-save.c core/module-text.c core/module-store.c core/flow.c STORE = store/store-tree.c store/store-log.c -LINK = link/link-protocol.c link/link.c link/crypto.c +LINK = link/link-protocol.c link/link.c link/sending.c link/crypto.c SERVE = serve/serve.c serve/deliver.c serve/message-json.c HEADERS = $(wildcard core/*.h) $(wildcard link/*.h) $(wildcard serve/*.h) $(wildcard store/*.h) OBJECTS = $(patsubst %.c,$(BUILD)/obj/%.o,$(MODEL) $(STORE)) 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); 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 <pthread.h> +#include <stdio.h> +#include <string.h> +#include <sys/epoll.h> +#include <sys/eventfd.h> +#include <time.h> +#include <unistd.h> + +#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 <stdatomic.h> +#include <stdbool.h> +#include <stddef.h> +#include <stdint.h> + +/* 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 |