path: root/link/link.c

#define _GNU_SOURCE
/* Links over UDP (link/link.h): link/link-protocol.h's two sides with
 * sockets and threads. */
#include "link.h"
#include <errno.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/udp.h>
#include <poll.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/eventfd.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>

enum { DATAGRAM = PURE_LINK_DATAGRAM };
/* Packets sent at once with UDP GSO: each but the last is a whole datagram
 * (a full one padded), and 54 of them fit a UDP datagram's 65,507 bytes. */
enum { BURST = 54 };
#ifndef UDP_SEGMENT
#define UDP_SEGMENT 103
#endif
#ifndef UDP_GRO
#define UDP_GRO 104
#endif
#define MS UINT64_C(1000000)

static uint64_t now_ns(void) {
    struct timespec t;
    clock_gettime(CLOCK_MONOTONIC, &t);
    return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec;
}

/* Datagrams taken from a socket at once (recvmmsg). */
enum { TAKE = 64, SLOT = DATAGRAM + 64 };
typedef struct {
    unsigned char bytes[TAKE][SLOT];
    struct sockaddr_storage from[TAKE];
    struct iovec parts[TAKE];
    struct mmsghdr messages[TAKE];
} Taking;

/* With UDP GRO the kernel hands a listener the datagrams of one sender that
 * came one after another as one buffer, each the segment's size but the
 * last (a burst sent with GSO comes whole, not cut up): up to GATHER such
 * buffers at once. */
enum { GATHER = 8, GATHERED = 65536 };
typedef struct {
    unsigned char bytes[GATHER][GATHERED];
    struct sockaddr_storage from[GATHER];
    struct iovec parts[GATHER];
    struct mmsghdr messages[GATHER];
    _Alignas(8) unsigned char control[GATHER][64]; /* a cmsghdr's alignment */
} Gathering;

static int gather(int fd, Gathering *g) {
    for (unsigned k = 0; k < GATHER; ++k) {
        g->parts[k] = (struct iovec){g->bytes[k], GATHERED};
        g->messages[k] = (struct mmsghdr){.msg_hdr = {.msg_name = &g->from[k], .msg_namelen = sizeof(g->from[k]),
                                                      .msg_iov = &g->parts[k], .msg_iovlen = 1,
                                                      .msg_control = g->control[k],
                                                      .msg_controllen = sizeof(g->control[k])}};
    }
    int n = recvmmsg(fd, g->messages, GATHER, MSG_DONTWAIT, NULL);
    return n > 0 ? n : 0;
}

/* The size of each datagram in a buffer gathered (all of it when one). */
static size_t segment_of(struct msghdr *m, size_t length) {
    for (struct cmsghdr *c = CMSG_FIRSTHDR(m); c; c = CMSG_NXTHDR(m, c))
        if (c->cmsg_level == SOL_UDP && c->cmsg_type == UDP_GRO) {
            int size;
            memcpy(&size, CMSG_DATA(c), sizeof(size));
            if (size > 0) return (size_t)size;
        }
    return length ? length : 1;
}

/* Up to TAKE datagrams from fd: how many (0 when none waits). */
static int take_datagrams(int fd, Taking *t) {
    for (unsigned k = 0; k < TAKE; ++k) {
        t->parts[k] = (struct iovec){t->bytes[k], SLOT};
        t->messages[k] = (struct mmsghdr){.msg_hdr = {.msg_name = &t->from[k], .msg_namelen = sizeof(t->from[k]),
                                                      .msg_iov = &t->parts[k], .msg_iovlen = 1}};
    }
    int n = recvmmsg(fd, t->messages, TAKE, MSG_DONTWAIT, NULL);
    return n > 0 ? n : 0;
}

/* ---- The sending side */

struct PureLink {
    PureLinkSender *sender; /* under the lock */
    char host[256], port[8], input[PURE_LINK_NAME + 1];
    bool peered, keyed; /* given the receiver's key; the keys made from it */
    pthread_mutex_t lock;
    bool stopping, started;
    /* The thread sleeps (in poll), and has been woken since; records were
     * queued since it last sent. A queue wakes it only when it sleeps, once. */
    bool sleeping, signaled, queued_new;
    uint64_t told; /* the last stored the owner was told of */
    int fd, wake;
    pthread_t thread;
    void (*notify)(void *);
    void *context;
    /* The thread's own. */
    Taking *taking;       /* answers, received at once */
    unsigned char *burst; /* packets sealed and not yet sent (BURST datagrams) */
    size_t burst_size;
    unsigned burst_count;
    uint64_t burst_pn[BURST];
    bool gso, no_gso;
};

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) {
        snprintf(error, size, "a pure:// link needs the node's --key");
        return NULL;
    }
    size_t length = strlen(source);
    const char *host = url + 7, *slash = strchr(host, '/'), *colon = slash ? memchr(host, ':', (size_t)(slash - host)) : NULL;
    if (!length || length > PURE_LINK_NAME || strncmp(url, "pure://", 7) || !slash || !colon || colon == host ||
        (size_t)(colon - host) >= 256 || slash - colon - 1 < 1 || slash - colon - 1 > 5 || !slash[1] ||
        strlen(slash + 1) > PURE_LINK_NAME) {
        snprintf(error, size, "a link is pure://host:port/INPUT");
        return NULL;
    }
    PureLink *l = calloc(1, sizeof(*l));
    if (!l || pthread_mutex_init(&l->lock, NULL)) {
        free(l);
        snprintf(error, size, "out of memory");
        return NULL;
    }
    memcpy(l->host, host, (size_t)(colon - host));
    memcpy(l->port, colon + 1, (size_t)(slash - colon - 1));
    strcpy(l->input, slash + 1);
    l->fd = l->wake = -1;
    if (!(l->sender = pure_link_sender(source, l->input, shape, own, own_public))) {
        pthread_mutex_destroy(&l->lock);
        free(l);
        snprintf(error, size, "out of memory");
        return NULL;
    }
    return l;
}

void pure_link_gso(PureLink *l, bool on) { l->no_gso = !on; }

void pure_link_peer(PureLink *l, const unsigned char peer[PURE_SEAL_KEY]) {
    l->peered = true;
    l->keyed = pure_link_sender_peer(l->sender, peer);
}

static void wake(int fd) {
    uint64_t one = 1;
    if (fd >= 0 && write(fd, &one, sizeof(one)) != sizeof(one)) { /* full: it is readable already */ }
}

bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) {
    pthread_mutex_lock(&l->lock);
    bool ok = pure_link_sender_queue(l->sender, seq, bytes, size), signal = false;
    if (ok) {
        l->queued_new = true;
        signal = l->sleeping && !l->signaled;
        l->signaled = l->signaled || signal;
    }
    pthread_mutex_unlock(&l->lock);
    if (signal) wake(l->wake);
    return ok;
}

uint64_t pure_link_queued(PureLink *l) {
    pthread_mutex_lock(&l->lock);
    uint64_t queued = pure_link_sender_queued(l->sender);
    pthread_mutex_unlock(&l->lock);
    return queued;
}

uint64_t pure_link_delivered(PureLink *l) {
    pthread_mutex_lock(&l->lock);
    uint64_t stored = pure_link_sender_stored(l->sender);
    pthread_mutex_unlock(&l->lock);
    return stored;
}

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, l->taking);
        for (unsigned k = 0; k < n; ++k)
            pure_link_sender_take(l->sender, l->taking->bytes[k], l->taking->messages[k].msg_len, now);
    }
}

static bool transient(void) { return errno == EAGAIN || errno == EWOULDBLOCK || errno == ENOBUFS || errno == EINTR; }

/* The packets of the burst to the socket (each but the last a whole
 * datagram): with GSO in one call, else one by one. */
static void flush(PureLink *l) {
    unsigned n = l->burst_count, sent = 0;
    bool whole = n > 1 && l->gso;
    if (whole && send(l->fd, l->burst, l->burst_size, 0) >= 0) sent = n;
    else if (whole && !transient()) l->gso = false, whole = false; /* no GSO here: one by one from now on */
    for (size_t at = 0; !whole && sent < n; ++sent) {
        size_t size = sent + 1 < n ? DATAGRAM : l->burst_size - at;
        if (send(l->fd, l->burst + at, size, 0) < 0 && transient()) break;
        at += size;
    }
    for (unsigned k = sent; k < n; ++k) pure_link_sender_refused(l->sender, l->burst_pn[k]);
    l->burst_count = 0;
    l->burst_size = 0;
}

/* Packets while the sender has them (up to 128 a turn), into bursts: a
 * short packet ends one. */
static void send_packets(PureLink *l, uint64_t now) {
    l->queued_new = false; /* all queued is looked at now */
    for (unsigned n = 0; n < 128; ++n) {
        if (l->burst_count && l->burst_size != (size_t)l->burst_count * DATAGRAM) flush(l); /* the last was short */
        uint64_t pn;
        size_t size = pure_link_sender_packet(l->sender, l->burst + l->burst_size, l->gso, now, &pn);
        if (!size) break;
        l->burst_pn[l->burst_count++] = pn;
        l->burst_size += size;
        if (l->burst_count == BURST) flush(l);
    }
    flush(l);
}

static void *run_link(void *context) {
    PureLink *l = context;
    for (;;) {
        pthread_mutex_lock(&l->lock);
        uint64_t now = now_ns(), due = pure_link_sender_due(l->sender);
        bool stopping = l->stopping, more = l->queued_new; /* queued while it sent: no sleep */
        l->sleeping = !more;
        pthread_mutex_unlock(&l->lock);
        if (stopping) break;
        int timeout = more ? 0 : due ? (int)((due > now ? due - now : 0) / MS) + 1 : -1;
        struct pollfd waits[2] = {{l->fd, POLLIN, 0}, {l->wake, POLLIN, 0}};
        if (poll(waits, 2, timeout) < 0 && errno != EINTR) break;
        if (waits[1].revents & POLLIN) {
            uint64_t signals;
            if (read(l->wake, &signals, sizeof(signals)) < 0) { /* cleared already */ }
        }
        pthread_mutex_lock(&l->lock);
        l->sleeping = l->signaled = false;
        now = now_ns();
        if (waits[0].revents) take_answers(l, now);
        pure_link_sender_tick(l->sender, now);
        send_packets(l, now);
        uint64_t stored = pure_link_sender_stored(l->sender);
        bool told = stored != l->told;
        l->told = stored;
        pthread_mutex_unlock(&l->lock);
        if (told && l->notify) l->notify(l->context);
    }
    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);
        return false;
    }
    struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found = NULL;
    int resolved = getaddrinfo(l->host, l->port, &hints, &found);
    if (resolved) {
        snprintf(error, size, "%s: %s", l->host, gai_strerror(resolved));
        return false;
    }
    l->fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    bool ok = l->fd >= 0 && !connect(l->fd, found->ai_addr, found->ai_addrlen);
    freeaddrinfo(found);
    int buffer = 4 << 20;
    if (ok) {
        setsockopt(l->fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer));
        setsockopt(l->fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer));
    }
    int segment = DATAGRAM;
    l->gso = ok && !l->no_gso && !setsockopt(l->fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment));
    ok = ok && l->keyed && (l->burst = malloc((size_t)BURST * DATAGRAM)) && (l->taking = malloc(sizeof(*l->taking))) &&
         (l->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) >= 0;
    l->notify = notify;
    l->context = context;
    if (!ok || pthread_create(&l->thread, NULL, run_link, l)) {
        snprintf(error, size, "pure://%s:%s/%s: cannot start (a socket, the keys)", l->host, l->port, l->input);
        return false;
    }
    pthread_setname_np(l->thread, "pure-link"); /* a link's CPU, seen apart */
    l->started = true;
    return true;
}

void pure_link_counts(PureLink *l, PureLinkCounts *out) {
    pthread_mutex_lock(&l->lock);
    pure_link_sender_counts(l->sender, out);
    pthread_mutex_unlock(&l->lock);
}

void pure_link_close(PureLink *l) {
    if (!l) return;
    if (l->started) {
        pthread_mutex_lock(&l->lock);
        l->stopping = true;
        pthread_mutex_unlock(&l->lock);
        wake(l->wake);
        pthread_join(l->thread, NULL);
    }
    pure_link_sender_free(l->sender);
    free(l->burst);
    free(l->taking);
    if (l->fd >= 0) close(l->fd);
    if (l->wake >= 0) close(l->wake);
    pthread_mutex_destroy(&l->lock);
    free(l);
}

/* ---- The receiving side */

struct PureLinkListener {
    PureLinkReceiver *receiver; /* the thread's */
    unsigned port;
    int fd, wake;
    pthread_t thread;
    bool started;
    pthread_mutex_t lock; /* the done list and stopping */
    PureLinkDone *done;
    size_t done_count, done_capacity;
    bool stopping;
};

PureLinkListener *pure_link_listen(PureLinkReceiver *r, unsigned port, char *error, size_t size) {
    PureLinkListener *s = calloc(1, sizeof(*s));
    if (!s || pthread_mutex_init(&s->lock, NULL)) {
        free(s);
        snprintf(error, size, "out of memory");
        return NULL;
    }
    s->receiver = r;
    s->port = port;
    s->fd = s->wake = -1;
    return s;
}

static bool send_answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to,
                        size_t to_size) {
    const PureLinkListener *s = context;
    return sendto(s->fd, packet, size, 0, (const struct sockaddr *)(const void *)to, (socklen_t)to_size) >= 0;
}

static void *run_listener(void *context) {
    PureLinkListener *s = context;
    Gathering *gathering = malloc(sizeof(*gathering));
    if (!gathering) return NULL;
    for (;;) {
        uint64_t now = now_ns(), due = pure_link_receiver_due(s->receiver);
        int timeout = due ? (int)((due > now ? due - now : 0) / MS) + 1 : -1;
        struct pollfd waits[2] = {{s->fd, POLLIN, 0}, {s->wake, POLLIN, 0}};
        if (poll(waits, 2, timeout) < 0 && errno != EINTR) break;
        pthread_mutex_lock(&s->lock);
        bool stopping = s->stopping;
        pthread_mutex_unlock(&s->lock);
        if (stopping) break;
        if (waits[1].revents & POLLIN) {
            uint64_t signals;
            if (read(s->wake, &signals, sizeof(signals)) < 0) { /* cleared already */ }
        }
        now = now_ns();
        for (unsigned rounds = 0, n = GATHER; rounds < 8 && n == GATHER; ++rounds) {
            n = (unsigned)gather(s->fd, gathering);
            for (unsigned k = 0; k < n; ++k) {
                struct mmsghdr *m = &gathering->messages[k];
                size_t length = m->msg_len, step = segment_of(&m->msg_hdr, length);
                for (size_t at = 0; at < length; at += step)
                    pure_link_receiver_take(s->receiver, gathering->bytes[k] + at, step < length - at ? step : length - at,
                                            &gathering->from[k], m->msg_hdr.msg_namelen, now);
            }
        }
        pthread_mutex_lock(&s->lock);
        PureLinkDone *done = s->done;
        size_t count = s->done_count;
        s->done = NULL;
        s->done_count = s->done_capacity = 0;
        pthread_mutex_unlock(&s->lock);
        pure_link_receiver_turn(s->receiver, done, count, now, send_answer, s);
        free(done);
    }
    free(gathering);
    return NULL;
}

bool pure_link_serve(PureLinkListener *s, char *error, size_t size) {
    struct sockaddr_in6 six = {.sin6_family = AF_INET6, .sin6_port = htons((uint16_t)s->port), .sin6_addr = in6addr_any};
    int off = 0, buffer = 4 << 20;
    s->fd = socket(AF_INET6, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    bool bound = s->fd >= 0 && !setsockopt(s->fd, IPPROTO_IPV6, IPV6_V6ONLY, &off, sizeof(off)) &&
                 !bind(s->fd, (struct sockaddr *)&six, sizeof(six));
    if (!bound) { /* no IPv6 here */
        if (s->fd >= 0) close(s->fd);
        struct sockaddr_in four = {.sin_family = AF_INET, .sin_port = htons((uint16_t)s->port), .sin_addr = {INADDR_ANY}};
        s->fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
        bound = s->fd >= 0 && !bind(s->fd, (struct sockaddr *)&four, sizeof(four));
    }
    if (!bound) {
        snprintf(error, size, "--link-port %u: %s", s->port, strerror(errno));
        return false;
    }
    setsockopt(s->fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer));
    setsockopt(s->fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer));
    int one = 1; /* without GRO, a buffer is a datagram */
    setsockopt(s->fd, SOL_UDP, UDP_GRO, &one, sizeof(one));
    s->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (s->wake < 0 || pthread_create(&s->thread, NULL, run_listener, s)) {
        snprintf(error, size, "--link-port %u: cannot start", s->port);
        return false;
    }
    pthread_setname_np(s->thread, "pure-listen");
    s->started = true;
    return true;
}

void pure_link_done(PureLinkListener *s, void *token, uint64_t heard, bool stored, bool gap) {
    pthread_mutex_lock(&s->lock);
    if (s->done_count == s->done_capacity) {
        size_t capacity = s->done_capacity ? 2 * s->done_capacity : 16;
        PureLinkDone *grown = realloc(s->done, capacity * sizeof(*grown));
        if (!grown) abort(); /* a delivery's word cannot be lost */
        s->done = grown;
        s->done_capacity = capacity;
    }
    s->done[s->done_count++] = (PureLinkDone){token, heard, stored, gap};
    pthread_mutex_unlock(&s->lock);
    wake(s->wake);
}

void pure_link_listener_close(PureLinkListener *s, PureLinkCounts *counts) {
    if (!s) return;
    if (s->started) {
        pthread_mutex_lock(&s->lock);
        s->stopping = true;
        pthread_mutex_unlock(&s->lock);
        wake(s->wake);
        pthread_join(s->thread, NULL);
    }
    if (counts) pure_link_receiver_counts(s->receiver, counts);
    pure_link_receiver_free(s->receiver);
    free(s->done);
    if (s->fd >= 0) close(s->fd);
    if (s->wake >= 0) close(s->wake);
    pthread_mutex_destroy(&s->lock);
    free(s);
}