046: 029's principle 2 put to its breaking experiment on one Pi, the tools and the numbers: an owner (pure-run --store) with shim.so preloaded ships every change its store makes to its files (ops.h: create, write, truncate, allocate, rename, unlink, mkdir), cut at each fsync or fdatasync, over pure link (shipper.c, follower.c), and its sync returns once the follower has the cut (synchronous: one cut in flight, no log); identity: 100,000 messages (200,081 cuts; 34 compactions, an image written again) and 200,000 (400,149 cuts; 67 compactions, a second log segment), the stores compared byte for byte every 1,999 and 3,001 cuts and at the end: 0 differences, the follower's store opened as owner holding every message; takeover: 70 kill -9s (owner alone, the whole node, the node with cuts of 512 and 64 B records): all open, hold every message the owner said done, none not sent, the entries in order (and equal to the owner's own store each time); prefixes: 867 op boundaries of a 100,000-message stream (568 within cuts, 266 around installs) opened as stores: all pass; the price: 204 to 400 us a message on tmpfs (155 us waiting for acks, two cuts a message), 903 bytes shipped a message, 42% of them zeros written ahead (a compaction's new file and the tree's tail)

1 files changed, 211 insertions, 0 deletions

diff --git a/research/experiments/046-commit-replica/follower.c b/research/experiments/046-commit-replica/follower.c
new file mode 100644
index 0000000..b097c25
--- /dev/null
+++ b/research/experiments/046-commit-replica/follower.c
@@ -0,0 +1,211 @@
+/* Research 046: the follower. Hears the owner's records over pure link (a receiver on a UDP port) and does each op
+ * again under its own directory, in order (ops.h): the same bytes at the same places, the same names created, renamed
+ * and removed. A record that ends its cut is followed by an fsync of every file and directory the cut touched; then the
+ * records are said saved (pure_link_done), which is what the owner's sync waits for. Its heard (the last record done)
+ * is kept in memory: a follower started again is not this experiment's.
+ *
+ *   follower keys DIR                                   makes DIR/owner.key and DIR/follower.key, prints both public halves
+ *   follower DIR PORT KEY OWNER_PUBLIC_HEX              runs until SIGTERM, then one JSON line of counts */
+#define _GNU_SOURCE
+#include "link.h"
+#include "seal.h"
+#include "ops.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <limits.h>
+#include <pthread.h>
+#include <signal.h>
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+static int directory = -1;
+static PureLinkListener *listener;
+static uint64_t heard, cuts, applied_ops, applied_bytes;
+static bool within_cut; /* the last record done did not end its cut: the store holds part of one */
+static volatile sig_atomic_t stopping;
+
+typedef struct Job { struct Job *next; void *token; uint64_t first; unsigned count; unsigned char *records; size_t size; } Job;
+static Job *head, *tail;
+static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t waiting = PTHREAD_COND_INITIALIZER;
+
+static void fail(const char *what, const char *path) {
+    fprintf(stderr, "follower: %s %s: %s\n", what, path, strerror(errno));
+    exit(1);
+}
+
+/* The paths a cut touched, to sync at its end. */
+static char *touched[4096];
+static unsigned touched_count;
+static void touch(const char *path) {
+    for (unsigned i = 0; i < touched_count; ++i) if (!strcmp(touched[i], path)) return;
+    if (touched_count < 4096) touched[touched_count++] = strdup(path);
+}
+static void sync_touched(void) {
+    for (unsigned i = 0; i < touched_count; ++i) {
+        int fd = openat(directory, touched[i][0] ? touched[i] : ".", O_RDONLY | O_CLOEXEC);
+        if (fd >= 0) fsync(fd), close(fd); /* one renamed or removed since is synced as its directory is */
+        free(touched[i]);
+    }
+    touched_count = 0;
+    fsync(directory);
+}
+
+static const char *name_of(const unsigned char *p, char *out) {
+    uint16_t n = get16(p + 1);
+    memcpy(out, p + 3, n);
+    out[n] = 0;
+    return out[0] ? out : ".";
+}
+
+static void apply(const unsigned char *p, size_t size) {
+    char path[PATH_MAX], other[PATH_MAX];
+    const char *at = name_of(p, path);
+    const unsigned char *rest = p + 3 + get16(p + 1);
+    int fd;
+    switch (p[0]) {
+    case OP_CREATE: {
+        uint32_t flags = get32(rest), mode = get32(rest + 4);
+        if ((fd = openat(directory, at, O_WRONLY | O_CLOEXEC | (int)flags, (mode_t)mode)) < 0) fail("create", at);
+        close(fd);
+        break;
+    }
+    case OP_WRITE: {
+        uint64_t offset = get64(rest);
+        uint32_t length = get32(rest + 8);
+        if ((fd = openat(directory, at, O_WRONLY | O_CLOEXEC)) < 0) fail("open", at);
+        for (const unsigned char *b = rest + 12; length;) {
+            ssize_t n = pwrite(fd, b, length, (off_t)offset);
+            if (n <= 0) fail("write", at);
+            b += n, offset += (uint64_t)n, length -= (uint32_t)n;
+        }
+        close(fd);
+        break;
+    }
+    case OP_TRUNCATE:
+        if ((fd = openat(directory, at, O_WRONLY | O_CLOEXEC)) < 0 || ftruncate(fd, (off_t)get64(rest))) fail("truncate", at);
+        close(fd);
+        break;
+    case OP_ALLOCATE:
+        if ((fd = openat(directory, at, O_WRONLY | O_CLOEXEC)) < 0 || posix_fallocate(fd, (off_t)get64(rest), (off_t)get64(rest + 8)))
+            fail("allocate", at);
+        close(fd);
+        break;
+    case OP_RENAME: {
+        uint16_t n = get16(rest);
+        memcpy(other, rest + 2, n);
+        other[n] = 0;
+        if (renameat(directory, at, directory, other)) fail("rename", at);
+        touch(other);
+        break;
+    }
+    case OP_UNLINK:
+        if (unlinkat(directory, at, (int)get32(rest))) fail("unlink", at);
+        break;
+    case OP_MKDIR:
+        if (path[0] && mkdirat(directory, at, (mode_t)get32(rest)) && errno != EEXIST) fail("mkdir", at);
+        break;
+    case OP_SYNC: break;
+    }
+    if (p[0] != OP_UNLINK && p[0] != OP_RENAME) touch(path);
+    ++applied_ops, applied_bytes += size;
+}
+
+static void *work(void *unused) {
+    (void)unused;
+    for (;;) {
+        pthread_mutex_lock(&lock);
+        while (!head) pthread_cond_wait(&waiting, &lock);
+        Job *job = head;
+        if (!(head = job->next)) tail = NULL;
+        pthread_mutex_unlock(&lock);
+        const unsigned char *p = job->records;
+        for (unsigned i = 0; i < job->count; ++i) {
+            uint32_t size = get32(p);
+            const unsigned char *r = p + 4;
+            uint64_t seq = job->first + i;
+            p += 4 + size;
+            if (seq <= heard) continue;   /* heard already */
+            if (seq != heard + 1) break;  /* a gap: the link gives it again */
+            for (size_t at = 9; at < size;) {
+                size_t n = op_size(r + at, size - at);
+                if (!n) { fprintf(stderr, "follower: a torn op in record %llu\n", (unsigned long long)seq); exit(1); }
+                apply(r + at, n);
+                at += n;
+            }
+            if (r[0]) sync_touched(), ++cuts; /* the cut's end: what it touched is durable before it is said saved */
+            within_cut = !r[0];
+            heard = seq;
+        }
+        pure_link_done(listener, (PureLinkDone){job->token, heard, true, false, false});
+        free(job->records);
+        free(job);
+    }
+    return NULL;
+}
+
+static int input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) {
+    (void)context, (void)shape, (void)fields, (void)size;
+    return strcmp(name, "store") ? -1 : 0;
+}
+
+static bool deliver(void *context, const PureLinkDelivery *d) {
+    (void)context;
+    Job *job = malloc(sizeof(*job));
+    unsigned char *copy = malloc(d->size ? d->size : 1);
+    if (!job || !copy) return false;
+    memcpy(copy, d->records, d->size);
+    *job = (Job){NULL, d->token, d->first, d->count, copy, d->size};
+    pthread_mutex_lock(&lock);
+    if (tail) tail->next = job;
+    else head = job;
+    tail = job;
+    pthread_cond_signal(&waiting);
+    pthread_mutex_unlock(&lock);
+    return true;
+}
+
+static void on_term(int signal) { (void)signal; stopping = 1; }
+
+int main(int argc, char **argv) {
+    char error[256], hex[2 * PURE_SEAL_KEY + 1], other_hex[2 * PURE_SEAL_KEY + 1];
+    if (argc == 3 && !strcmp(argv[1], "keys")) {
+        char a[PATH_MAX], b[PATH_MAX];
+        snprintf(a, sizeof(a), "%s/owner.key", argv[2]), snprintf(b, sizeof(b), "%s/follower.key", argv[2]);
+        PureSealKey *x = pure_seal_key(a, hex, error, sizeof(error)), *y = pure_seal_key(b, other_hex, error, sizeof(error));
+        if (!x || !y) { fprintf(stderr, "follower: %s\n", error); return 1; }
+        printf("{\"owner\":\"%s\",\"follower\":\"%s\"}\n", hex, other_hex);
+        return 0;
+    }
+    if (argc != 5) { fprintf(stderr, "usage: %s DIR PORT KEY OWNER_PUBLIC_HEX | keys DIR\n", argv[0]); return 2; }
+    mkdir(argv[1], 0700);
+    if ((directory = open(argv[1], O_RDONLY | O_DIRECTORY | O_CLOEXEC)) < 0) fail("open", argv[1]);
+    unsigned char own_public[PURE_SEAL_KEY], owner[PURE_SEAL_KEY];
+    PureSealKey *key = pure_seal_key(argv[3], hex, error, sizeof(error));
+    if (!key || !pure_seal_peer(hex, own_public) || !pure_seal_peer(argv[4], owner)) { fprintf(stderr, "follower: %s\n", error); return 1; }
+    PureLinkReceiver *receiver = pure_link_receiver(key, own_public, &(PureLinkHost){input, deliver, NULL}, error, sizeof(error));
+    if (!receiver || !pure_link_allow(receiver, "owner", owner, error, sizeof(error)) ||
+        !(listener = pure_link_listen(receiver, (unsigned)atoi(argv[2]), error, sizeof(error)))) {
+        fprintf(stderr, "follower: %s\n", error);
+        return 1;
+    }
+    pthread_t worker;
+    pthread_create(&worker, NULL, work, NULL);
+    if (!pure_link_serve(listener, error, sizeof(error))) { fprintf(stderr, "follower: %s\n", error); return 1; }
+    signal(SIGTERM, on_term), signal(SIGINT, on_term);
+    printf("{\"ready\":true}\n");
+    fflush(stdout);
+    while (!stopping) pause();
+    pthread_mutex_lock(&lock); /* the worker between records: what it did is whole */
+    PureLinkCounts counts;
+    pure_link_listener_close(listener, &counts);
+    printf("{\"within_cut\":%s,\"heard\":%llu,\"cuts\":%llu,\"ops\":%llu,\"op_bytes\":%llu,\"received\":%llu,\"deliveries\":%llu,\"stored\":%llu}\n",
+           within_cut ? "true" : "false", (unsigned long long)heard, (unsigned long long)cuts, (unsigned long long)applied_ops,
+           (unsigned long long)applied_bytes, (unsigned long long)counts.received, (unsigned long long)counts.deliveries,
+           (unsigned long long)counts.stored);
+    fflush(stdout);
+    _exit(0);
+}