Measure whether sharding owners by key would add writes (084)

It would not, on this device. One file's syncs or four at once, the SD card
completes the same 510-690 commits per second, the latency growing with the
number of files. The scoreboard app with two owners, each with a store of
its own, took 23% fewer durable posts than one owner (median 16,230 against
20,974 a second) with 2-3x the p99: syncs queue, and each carries half the
events. One owner's group commit stays; cores serve reads, and writes grow
by making the sync itself cheaper.

3 files changed, 380 insertions, 0 deletions

diff --git a/research/experiments/013-parallel-syncs/2026-09-27-ext4-linux-aarch64.json b/research/experiments/013-parallel-syncs/2026-09-27-ext4-linux-aarch64.json
new file mode 100644
index 0000000..14d57e4
--- /dev/null
+++ b/research/experiments/013-parallel-syncs/2026-09-27-ext4-linux-aarch64.json
@@ -0,0 +1,192 @@
+{
+ "rounds": [
+  {
+   "probe": [
+    {
+     "threads": 1,
+     "bytes": 4096,
+     "commits": 1920,
+     "commits_per_second": 637.7,
+     "p50_ms": 1.22,
+     "p90_ms": 1.58,
+     "p99_ms": 11.27
+    },
+    {
+     "threads": 2,
+     "bytes": 4096,
+     "commits": 2026,
+     "commits_per_second": 663.6,
+     "p50_ms": 2.43,
+     "p90_ms": 3.24,
+     "p99_ms": 21.53
+    },
+    {
+     "threads": 4,
+     "bytes": 4096,
+     "commits": 1882,
+     "commits_per_second": 626.1,
+     "p50_ms": 5.09,
+     "p90_ms": 7.88,
+     "p99_ms": 59.99
+    }
+   ],
+   "app": [
+    {
+     "owners": 1,
+     "posts_per_second": 22275.0,
+     "p50_us": 2176,
+     "p99_us": 13824,
+     "errors": 0
+    },
+    {
+     "owners": 2,
+     "posts_per_second": 16833.333333333332,
+     "p50_us": 2432,
+     "p99_us": 55296,
+     "errors": 0
+    }
+   ]
+  },
+  {
+   "probe": [
+    {
+     "threads": 2,
+     "bytes": 4096,
+     "commits": 1695,
+     "commits_per_second": 564.7,
+     "p50_ms": 2.44,
+     "p90_ms": 4.6,
+     "p99_ms": 55.87
+    },
+    {
+     "threads": 4,
+     "bytes": 4096,
+     "commits": 1971,
+     "commits_per_second": 655.9,
+     "p50_ms": 5.06,
+     "p90_ms": 7.79,
+     "p99_ms": 58.39
+    },
+    {
+     "threads": 1,
+     "bytes": 4096,
+     "commits": 2006,
+     "commits_per_second": 668.6,
+     "p50_ms": 1.21,
+     "p90_ms": 1.57,
+     "p99_ms": 4.7
+    }
+   ],
+   "app": [
+    {
+     "owners": 2,
+     "posts_per_second": 17716.333333333332,
+     "p50_us": 2560,
+     "p99_us": 36864,
+     "errors": 0
+    },
+    {
+     "owners": 1,
+     "posts_per_second": 23384.333333333332,
+     "p50_us": 2176,
+     "p99_us": 14336,
+     "errors": 0
+    }
+   ]
+  },
+  {
+   "probe": [
+    {
+     "threads": 4,
+     "bytes": 4096,
+     "commits": 2056,
+     "commits_per_second": 684.3,
+     "p50_ms": 4.93,
+     "p90_ms": 7.68,
+     "p99_ms": 35.01
+    },
+    {
+     "threads": 1,
+     "bytes": 4096,
+     "commits": 1536,
+     "commits_per_second": 511.6,
+     "p50_ms": 1.23,
+     "p90_ms": 3.07,
+     "p99_ms": 12.52
+    },
+    {
+     "threads": 2,
+     "bytes": 4096,
+     "commits": 1692,
+     "commits_per_second": 556.2,
+     "p50_ms": 2.45,
+     "p90_ms": 4.57,
+     "p99_ms": 53.54
+    }
+   ],
+   "app": [
+    {
+     "owners": 1,
+     "posts_per_second": 18541.666666666668,
+     "p50_us": 2304,
+     "p99_us": 25600,
+     "errors": 0
+    },
+    {
+     "owners": 2,
+     "posts_per_second": 13991.333333333334,
+     "p50_us": 2560,
+     "p99_us": 57344,
+     "errors": 0
+    }
+   ]
+  },
+  {
+   "probe": [
+    {
+     "threads": 1,
+     "bytes": 4096,
+     "commits": 1981,
+     "commits_per_second": 660.2,
+     "p50_ms": 1.21,
+     "p90_ms": 1.57,
+     "p99_ms": 4.0
+    },
+    {
+     "threads": 2,
+     "bytes": 4096,
+     "commits": 1994,
+     "commits_per_second": 664.4,
+     "p50_ms": 2.42,
+     "p90_ms": 3.2,
+     "p99_ms": 20.6
+    },
+    {
+     "threads": 4,
+     "bytes": 4096,
+     "commits": 2011,
+     "commits_per_second": 669.8,
+     "p50_ms": 4.96,
+     "p90_ms": 7.96,
+     "p99_ms": 41.8
+    }
+   ],
+   "app": [
+    {
+     "owners": 2,
+     "posts_per_second": 15627.0,
+     "p50_us": 2432,
+     "p99_us": 47104,
+     "errors": 0
+    },
+    {
+     "owners": 1,
+     "posts_per_second": 19672.333333333332,
+     "p50_us": 2304,
+     "p99_us": 16384,
+     "errors": 0
+    }
+   ]
+  }
+ ]
+}
diff --git a/research/experiments/013-parallel-syncs/probe.c b/research/experiments/013-parallel-syncs/probe.c
new file mode 100644
index 0000000..d7f4974
--- /dev/null
+++ b/research/experiments/013-parallel-syncs/probe.c
@@ -0,0 +1,96 @@
+#define _GNU_SOURCE
+/* Whether one device syncs several files at once: THREADS threads, each with
+ * a file of its own, each commit SIZE bytes written over zeros written
+ * earlier (no new size or blocks, so no journal commit) and fdatasync, for
+ * SECONDS. Sharding apps by key gives each owner a store of its own, so this
+ * is what separate owners would get from the device.
+ * probe DIRECTORY THREADS SECONDS [SIZE] -> one JSON object. */
+#include <fcntl.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <unistd.h>
+
+enum { MOST = 20000 };
+typedef struct {
+    int fd;
+    size_t size;
+    double until, times[MOST];
+    int count;
+} Syncer;
+
+static double now_ms(void) {
+    struct timespec t;
+    clock_gettime(CLOCK_MONOTONIC, &t);
+    return t.tv_sec * 1e3 + t.tv_nsec / 1e6;
+}
+
+static int order(const void *a, const void *b) {
+    double x = *(const double *)a, y = *(const double *)b;
+    return x < y ? -1 : x > y;
+}
+
+static void *commit_all(void *argument) {
+    Syncer *s = argument;
+    char *bytes = malloc(s->size);
+    memset(bytes, 'c', s->size);
+    off_t at = 0;
+    while (s->count < MOST && now_ms() < s->until) {
+        double start = now_ms();
+        if (pwrite(s->fd, bytes, s->size, at) != (ssize_t)s->size || fdatasync(s->fd)) break;
+        at += (off_t)s->size;
+        s->times[s->count++] = now_ms() - start;
+    }
+    free(bytes);
+    return NULL;
+}
+
+int main(int argc, char **argv) {
+    if (argc < 4) return 2;
+    const char *directory = argv[1];
+    int threads = atoi(argv[2]);
+    double seconds = atof(argv[3]);
+    size_t size = argc > 4 ? (size_t)atoi(argv[4]) : 4096;
+    if (threads < 1 || threads > 16) return 2;
+    Syncer *syncers = calloc((size_t)threads, sizeof(*syncers));
+    char paths[16][512];
+    size_t zeros_size = (size_t)MOST * size;
+    char *zeros = calloc(1, zeros_size);
+    for (int i = 0; i < threads; ++i) {
+        snprintf(paths[i], sizeof(paths[i]), "%s/pure-sync-probe-%d", directory, i);
+        syncers[i].fd = open(paths[i], O_RDWR | O_CREAT | O_TRUNC, 0600);
+        if (syncers[i].fd < 0 || pwrite(syncers[i].fd, zeros, zeros_size, 0) != (ssize_t)zeros_size) return 1;
+        fsync(syncers[i].fd);
+        syncers[i].size = size;
+    }
+    free(zeros);
+    pthread_t running[16];
+    double start = now_ms();
+    for (int i = 0; i < threads; ++i) {
+        syncers[i].until = start + seconds * 1e3;
+        pthread_create(&running[i], NULL, commit_all, &syncers[i]);
+    }
+    int total = 0;
+    for (int i = 0; i < threads; ++i) {
+        pthread_join(running[i], NULL);
+        total += syncers[i].count;
+    }
+    double elapsed = now_ms() - start;
+    double *all = malloc((size_t)total * sizeof(*all));
+    int n = 0;
+    for (int i = 0; i < threads; ++i) {
+        memcpy(all + n, syncers[i].times, (size_t)syncers[i].count * sizeof(double));
+        n += syncers[i].count;
+        close(syncers[i].fd);
+        unlink(paths[i]);
+    }
+    qsort(all, (size_t)n, sizeof(double), order);
+    printf("{\"threads\":%d,\"bytes\":%zu,\"commits\":%d,\"commits_per_second\":%.1f,\"p50_ms\":%.2f,\"p90_ms\":%.2f,"
+           "\"p99_ms\":%.2f}\n", threads, size, n, n / (elapsed / 1e3), n ? all[n / 2] : 0, n ? all[n * 9 / 10] : 0,
+           n ? all[n * 99 / 100] : 0);
+    free(all);
+    free(syncers);
+    return 0;
+}
diff --git a/research/experiments/013-parallel-syncs/run.py b/research/experiments/013-parallel-syncs/run.py
new file mode 100644
index 0000000..ec2f230
--- /dev/null
+++ b/research/experiments/013-parallel-syncs/run.py
@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Does the device sync several stores at once? (docs/research/084-parallel-syncs.md)
+
+1. probe.c: 1, 2 and 4 threads, each committing to a file of its own
+   (bytes over zeros, fdatasync), the order rotated per round.
+2. The scoreboard app (benchmarks/http/scores.aup) taking durable posts:
+   one pure-serve owning one store under 64 connections, against two
+   pure-serve, each owning a store of its own (what sharding by key would
+   make), under 32 connections each at once. Servers on CPUs 0 and 1, load
+   generators on 3 and 2.
+Prints one JSON document; each round also goes to stderr as it ends.
+"""
+import argparse
+import json
+import select
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[2]
+HTTP = ROOT / 'benchmarks' / 'http'
+
+
+def start(cpu, module, store):
+    process = subprocess.Popen(['taskset', '-c', str(cpu), str(ROOT / 'build/pure-serve'), '--module', str(module),
+                                '--store', str(store), '--port', '0'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    ready, _, _ = select.select([process.stdout], [], [], 20)
+    assert ready, 'pure-serve did not start'
+    return process, json.loads(process.stdout.readline())['port']
+
+
+def loads(targets, seconds):
+    """Load generators at once, one per (cpu, port, connections); their JSON."""
+    running = [subprocess.Popen(['taskset', '-c', str(cpu), str(HTTP / 'load'), str(port), str(connections),
+                                 str(seconds), 'score', str(connections), 'keep-alive', '16'],
+                                stdout=subprocess.PIPE, text=True) for cpu, port, connections in targets]
+    return [json.loads(process.communicate()[0]) for process in running]
+
+
+def owners(count, module, directory, seconds):
+    stores = [Path(tempfile.mkdtemp(dir=directory, prefix='pure-sync-owner-')) for _ in range(count)]
+    servers = []
+    try:
+        for i, store in enumerate(stores):
+            shutil.rmtree(store)
+            servers.append(start(i, module, store))
+        measured = loads([(3 - i, port, 64 // count) for i, (_, port) in enumerate(servers)], seconds)
+        return {'owners': count, 'posts_per_second': sum(m['2xx'] for m in measured) / seconds,
+                'p50_us': max(m['p50_us'] for m in measured), 'p99_us': max(m['p99_us'] for m in measured),
+                'errors': sum(m['errors'] + m['4xx'] + m['5xx'] for m in measured)}
+    finally:
+        for process, _ in servers:
+            process.terminate()
+            process.wait()
+        for store in stores:
+            shutil.rmtree(store, ignore_errors=True)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--directory', type=Path, default=Path('/var/tmp'), help='on the device to measure')
+    parser.add_argument('--aup', type=Path, default=ROOT.parent / 'aup' / 'build' / 'aup')
+    parser.add_argument('--rounds', type=int, default=3)
+    parser.add_argument('--seconds', type=float, default=3)
+    args = parser.parse_args()
+    with tempfile.TemporaryDirectory() as work:
+        work = Path(work)
+        subprocess.run(['cc', '-O2', '-std=c17', '-pthread', '-o', str(work / 'probe'), str(HERE / 'probe.c')], check=True)
+        subprocess.run(['cc', '-O2', '-std=c17', '-o', str(HTTP / 'load'), str(HTTP / 'load.c')], check=True)
+        module = work / 'scores.pmod'
+        subprocess.run([str(args.aup), 'build', str(HTTP / 'scores.aup'), '-o', str(module)], check=True)
+        results = {'rounds': []}
+        for round_ in range(args.rounds):
+            measured = {'probe': [], 'app': []}
+            threads = [1, 2, 4]
+            threads = threads[round_ % 3:] + threads[:round_ % 3]
+            for count in threads:
+                probe = subprocess.run([str(work / 'probe'), str(args.directory), str(count), str(args.seconds), '4096'],
+                                       capture_output=True, text=True, check=True)
+                measured['probe'].append(json.loads(probe.stdout))
+            for count in ((1, 2) if round_ % 2 == 0 else (2, 1)):
+                measured['app'].append(owners(count, module, args.directory, args.seconds))
+            results['rounds'].append(measured)
+            print(json.dumps(measured), file=sys.stderr, flush=True)
+        print(json.dumps(results, indent=1))
+
+
+if __name__ == '__main__':
+    sys.exit(main())