From cbfec0d3ac239e050597bdb080645ffb53bad66e Mon Sep 17 00:00:00 2001 From: eunhhu Date: Mon, 19 Oct 2026 14:08:53 +0100 Subject: 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) --- research/experiments/046-commit-replica/follower.c | 211 +++++++++++++++++++++ 1 file changed, 211 insertions(+) create mode 100644 research/experiments/046-commit-replica/follower.c (limited to 'research/experiments/046-commit-replica/follower.c') 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 +#include +#include +#include +#include +#include +#include +#include +#include +#include + +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); +} -- cgit v1.2.3