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.

4 files changed, 427 insertions, 0 deletions

diff --git a/docs/research/084-parallel-syncs.md b/docs/research/084-parallel-syncs.md
new file mode 100644
index 0000000..43c24df
--- /dev/null
+++ b/docs/research/084-parallel-syncs.md
@@ -0,0 +1,47 @@
+# 084 — 소유자를 키로 나누면: 이 장치는 동기화를 한 줄로 세운다
+
+2026-09-27. [로드맵](../roadmap.md) 중기 2 "키 해시 기반 단일 소유자 파티셔닝(쓰기 병목 선형 확장)"을 할지 정한 것.
+
+착수 기준:
+
+- 바꿀 결정: 상태를 가진 앱의 소유자를 키 해시로 여럿 두고(소유자마다 저장소 하나) 내구성 있는 쓰기를 코어 수만큼
+  늘릴지, 지금처럼 소유자 하나가 한 번의 동기화에 모든 이벤트를 모으게 둘지.
+- 남은 질문: 이 장치가 파일 여러 개의 `fdatasync`를 동시에 처리하는가? 소유자를 나누면 앱의 초당 쓰기가 느는가?
+- 최소 확인: 저장소 커밋과 같은 모양(0을 써 둔 곳을 덮어쓰고 `fdatasync`)을 파일마다 스레드 하나로 1·2·4개 동시에 돌려
+  합계를 재고, 점수판 앱을 저장소 하나의 소유자 하나와 저장소를 따로 가진 소유자 둘로 번갈아 잰다.
+- 중단 기준: 동기화 합계가 스레드 수를 따라 늘면(장치가 병렬로 처리하면) 샤딩을 설계한다.
+
+## 장치가 말한 것
+
+Raspberry Pi 5, SD 카드의 ext4, 4 KiB 커밋, 3초씩, 순서를 돌려 가며 네 라운드
+([probe.c](../../research/experiments/013-parallel-syncs/probe.c), [run.py](../../research/experiments/013-parallel-syncs/run.py),
+[원시 결과](../../research/experiments/013-parallel-syncs/2026-09-27-ext4-linux-aarch64.json)):
+
+| 동시에 동기화하는 파일 | 합계 커밋/s | p50 |
+|---|---:|---:|
+| 1 | 512–669 | 1.21–1.23 ms |
+| 2 | 556–664 | 2.42–2.45 ms |
+| 4 | 626–684 | 4.93–5.09 ms |
+
+합계는 그대로이고 지연만 파일 수에 비례한다. 장치가 동기화를 하나씩 처리한다(SD는 명령을 한 줄로 받는다).
+
+점수판 앱의 내구성 POST(연결 64개를 나눠 받음, 서버는 CPU 0·1, 부하는 CPU 3·2):
+
+| | posts/s | p99 |
+|---|---:|---:|
+| 소유자 하나, 저장소 하나 | 18,542–23,384 (중앙값 20,974) | 13.8–25.6 ms |
+| 소유자 둘, 저장소 둘 | 13,991–17,716 (중앙값 16,230) | 36.9–57.3 ms |
+
+나누면 23% 느리고 꼬리 지연이 2–3배다. 동기화는 줄을 서는데, 한 번의 동기화가 싣는 이벤트는 소유자마다 반으로 준다.
+소유자 하나의 그룹 커밋은 기다리는 이벤트를 모두 한 번의 동기화에 싣는다. 이 장치에서는 그것이 쓰기의 최선이다.
+
+## 결정
+
+- 키 해시 샤딩은 하지 않는다. 쓰기의 병목은 CPU가 아니라(소유자의 라운드에서 적용은 4%,
+  [081](081-tail-records.md)) 장치의 동기화 한 줄이다. 코어를 늘려 줄 것이 없다.
+- 병렬로 동기화하는 장치(여러 큐의 NVMe)에서도 저장소를 소유자마다 두기보다 한 저장소의 그룹 커밋을 여러 소유자가 함께
+  쓰는 쪽이 먼저다. 문서 서버는 이미 그렇게 스레드들의 커밋을 묶는다(`d4b01a7`). 앱의 적용이 CPU 하나를 넘을 때 소유자를
+  나누되 동기화는 장치마다 한 줄로 모은다.
+- 멀티코어는 읽기에 쓴다. 연결 스레드가 스냅숏으로 보기를 답한다(`--threads auto`, [080](080-snapshot-reads.md)).
+- 이 장치에서 쓰기를 더 늘리는 길은 동기화의 비용 자체다. 커밋 경로에서 꼬리 늘리기와 압축을 빼면 처리량은 5–8% 늘고
+  p99는 반이 된다([081](081-tail-records.md) 다음 결정).
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())