diff options
Diffstat (limited to 'research')
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()) |