From 5e90a99dfdcc0a16f52213370cda29779571e36c Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 27 Sep 2026 13:57:51 +0100 Subject: 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. --- docs/research/084-parallel-syncs.md | 47 +++++ .../2026-09-27-ext4-linux-aarch64.json | 192 +++++++++++++++++++++ research/experiments/013-parallel-syncs/probe.c | 96 +++++++++++ research/experiments/013-parallel-syncs/run.py | 92 ++++++++++ 4 files changed, 427 insertions(+) create mode 100644 docs/research/084-parallel-syncs.md create mode 100644 research/experiments/013-parallel-syncs/2026-09-27-ext4-linux-aarch64.json create mode 100644 research/experiments/013-parallel-syncs/probe.c create mode 100644 research/experiments/013-parallel-syncs/run.py 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 +#include +#include +#include +#include +#include +#include + +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()) -- cgit v1.2.3