Wake a link's thread only when it sleeps, once

Every queued record wrote the link's eventfd, a system call each, and woke
the thread to send a packet of a few records. A record queued now wakes
the thread only when it sleeps and nobody woke it yet; one queued while it
sends marks the link, and the thread looks again before it sleeps. With
records queued one at a time as fast as a thread can, 200,000 small ones
went at 400k a second, and now at 730k; queuing one costs 0.43 us, not
1.0.

research/experiments/019-idle-links/transport.c measures the transport
alone: a link and a listener in one process, no app, no store. A backlog
of 200,000 records of 61 bytes drains at 3 to 4 million a second, 0.24 us
a record on each side.

2 files changed, 178 insertions, 3 deletions

diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c
new file mode 100644
index 0000000..ad8e5f3
--- /dev/null
+++ b/research/experiments/019-idle-links/transport.c
@@ -0,0 +1,165 @@
+#define _GNU_SOURCE
+/* A link and a listener in one process, with no app and no store: what the
+ * transport alone costs a record (M5 of docs/design/011-pure-transport.md).
+ * The listener says each delivery stored as it hands it over.
+ *
+ *   transport SENDER.pmod RECEIVER.pmod COUNT SIZE [WINDOW]
+ *       COUNT records of SIZE item bytes (an order of the sender's output
+ *       "ordered", to the receiver's input "order"), at most WINDOW waiting
+ *       at once (100,000); a line of JSON: how long, records a second, and
+ *       CPU a record by thread (the link's, the listener's, the queuing
+ *       one's). WINDOW 0: all are queued before the link starts (a backlog
+ *       drained). Built beside pure-crypto.so (build/). */
+#include "link.h"
+#include "module.h"
+#include "seal.h"
+#include <arpa/inet.h>
+#include <dirent.h>
+#include <netinet/in.h>
+#include <stdatomic.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <time.h>
+#include <unistd.h>
+
+static PureModule sender, receiver;
+static PureLinkListener *listener;
+static atomic_uint_fast64_t delivered;
+
+static bool deliver(void *context, const PureLinkDelivery *d) {
+    (void)context;
+    atomic_fetch_add(&delivered, d->count);
+    pure_link_done(listener, d->token, d->first + d->count - 1, true, false);
+    return true;
+}
+
+static void notify(void *context) { (void)context; }
+
+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;
+}
+
+/* CPU ns of this process's threads named name (NULL: all). */
+static uint64_t thread_cpu(const char *name) {
+    uint64_t total = 0;
+    DIR *tasks = opendir("/proc/self/task");
+    for (struct dirent *e; tasks && (e = readdir(tasks));) {
+        if (e->d_name[0] == '.') continue;
+        char path[320], comm[64] = "";
+        snprintf(path, sizeof(path), "/proc/self/task/%s/comm", e->d_name);
+        FILE *f = fopen(path, "r");
+        if (f) {
+            if (!fgets(comm, sizeof(comm), f)) comm[0] = 0;
+            fclose(f);
+        }
+        comm[strcspn(comm, "\n")] = 0;
+        if (name && strcmp(comm, name)) continue;
+        snprintf(path, sizeof(path), "/proc/self/task/%s/schedstat", e->d_name);
+        unsigned long long ns = 0;
+        if ((f = fopen(path, "r"))) {
+            if (fscanf(f, "%llu", &ns) != 1) ns = 0;
+            fclose(f);
+        }
+        total += ns;
+    }
+    if (tasks) closedir(tasks);
+    return total;
+}
+
+static void put32(unsigned char *p, uint32_t v) { for (int i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> 8 * i); }
+
+/* The stored form of an order of input: u32 fields 1, texts of size bytes. */
+static size_t order(const PureModule *m, uint32_t input, size_t size, unsigned char *out) {
+    const PureInputShape *s = &m->input[input];
+    memset(out, 0, s->prefix);
+    size_t at = s->prefix;
+    for (uint32_t f = 0; f < s->field_count; ++f) {
+        const PureField *field = &m->field[s->field_start + f];
+        if (field->type == PURE_FIELD_U32) put32(out + field->offset, 1);
+        else if (field->type == PURE_FIELD_TEXT || field->type == PURE_FIELD_BYTES) {
+            size_t length = f ? 5 : size; /* the first text is the item; the rest ("Seoul") short */
+            if (field->offset != PURE_FIELD_REST) put32(out + field->offset, (uint32_t)length);
+            memset(out + at, 'x', length);
+            at += length;
+        }
+    }
+    return at;
+}
+
+int main(int argc, char **argv) {
+    if (argc < 5) {
+        fputs("transport SENDER.pmod RECEIVER.pmod COUNT SIZE [WINDOW]\n", stderr);
+        return 2;
+    }
+    char error[256], dir[] = "/tmp/pure-transport-XXXXXX", path[512], hex[2 * PURE_SEAL_KEY + 1];
+    uint64_t count = strtoull(argv[3], NULL, 10), window = argc > 5 ? strtoull(argv[5], NULL, 10) : 100000;
+    size_t size = strtoull(argv[4], NULL, 10);
+    if (pure_module_read(argv[1], &sender, error, sizeof(error)) != PURE_OK ||
+        pure_module_read(argv[2], &receiver, error, sizeof(error)) != PURE_OK || !mkdtemp(dir)) {
+        fprintf(stderr, "transport: %s\n", error);
+        return 1;
+    }
+    unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY];
+    snprintf(path, sizeof(path), "%s/a.key", dir);
+    PureSealKey *a = pure_seal_key(path, hex, error, sizeof(error));
+    if (!a || !pure_seal_peer(hex, public_a)) return fprintf(stderr, "transport: %s\n", error), 1;
+    snprintf(path, sizeof(path), "%s/b.key", dir);
+    PureSealKey *b = pure_seal_key(path, hex, error, sizeof(error));
+    if (!b || !pure_seal_peer(hex, public_b)) return fprintf(stderr, "transport: %s\n", error), 1;
+    /* A port nobody uses. */
+    int probe = socket(AF_INET, SOCK_DGRAM, 0);
+    struct sockaddr_in any = {.sin_family = AF_INET, .sin_addr.s_addr = htonl(INADDR_LOOPBACK)};
+    socklen_t length = sizeof(any);
+    if (probe < 0 || bind(probe, (struct sockaddr *)&any, sizeof(any)) || getsockname(probe, (struct sockaddr *)&any, &length))
+        return 1;
+    unsigned port = ntohs(any.sin_port);
+    close(probe);
+    if (!(listener = pure_link_listen(&receiver, port, b, public_b, deliver, NULL, error, sizeof(error))) ||
+        !pure_link_allow(listener, "bench.ordered", public_a, error, sizeof(error)) ||
+        !pure_link_serve(listener, error, sizeof(error)))
+        return fprintf(stderr, "transport: %s\n", error), 1;
+    char url[128];
+    snprintf(url, sizeof(url), "pure://127.0.0.1:%u/order", port);
+    PureLink *link = pure_link_open(&sender, "ordered", url, "bench.ordered", a, public_a, error, sizeof(error));
+    if (!link) return fprintf(stderr, "transport: %s\n", error), 1;
+    pure_link_peer(link, public_b);
+    int output = pure_module_input(&sender, "ordered");
+    unsigned char *payload = malloc(size + 256);
+    size_t payload_size = order(&sender, (uint32_t)output, size, payload);
+    uint64_t first = 1;
+    if (!window) /* a backlog: all queued before the link starts */
+        for (; first <= count; ++first)
+            if (!pure_link_queue(link, first, (PureSpan){payload, payload_size})) return fputs("transport: cannot queue\n", stderr), 1;
+    uint64_t began = now_ns(), cpu_link = 0, cpu_listen = thread_cpu("pure-listen"), cpu_all = thread_cpu(NULL);
+    if (!pure_link_start(link, notify, NULL, error, sizeof(error))) return fprintf(stderr, "transport: %s\n", error), 1;
+    for (uint64_t seq = first; seq <= count; ++seq) {
+        while (seq - pure_link_delivered(link) > window) {
+            struct timespec pause = {0, 50000};
+            nanosleep(&pause, NULL);
+        }
+        if (!pure_link_queue(link, seq, (PureSpan){payload, payload_size})) return fputs("transport: cannot queue\n", stderr), 1;
+    }
+    while (pure_link_delivered(link) < count) {
+        struct timespec pause = {0, 50000};
+        nanosleep(&pause, NULL);
+    }
+    double seconds = (double)(now_ns() - began) / 1e9;
+    uint64_t link_ns = thread_cpu("pure-link") - cpu_link, listen_ns = thread_cpu("pure-listen") - cpu_listen,
+             all_ns = thread_cpu(NULL) - cpu_all;
+    PureLinkCounts sent;
+    pure_link_counts(link, &sent);
+    printf("{\"records\":%llu,\"record_bytes\":%zu,\"seconds\":%.3f,\"per_second\":%.0f,\"packets\":%llu,"
+           "\"resent\":%llu,\"link_us_per_record\":%.3f,\"listener_us_per_record\":%.3f,\"other_us_per_record\":%.3f}\n",
+           (unsigned long long)count, payload_size, seconds, (double)count / seconds, (unsigned long long)sent.packets,
+           (unsigned long long)sent.resent, (double)link_ns / 1e3 / (double)count, (double)listen_ns / 1e3 / (double)count,
+           (double)(all_ns - link_ns - listen_ns) / 1e3 / (double)count);
+    pure_link_close(link);
+    PureLinkCounts heard;
+    pure_link_listener_close(listener, &heard);
+    snprintf(path, sizeof(path), "rm -rf %s", dir);
+    return system(path) ? 0 : 0;
+}
diff --git a/serve/link.c b/serve/link.c
index 016f306..1dde29d 100644
--- a/serve/link.c
+++ b/serve/link.c
@@ -195,6 +195,9 @@ struct PureLink {
     size_t head, count, capacity;
     uint64_t queued, stored, told; /* the last queued; saved by the receiver; told the owner */
     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;
     int fd, wake;
     pthread_t thread;
     void (*notify)(void *);
@@ -316,9 +319,13 @@ bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) {
         } else
             ok = false;
     }
+    bool signal = false;
     if (ok) {
         *record_at(l, l->count++) = r;
         l->queued = seq;
+        l->queued_new = true;
+        signal = l->sleeping && !l->signaled;
+        l->signaled = l->signaled || signal;
     }
     pthread_mutex_unlock(&l->lock);
     if (!ok) {
@@ -326,7 +333,7 @@ bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) {
         free(r.piece);
         return false;
     }
-    wake(l->wake);
+    if (signal) wake(l->wake);
     return true;
 }
 
@@ -648,6 +655,7 @@ static bool transmit(PureLink *l, Flight *f, const unsigned char *body, size_t s
 
 static void send_packets(PureLink *l, uint64_t now) {
     unsigned char body[BODY];
+    l->queued_new = false; /* all queued is looked at now */
     for (unsigned n = 0; n < 128; ++n) {
         bool probe = l->probes > 0;
         if (!probe && l->in_flight + DATAGRAM > l->window) break;
@@ -721,10 +729,11 @@ static void *run_link(void *context) {
     for (;;) {
         pthread_mutex_lock(&l->lock);
         uint64_t now = now_ns(), due = next_timer(l);
-        bool stopping = l->stopping;
+        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 = due ? (int)((due > now ? due - now : 0) / MS) + 1 : -1;
+        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) {
@@ -732,6 +741,7 @@ static void *run_link(void *context) {
             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);
         run_timers(l, now);