diff options
Diffstat (limited to 'link/sending.c')
| -rw-r--r-- | link/sending.c | 160 |
1 files changed, 160 insertions, 0 deletions
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); + } +} |