diff options
| -rw-r--r-- | docs/research/073-single-file-persistent-tree.md | 2 | ||||
| -rw-r--r-- | docs/research/075-http-host-against-axum.md | 15 | ||||
| -rw-r--r-- | docs/research/076-one-sync-commit.md | 144 | ||||
| -rw-r--r-- | prototypes/connections/store-tree.c | 392 | ||||
| -rw-r--r-- | prototypes/connections/test-tree-store.c | 173 | ||||
| -rw-r--r-- | research/experiments/010-one-sync-commit/2026-09-26-ext4-linux-aarch64.json | 148 | ||||
| -rw-r--r-- | research/experiments/010-one-sync-commit/commits.c | 47 | ||||
| -rw-r--r-- | research/experiments/010-one-sync-commit/probe.c | 70 | ||||
| -rw-r--r-- | research/experiments/010-one-sync-commit/run.py | 63 |
9 files changed, 938 insertions, 116 deletions
diff --git a/docs/research/073-single-file-persistent-tree.md b/docs/research/073-single-file-persistent-tree.md index 092b6ad..13884e3 100644 --- a/docs/research/073-single-file-persistent-tree.md +++ b/docs/research/073-single-file-persistent-tree.md @@ -36,6 +36,8 @@ - 커밋: 새 노드 추가 → `fsync` → 오래된 head 슬롯에 새 head → `fsync`. OK는 두 번째 동기화 뒤다. 복구는 유효한 슬롯 중 순번이 큰 쪽을 쓰고, 그 끝 뒤의 바이트는 확인받지 않은 커밋이라 다음 writer가 잘라낸다. 쓰기 중 고장 단계는 기존 저장소와 같은 결과(오류 또는 결과 불명)를 낸다. + (2026-09-26: 동기화 한 번으로 바뀌었다. head가 커밋 바이트의 다이제스트를 갖고, 커밋은 0을 써 둔 꼬리를 + 덮어쓴다 — [076](076-one-sync-commit.md).) - 압축: 공개 크기의 증가가 `4 × 압축 후 크기 + 1 MiB`를 넘으면 도달 가능한 노드만 새 파일에 쓰고 교체한다. 이미 읽은 쪽은 이전 파일의 매핑을 계속 사용한다. - 읽기: 한 leaf 안의 범위는 파일을 그대로 매핑한다(기본 저장소와 같은 무복사). 여러 leaf에 diff --git a/docs/research/075-http-host-against-axum.md b/docs/research/075-http-host-against-axum.md index ab26a37..8a03889 100644 --- a/docs/research/075-http-host-against-axum.md +++ b/docs/research/075-http-host-against-axum.md @@ -123,4 +123,17 @@ ThreadSanitizer는 이 커널(47비트 가상 주소)을 지원하지 않아 쓰 **저장 보장 쓰기 측정의 함정.** 저장 장치는 SD 카드 하나뿐이고, 동기 쓰기가 몇 분 이어지면 내부 정리로 크게 느려진다 (같은 설정이 순서에 따라 연결 8개에서 350/s → 17/s, p99 4.4초). 앞의 표들의 쓰기 수치는 순서 영향을 받았다. 쓰기 비교는 -설정 순서를 섞어 여러 번 돌리고, 매번 장치의 fsync 지연을 함께 재서 보고해야 한다(다음 작업). +설정 순서를 섞어 여러 번 돌리고, 매번 장치의 fsync 지연을 함께 재서 보고해야 한다. 하네스가 이제 그렇게 한다 +(`--rounds`, 라운드마다 서버 순서 회전, 쓰기 직전 장치 탐침). + +## 4단계: 커밋당 동기화 한 번 + +저장소 커밋을 동기화 한 번으로 줄였다([076](076-one-sync-commit.md)). 같은 하네스로 3초 × 3라운드의 중앙값: + +| PUT 16 B, keep-alive | pure-serve | pure-serve 3샤드 | axum 비안전 | axum 충돌 안전 | +|---|---:|---:|---:|---:| +| 연결 64 | **2,012** (p50 20.5 ms) | **2,611** (12.3 ms) | 1,040 (41.0 ms) | 583 (69.6 ms) | +| 연결 8 | **601** (6.4 ms) | **637** (7.7 ms) | 260 (18.4 ms) | 166 (30.7 ms) | + +충돌에 안전한 axum 대비 쓰기 처리량 3.4–4.5배, 봉우리 RSS 3.0–4.6 MiB 대 7.3–9.6 MiB다. 이제 읽기, 1 MiB 읽기, +저장 보장 쓰기 모두에서 axum보다 빠르고 메모리를 덜 쓴다. diff --git a/docs/research/076-one-sync-commit.md b/docs/research/076-one-sync-commit.md new file mode 100644 index 0000000..8176a82 --- /dev/null +++ b/docs/research/076-one-sync-commit.md @@ -0,0 +1,144 @@ +# 076 — 커밋당 동기화 한 번: 순서를 동기화로 강제하지 않고 헤드가 검증한다 + +2026-09-26. 트리 저장소(`store-tree.c`)의 커밋을 동기화 두 번에서 한 번으로 줄였다. 파일 형식이 바뀌었다 +(`PURETRE1` → `PURETRE2`, 이전 파일은 읽지 않는다). + +착수 기준: + +- 바꿀 결정: 커밋의 내구성을 "데이터 동기화 → 헤드 기록 → 헤드 동기화"라는 순서로 보장할지, 헤드가 자기 + 커밋을 검증하게 해서 한 번의 동기화로 보장할지. +- 남은 질문: 동기화 한 번이 이 장치에서 실제로 얼마나 싼가? 동기화 도중 전원이 나가 헤드만 남은 커밋을 확실히 + 버릴 수 있는가? 그 확인 비용을 읽기 경로가 매번 치르지 않을 수 있는가? +- 최소 확인: 장치에서 방식별 비용을 같은 조건으로 재고, 새 커밋을 기존 고장·오라클 검사와 새 "잃어버린 커밋" + 검사에 통과시킨 뒤 저장소와 HTTP 수준에서 이전 커밋과 번갈아 잰다. +- 중단 기준: 장치 비용이 절반 이하로 줄지 않거나, 잃어버린 커밋을 받아들이는 경우가 하나라도 있으면 닫는다. + +## 의심한 관례 + +| 관례 | 바꾼 것 | +|---|---| +| 커밋의 원자성은 쓰기 순서를 동기화로 강제해 얻는다 | 헤드가 커밋 바이트의 다이제스트를 가진다. 바이트가 다 닿지 않은 헤드는 스스로 드러난다 | +| 파일 끝에 덧붙인다 | 미리 0을 써 둔 꼬리를 덮어쓴다. 크기도 블록 할당도 바뀌지 않아 저널 커밋이 없다 | +| `fsync` | `fdatasync`(시각은 동기화하지 않는다) | +| 두 헤드 슬롯은 파일 첫 블록에 나란히 | 슬롯마다 4 KiB 블록 하나. 한 슬롯의 찢긴 쓰기가 다른 슬롯에 닿지 않는다 | +| 읽기는 보이는 최신 헤드를 믿는다 | 같은 프로세스가 아직 동기화 중인 헤드는 읽기가 건너뛴다 | + +## 장치가 말한 것 + +같은 Raspberry Pi 5, ext4 SD 카드(`/dev/mmcblk0p2`). 8 KiB 커밋 + 64 B 헤드를 방식별로 40번, 방식들을 번갈아 +가며 쟀다([probe.c](../../research/experiments/010-one-sync-commit/probe.c)). 세 번 반복의 p50 범위(ms): + +| 방식 | p50 | +|---|---:| +| 덧붙이기, fsync, 헤드, fsync (이전) | 8.82–9.19 | +| 덧붙이기, 헤드, fdatasync | 5.61–5.82 | +| 덧붙이기, 헤드, fsync | 5.66–6.21 | +| **0을 써 둔 곳 덮어쓰기, 헤드, fdatasync** | **2.41–2.53** | +| 0을 써 둔 곳 덮어쓰기, 헤드, fsync | 5.07–5.75 | + +동기화 횟수만 줄이면 약 1.6배다. 나머지는 파일 크기·블록 할당·수정 시각 같은 메타데이터 때문에 생기는 저널 커밋이다. +세 가지를 모두 없애야 약 3.7배가 된다. + +## 구조 + +- 파일: `[슬롯 0 블록][슬롯 1 블록][커밋된 노드…][0으로 채운 꼬리]`. 데이터는 8 KiB부터다. +- 헤드 슬롯(128 B): 순번, 공개 끝, 카탈로그 루트, 압축 후 크기, **커밋 시작 위치, 커밋 바이트의 다이제스트**, + 슬롯 자체의 다이제스트. 다이제스트는 xxHash64 구성(네 갈래 곱셈-회전, 수 GB/s)이고, 찢김과 누락 검출용이지 + 적대적 파일 방어용은 아니다. +- 커밋: 쓰는 쪽이 노드를 버퍼에서 내보낼 때마다 다이제스트를 갱신한다(데이터를 다시 읽지 않는다). 노드를 + 공개 끝부터 쓰고, 오래된 슬롯에 헤드를 쓰고, `fdatasync`를 한 번 한다. OK는 그 뒤다. +- 복구: 최신 슬롯의 헤드가 가리키는 바이트의 다이제스트가 맞아야 그 헤드를 쓴다. 틀리면 이전 슬롯을 쓴다. + 쓰는 쪽은 항상 내구성 있는 헤드 위에서만 쓰기 때문에 이전 슬롯은 검사할 필요가 없다. 새 헤드는 이전 커밋의 + 동기화가 끝난 뒤에야 쓰였다. 공개되지 않은 바이트는 꼬리에 남아 있다가 다음 커밋에 덮어써진다. 잘라 내지 않는다. +- 헤드 확인 비용: 프로세스는 자기가 공개했거나 검사한 헤드를 기억한다(32개). 그래서 커밋 바이트를 헤드당 + 최대 한 번 읽고, 자기가 쓴 커밋은 읽지 않는다. 쓰는 쪽은 자기가 내구성을 확인하지 않은 헤드(죽은 writer가 + 동기화 전에 남긴 것일 수 있다)를 만나면 한 번 동기화한 뒤 그 위에 쓴다. 검사에 실패한 헤드도 기억해 다시 + 읽지 않는다. +- 읽기 격리: 헤드를 쓰기 직전에 "동기화 중" 표시를 하고 동기화가 끝나면 지운다. 이 프로세스의 읽기는 표시된 + 헤드를 건너뛰고 이전 헤드를 읽는다. 그래서 전원이 나가면 사라질 커밋으로 답하지 않는다. 샤드가 여러 개일 때도 + 같다. 다른 프로세스의 읽기는 이전 설계와 마찬가지로 동기화 전의 헤드를 볼 수 있다. +- 꼬리: writer 세션을 열 때 꼬리가 64 KiB보다 짧으면 `max(256 KiB, 파일의 1/8)`만큼 0을 쓴다. 그 0은 다음 + 커밋의 동기화와 함께 내구성을 얻는다. 저널 커밋은 그 한 커밋만 치른다. 압축으로 만든 새 파일도 꼬리를 가진 채 + 설치된다. +- 핸들의 새로 고침: 매핑이 헤드 슬롯을 보여 주므로, 바뀌지 않은 저장소를 확인하는 데는 `fstatat` 하나만 든다 + (이전에는 `pread`까지 둘). 새 헤드는 언제나 현재 헤드가 없는 쪽 슬롯에 먼저 쓰이므로 그 슬롯의 순번만 보면 된다. + +## 측정 + +**저장소 단독.** [commits.c](../../research/experiments/010-one-sync-commit/commits.c)는 64개 문서 중 하나의 앞에 +16 B를 넣는 커밋을 200번 연달아 한다. 동기화 두 번 판(`ad2197c`)과 한 번 판을 라운드마다 순서를 바꿔 가며 세 번 +쟀다([run.py](../../research/experiments/010-one-sync-commit/run.py), +[원시 결과](../../research/experiments/010-one-sync-commit/2026-09-26-ext4-linux-aarch64.json)). + +| 커밋 | 동기화 두 번 | 동기화 한 번 | +|---|---:|---:| +| 초당 커밋 | 94–109 | **289–304** | +| p50 | 8.82–9.19 ms | **2.44–2.48 ms** | +| p90 | 11.15–12.20 ms | **4.69–5.49 ms** | +| p99 | 61.7–63.2 ms | 19.8–56.8 ms | + +p99는 장치 쪽 잡음이 지배한다. 꼬리를 늘리는 단위를 64 KiB, 256 KiB, 1 MiB로 바꿔 봐도 p99가 일정하게 움직이지 +않았다. + +**HTTP.** [075](075-http-host-against-axum.md)의 하네스를 고쳐 라운드마다 서버 순서를 바꾸고 쓰기 직전에 장치를 +재게 했다. 3초 × 3라운드의 중앙값이고, 성공 응답/초와 괄호 안 p50(ms)이다. 두 판 모두 장치 탐침은 1.0–1.5 ms였다 +(연결 8개 1스레드 이전 판은 2.5–3.1 ms). + +| PUT 16 B, keep-alive | pure-serve 이전 | pure-serve | 3샤드 이전 | 3샤드 | axum 비안전 | axum 충돌 안전 | +|---|---:|---:|---:|---:|---:|---:| +| 연결 64 | 1,245 | **2,012** (20.5) | 1,980 | **2,611** (12.3) | 1,040 (41.0) | 583 (69.6) | +| 연결 8 | 380 | **601** (6.4) | 313 | **637** (7.7) | 260 (18.4) | 166 (30.7) | + +충돌에 안전한 axum과 비교하면 쓰기 처리량은 3.4–4.5배이고, p50은 3.8–5.7배 짧다. 봉우리 RSS는 pure-serve +3.0 MiB, 3샤드 4.6 MiB, axum 7.3–9.6 MiB였다. + +## 검증 + +[`test-tree-store.c`](../../prototypes/connections/test-tree-store.c)는 ASan·UBSan에서도 통과했다. 새로 넣은 검사: + +- **잃어버린 커밋**: 최신 커밋의 바이트 하나를 바꾸고 프로세스의 기억을 지우면(재부팅 뒤의 새 프로세스와 같다) + 읽기, 핸들, writer가 모두 이전 커밋을 쓴다. 다음 커밋은 잃어버린 커밋이 있던 자리부터 쓴다. +- **동기화 중인 헤드**: 헤드를 쓰고 동기화 전에 멈춘 사이 같은 프로세스의 핸들과 읽기는 이전 버전을 보고, + 동기화가 끝나면 새 버전을 본다. +- **동기화 한 번**: 커밋 20번에 동기화가 정확히 20번이고 파일 크기가 그대로다. +- **고장 단계**: 동기화 전에 죽은 writer의 헤드를 이어받는 첫 커밋은 동기화를 두 번 한다. 확인 한 번과 커밋 한 번이다. + 다른 단계의 커밋은 한 번이다. 찢긴 최신 슬롯, 1비트 손상, 격리, 압축, 그룹 커밋 검사도 새 배치에서 그대로 통과한다. + +모듈 검사(`make check-module`, ASan 포함), `make check-tree`(런타임·splice·async와 실제 PTY 편집기), aup와 vega의 +검사(서버 병렬성·파이프라이닝·kill -9 내구성 포함)가 모두 통과했다. + +이 과정에서 기존 Makefile 버그를 찾았다. `MODULE` 변수가 `pure-serve` 규칙보다 뒤에서 정의돼 있어서, 저장소나 모듈 +실행기를 고쳐도 `pure-serve`가 다시 빌드되지 않았다(규칙의 선행 조건은 읽을 때 전개된다). 정의를 앞으로 옮겼다. +첫 HTTP 측정은 이 때문에 이전 바이너리로 돌았고, 그 값이 위 표의 "이전" 열이다. + +## 비용과 한계 + +- **공간.** 파일 끝에 256 KiB 또는 파일 크기의 1/8까지 0이 있다. 헤드 두 슬롯이 8 KiB를 차지한다. 작은 저장소가 + 이전보다 커진다. +- **장치 쓰기가 두 배.** 커밋 바이트가 들어갈 자리에 0을 먼저 쓰므로 장치는 같은 블록을 두 번 쓴다. 작은 커밋에서는 + 저널 커밋(약 3.5 ms)보다 훨씬 싸지만, 큰 문서를 잇달아 쓰는 작업은 대역폭을 더 쓴다. 꼬리를 늘리는 커밋은 0을 + 쓰고 저널 커밋까지 치른다. +- **다른 프로세스의 첫 커밋.** 새 프로세스의 첫 writer 세션은 확인 동기화를 한 번 더 한다. `aup run --store`처럼 + 짧은 명령은 커밋마다 동기화가 두 번이다. 다른 프로세스의 커밋을 처음 읽을 때는 그 커밋의 바이트를 한 번 읽는다. +- **격리는 프로세스 안에서만.** 동기화가 끝나지 않은 커밋은 같은 프로세스에서는 보이지 않지만, 다른 프로세스의 + 읽기에는 보인다(이전 설계도 두 번째 동기화 동안 같았다). +- **형식 변경.** `PURETRE1` 파일은 열리지 않는다(`pure_store_init`은 읽을 수 없는 파일을 바꾸지 않는다). 변환 도구는 + 없다. +- macOS는 `fdatasync` 대신 `fsync`를 쓰고, `F_FULLFSYNC`는 쓰지 않는다. 실제 전원 차단, macOS, x86_64는 검증하지 + 않았다. 페이지 크기가 16 KiB인 이 커널에서 두 슬롯은 같은 페이지에 있다. 블록 단위로 쓰는 파일 시스템에서는 + 서로 닿지 않지만, 장치가 바뀌지 않은 블록까지 다시 쓰는 경우의 찢김은 장치 동작에 맡긴다. + +## 선행 기술 + +ext4/jbd2의 `journal_async_commit`은 커밋 블록에 체크섬을 넣어, 데이터 블록을 기다리지 않고 커밋 블록을 쓰게 한다. +SQLite WAL은 프레임마다 누적 체크섬을 두어 커밋당 동기화 한 번으로 끝낸다. PostgreSQL은 WAL 세그먼트를 0으로 채워 +재사용한다(`wal_init_zero`, `wal_recycle`). LMDB는 두 메타 페이지를 서로 다른 페이지에 둔다. 다이제스트는 xxHash64 +구성이다. 이번 일은 이 기법들을 Pure의 한 파일 트리에 맞게 결합한 것이며 신규성 주장이 아니다. Pure 쪽 기여는 +프로세스 안의 "동기화 중" 표시로, 커밋을 공개 경로에서 떼어 내지 않고도 이 프로세스의 읽기가 내구성 없는 +커밋을 보지 않게 했다는 점이다. + +## 다음 결정 + +1. 꼬리를 늘리는 비용을 커밋 경로에서 빼기(백그라운드 준비나 세그먼트 재사용). 지금은 한 커밋이 치른다. +2. 여러 프로세스가 같은 저장소를 쓸 때의 확인 동기화를 줄이는 방법(동기화 완료를 공유 메모리로 알리기). +3. 동기화를 다음 묶음의 실행과 겹치기. 커밋 결과를 기다리는 동안 다음 묶음의 이벤트를 준비한다. diff --git a/prototypes/connections/store-tree.c b/prototypes/connections/store-tree.c index 3163265..a5d24c2 100644 --- a/prototypes/connections/store-tree.c +++ b/prototypes/connections/store-tree.c @@ -3,8 +3,9 @@ * path it changes, so inserting or deleting in a large document costs O(log n) * nodes instead of a whole-document copy. Unchanged extents are shared by old * and new versions. The document catalog is the same structure holding fixed - * records sorted by key. Two alternating head slots publish a commit only after - * its nodes are durable; readers take no lock and map one immutable prefix. */ + * records sorted by key. A commit appends its nodes and writes a head naming + * them and their digest into the older of two slots, then syncs once; readers + * take no lock and map one immutable prefix. */ #define _POSIX_C_SOURCE 200809L #ifdef __APPLE__ #define _DARWIN_C_SOURCE @@ -23,9 +24,13 @@ #include <unistd.h> enum { FREE = 0xffff }; -enum { SLOT = 64, DATA = 2 * SLOT, LEAF_MAX = 4096, LEAF_MIN = 1024, +/* Each head slot has a block of its own: a torn write of one never touches + * the other. */ +enum { SLOT = 128, BLOCK = 4096, DATA = 2 * BLOCK, LEAF_MAX = 4096, LEAF_MIN = 1024, FAN_MAX = 32, FAN_MIN = 8, RECORD = 96, KEY = 64, HEIGHT_MAX = 12, BUFFER = 65536 }; +/* Written zeros kept after the published end (see prepare_tail). */ +enum { TAIL_MIN = 65536, TAIL_STEP = 262144 }; #ifdef PURE_STORE_TEST enum { COMPACT_SLACK = 65536 }; #else @@ -34,7 +39,7 @@ enum { COMPACT_SLACK = 1048576 }; #define LEAF_BIT (UINT64_C(1) << 63) #define MEMORY_BIT (UINT64_C(1) << 62) #define LOCATION (MEMORY_BIT - 1) -static const unsigned char head_magic[8] = {'P','U','R','E','T','R','E','1'}; +static const unsigned char head_magic[8] = {'P','U','R','E','T','R','E','2'}; static const unsigned char branch_magic[4] = {'P','T','B','1'}; static const char tree_name[] = "pure.tree"; /* Threads of one process take turns as the writer: the file lock arbitrates @@ -64,12 +69,18 @@ typedef struct { size_t count, capacity, free_head, free_count; /* free_head is slot + 1 */ size_t bytes; /* payload held by arena nodes, including unreachable ones */ } Tree; -typedef struct { uint64_t seq, end, catalog, catalog_bytes, live; } Head; +/* A commit's bytes are start..end (none for a file synced whole before it was + * published); check is their digest. */ +typedef struct { uint64_t seq, end, catalog, catalog_bytes, live, start, check; } Head; +/* The xxHash64 construction without a seed: four multiply-rotate lanes over + * 32-byte stripes, several GB/s. An all-zero Digest is an empty one. */ +typedef struct { uint64_t lane[4], length; unsigned char held[32]; size_t count; } Digest; typedef struct { int fd; uint64_t at; /* file offset of the next byte */ size_t used; bool failed; + Digest digest; /* of every byte written */ unsigned char buffer[BUFFER]; } Writer; typedef struct { PureSpan part[3]; size_t index, offset; } Source; @@ -111,19 +122,100 @@ static void encode64(unsigned char *bytes, uint64_t value) { for (unsigned i = 0; i < 8; ++i) bytes[i] = (unsigned char)(value >> (8 * i)); } -/* Detects a torn head slot; it is not a defence against a hostile file. */ -static uint64_t checksum(const unsigned char *bytes, size_t length) { - uint64_t hash = UINT64_C(14695981039346656037); - for (size_t i = 0; i < length; ++i) hash = (hash ^ bytes[i]) * UINT64_C(1099511628211); - return hash; +static uint64_t load64(const unsigned char *bytes) { + uint64_t value; + memcpy(&value, bytes, 8); +#if defined(__BYTE_ORDER__) && __BYTE_ORDER__ == __ORDER_BIG_ENDIAN__ + value = __builtin_bswap64(value); +#endif + return value; +} + +/* Digests find torn and missing writes; they are not a defence against a + * hostile file. */ +static const uint64_t P1 = UINT64_C(0x9E3779B185EBCA87), P2 = UINT64_C(0xC2B2AE3D27D4EB4F), + P3 = UINT64_C(0x165667B19E3779F9), P4 = UINT64_C(0x85EBCA77C2B2AE63), + P5 = UINT64_C(0x27D4EB2F165667C5); + +static uint64_t rotate(uint64_t value, unsigned bits) { return value << bits | value >> (64 - bits); } +static uint64_t mix(uint64_t lane, uint64_t word) { return rotate(lane + word * P2, 31) * P1; } + +static void stripes(Digest *digest, const unsigned char *bytes, size_t count) { + uint64_t a = digest->lane[0], b = digest->lane[1], c = digest->lane[2], d = digest->lane[3]; + for (; count; --count, bytes += 32) { + a = mix(a, load64(bytes)); + b = mix(b, load64(bytes + 8)); + c = mix(c, load64(bytes + 16)); + d = mix(d, load64(bytes + 24)); + } + digest->lane[0] = a, digest->lane[1] = b, digest->lane[2] = c, digest->lane[3] = d; +} + +static void digest_add(Digest *digest, const unsigned char *bytes, size_t length) { + digest->length += length; + if (digest->count) { + size_t take = 32 - digest->count < length ? 32 - digest->count : length; + memcpy(digest->held + digest->count, bytes, take); + digest->count += take; + bytes += take; + length -= take; + if (digest->count < 32) return; + stripes(digest, digest->held, 1); + digest->count = 0; + } + stripes(digest, bytes, length / 32); + digest->count = length % 32; + if (digest->count) memcpy(digest->held, bytes + length - digest->count, digest->count); } +static uint64_t digest_end(const Digest *digest) { + const uint64_t *lane = digest->lane; + uint64_t hash = rotate(lane[0], 1) + rotate(lane[1], 7) + rotate(lane[2], 12) + rotate(lane[3], 18); + for (unsigned i = 0; i < 4; ++i) hash = (hash ^ mix(0, lane[i])) * P1 + P4; + hash += digest->length; + size_t i = 0; + for (; i + 8 <= digest->count; i += 8) hash = rotate(hash ^ mix(0, load64(digest->held + i)), 27) * P1 + P4; + for (; i < digest->count; ++i) hash = rotate(hash ^ digest->held[i] * P5, 11) * P1; + hash ^= hash >> 33; + hash *= P2; + hash ^= hash >> 29; + hash *= P3; + return hash ^ hash >> 32; +} + +static uint64_t digest_of(const unsigned char *bytes, size_t length) { + Digest digest = {0}; + digest_add(&digest, bytes, length); + return digest_end(&digest); +} + +#ifdef PURE_STORE_TEST +static unsigned long syncs; +unsigned long pure_tree_test_syncs(void) { return __atomic_load_n(&syncs, __ATOMIC_RELAXED); } +#define COUNT_SYNC() __atomic_add_fetch(&syncs, 1, __ATOMIC_RELAXED) +#else +#define COUNT_SYNC() ((void)0) +#endif + static bool sync_fd(int fd) { int rc; + COUNT_SYNC(); do { rc = fsync(fd); } while (rc && errno == EINTR); return rc == 0; } +/* Data and the size needed to read it back, without timestamps. */ +static bool sync_data(int fd) { +#ifdef __APPLE__ + return sync_fd(fd); +#else + int rc; + COUNT_SYNC(); + do { rc = fdatasync(fd); } while (rc && errno == EINTR); + return rc == 0; +#endif +} + static bool write_at(int fd, const unsigned char *bytes, size_t length, uint64_t offset) { while (length) { ssize_t n = pwrite(fd, bytes, length, (off_t)offset); @@ -156,35 +248,130 @@ static void encode_head(unsigned char out[SLOT], const Head *head) { encode64(out + 24, head->catalog); encode64(out + 32, head->catalog_bytes); encode64(out + 40, head->live); - encode64(out + 56, checksum(out, 56)); + encode64(out + 48, head->start); + encode64(out + 56, head->check); + encode64(out + SLOT - 8, digest_of(out, SLOT - 8)); } static bool decode_head(const unsigned char in[SLOT], Head *head) { - if (memcmp(in, head_magic, sizeof(head_magic)) || decode64(in + 56) != checksum(in, 56)) + if (memcmp(in, head_magic, sizeof(head_magic)) || decode64(in + SLOT - 8) != digest_of(in, SLOT - 8)) return false; - *head = (Head){decode64(in + 8), decode64(in + 16), decode64(in + 24), - decode64(in + 32), decode64(in + 40)}; - return head->seq && head->end >= DATA && head->end <= (uint64_t)INT64_MAX && - head->end <= SIZE_MAX && head->catalog_bytes % RECORD == 0 && + *head = (Head){decode64(in + 8), decode64(in + 16), decode64(in + 24), decode64(in + 32), + decode64(in + 40), decode64(in + 48), decode64(in + 56)}; + return head->seq && head->end <= (uint64_t)INT64_MAX && head->end <= SIZE_MAX && + head->start >= DATA && head->start <= head->end && head->catalog_bytes % RECORD == 0 && !head->catalog == !head->catalog_bytes; } -/* The newer valid slot wins; a torn newer slot leaves the previous commit. */ -static bool read_head(int fd, Head *out) { - unsigned char slots[2 * SLOT]; - Head a, b; - if (!read_at(fd, slots, sizeof(slots), 0)) return false; - bool first = decode_head(slots, &a), second = decode_head(slots + SLOT, &b); - if (!first && !second) return false; - *out = first && (!second || a.seq > b.seq) ? a : b; - return true; +/* ---- Trusted heads -------------------------------------------------------- + * One sync makes a commit's bytes and its head durable together, so a crash + * during it can leave the head on the device without all of the bytes: a + * head counts only once its digest matches them. The older slot needs no + * such luck. A writer builds only on a durable head, so the newer head was + * written after the older one's commit was complete on the device. + * The process remembers the heads it published or checked, so it reads a + * commit's bytes at most once, and which of them it made durable. */ +enum { UNKNOWN, BROKEN, CHECKED, DURABLE, PUBLISHING }; +enum { KNOWN = 32 }; +typedef struct { dev_t device; ino_t inode; uint64_t seq, check; int state; } Known; +static struct { + pthread_mutex_t lock; + Known heads[KNOWN]; + unsigned next; + Known publishing; /* written here, not durable yet */ +} known = {.lock = PTHREAD_MUTEX_INITIALIZER}; + +static bool names(const Known *known_head, const struct stat *file, const Head *head) { + return known_head->state && known_head->device == file->st_dev && known_head->inode == file->st_ino && + known_head->seq == head->seq && known_head->check == head->check; +} + +static int recall(const struct stat *file, const Head *head) { + int state = UNKNOWN; + pthread_mutex_lock(&known.lock); + if (names(&known.publishing, file, head)) state = PUBLISHING; + for (unsigned i = 0; i < KNOWN && state == UNKNOWN; ++i) + if (names(&known.heads[i], file, head)) state = known.heads[i].state; + pthread_mutex_unlock(&known.lock); + return state; +} + +/* PUBLISHING marks the one head being made durable; any other state ends + * that mark (UNKNOWN only ends it). */ +static void remember(const struct stat *file, const Head *head, int state) { + Known entry = {file->st_dev, file->st_ino, head->seq, head->check, state}; + pthread_mutex_lock(&known.lock); + if (state == PUBLISHING) known.publishing = entry; + else if (names(&known.publishing, file, head)) known.publishing.state = UNKNOWN; + if (state != PUBLISHING && state != UNKNOWN) { + Known *at = NULL; + for (unsigned i = 0; i < KNOWN && !at; ++i) + if (names(&known.heads[i], file, head)) at = &known.heads[i]; + if (!at) known.heads[known.next++ % KNOWN] = entry; + else if (at->state != DURABLE) *at = entry; /* a durable head stays durable */ + } + pthread_mutex_unlock(&known.lock); } -static bool map_tree(int fd, const Head *head, Tree *tree) { - struct stat info; +#ifdef PURE_STORE_TEST +/* Test-only: forget every head, as a new process after a crash would. */ +void pure_tree_test_forget(void) { + pthread_mutex_lock(&known.lock); + memset(known.heads, 0, sizeof(known.heads)); + known.publishing.state = UNKNOWN; + pthread_mutex_unlock(&known.lock); +} +#endif + +/* Whether a head's commit has all of its bytes. */ +static bool whole(int fd, const Head *head) { + unsigned char buffer[16384]; + Digest digest = {0}; + for (uint64_t at = head->start; at < head->end;) { + size_t n = head->end - at < sizeof(buffer) ? (size_t)(head->end - at) : sizeof(buffer); + if (!read_at(fd, buffer, n, at)) return false; + digest_add(&digest, buffer, n); + at += n; + } + return digest_end(&digest) == head->check; +} + +/* The newest head that counts. A reader passes over the head this process is + * still making durable, so the process never answers with a commit that a + * crash could take back; a writer never meets one (it is the only writer). */ +static bool pick(int fd, const struct stat *file, const unsigned char *slots[2], bool writer, Head *out) { + Head heads[2]; + bool valid[2]; + for (unsigned i = 0; i < 2; ++i) + valid[i] = decode_head(slots[i], &heads[i]) && heads[i].end <= (uint64_t)file->st_size; + unsigned newest = valid[1] && (!valid[0] || heads[1].seq > heads[0].seq); + for (unsigned turn = 0; turn < 2; ++turn) { + unsigned slot = turn ? !newest : newest; + const Head *head = &heads[slot]; + if (!valid[slot]) continue; + int state = recall(file, head); + if (state == BROKEN || (state == PUBLISHING && !writer)) continue; + if (state == UNKNOWN) { + bool counts = whole(fd, head); + remember(file, head, counts ? CHECKED : BROKEN); + if (!counts) continue; + } + *out = *head; + return true; + } + return false; +} + +static bool read_head(int fd, struct stat *file, bool writer, Head *out) { + unsigned char slots[BLOCK + SLOT]; + return !fstat(fd, file) && S_ISREG(file->st_mode) && file->st_size >= 0 && + read_at(fd, slots, sizeof(slots), 0) && + pick(fd, file, (const unsigned char *[2]){slots, slots + BLOCK}, writer, out); +} + +static bool map_tree(int fd, const struct stat *file, const Head *head, Tree *tree) { *tree = (Tree){0}; - if (fstat(fd, &info) || !S_ISREG(info.st_mode) || info.st_size < 0 || - (uint64_t)info.st_size < head->end) return false; + if ((uint64_t)file->st_size < head->end) return false; void *map = mmap(NULL, (size_t)head->end, PROT_READ, MAP_SHARED, fd, 0); if (map == MAP_FAILED) return false; *tree = (Tree){.map = map, .end = head->end}; @@ -534,13 +721,29 @@ static bool rebuild(Tree *tree, Entry root, uint64_t offset, uint64_t remove, } static bool flush(Writer *writer) { - if (writer->used && !writer->failed && - !write_at(writer->fd, writer->buffer, writer->used, writer->at - writer->used)) - writer->failed = true; + if (writer->used && !writer->failed) { + digest_add(&writer->digest, writer->buffer, writer->used); + if (!write_at(writer->fd, writer->buffer, writer->used, writer->at - writer->used)) writer->failed = true; + } writer->used = 0; return !writer->failed; } +/* Zeros from..to; a failure only costs speed. */ +static void zero_fill(int fd, uint64_t from, uint64_t to) { + static const unsigned char zeros[BUFFER]; + while (from < to) { + size_t n = to - from < BUFFER ? (size_t)(to - from) : BUFFER; + if (!write_at(fd, zeros, n, from)) return; + from += n; + } +} + +/* Where the written zeros after a published end should reach. */ +static uint64_t tail_end(uint64_t end) { + return end + ((end - DATA) / 8 > TAIL_STEP ? (end - DATA) / 8 : TAIL_STEP); +} + static void put_bytes(Writer *writer, const unsigned char *bytes, size_t length) { while (length && !writer->failed) { if (writer->used == BUFFER && !flush(writer)) return; @@ -647,23 +850,29 @@ static int create_temporary(int directory, char name[96]) { return -1; } -/* Publish a fresh file with both slots pointing at head. */ -static bool install(int directory, char temporary[96], int fd, const Head *head) { - unsigned char slots[2 * SLOT]; - encode_head(slots, head); - memcpy(slots + SLOT, slots, SLOT); - bool ok = write_at(fd, slots, sizeof(slots), 0) && sync_fd(fd); +/* Publish a fresh file with both slots pointing at head. It is synced whole + * before it is renamed into place, so its head names no commit bytes. */ +static bool install(int directory, char temporary[96], int fd, Head *head) { + unsigned char slot[SLOT]; + struct stat file; + head->start = head->end; + head->check = digest_end(&(Digest){0}); + encode_head(slot, head); + zero_fill(fd, head->end, tail_end(head->end)); + bool ok = write_at(fd, slot, SLOT, 0) && write_at(fd, slot, SLOT, BLOCK) && sync_fd(fd) && !fstat(fd, &file); if (close(fd)) ok = false; ok = ok && !renameat(directory, temporary, directory, tree_name); if (ok) temporary[0] = 0; - return ok && sync_fd(directory); + ok = ok && sync_fd(directory); + if (ok) remember(&file, head, DURABLE); + return ok; } /* Rewrite only reachable nodes. Open readers keep their mapping of the * previous file. A failure leaves the published file untouched. */ -static void compact(int directory, int source, const Head *head) { +static void compact(int directory, int source, const struct stat *file, const Head *head) { Tree tree; - if (!map_tree(source, head, &tree)) return; + if (!map_tree(source, file, head, &tree)) return; char temporary[96]; int fd = create_temporary(directory, temporary); Writer *writer = fd >= 0 ? malloc(sizeof(*writer)) : NULL; @@ -686,7 +895,7 @@ static void compact(int directory, int source, const Head *head) { ok = ok && rebuild(&tree, (Entry){0}, 0, 0, (PureSpan){records, (size_t)(count * RECORD)}, &fresh) && write_root(&tree, writer, &fresh, false) && flush(writer); if (ok) { - Head compacted = {head->seq + 1, writer->at, fresh.ref, fresh.bytes, writer->at - DATA}; + Head compacted = {head->seq + 1, writer->at, fresh.ref, fresh.bytes, writer->at - DATA, 0, 0}; ok = install(directory, temporary, fd, &compacted); fd = -1; } @@ -716,9 +925,10 @@ PureResult pure_store_init(const char *path) { bool ok = lock >= 0 && fstat(lock, &info) == 0 && S_ISREG(info.st_mode) && lock_writer(lock, true); if (ok) { Head head; + struct stat file; int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC); if (fd >= 0) { - ok = read_head(fd, &head); /* An unreadable file is never replaced. */ + ok = read_head(fd, &file, true, &head); /* An unreadable file is never replaced. */ close(fd); } else if (errno != ENOENT) ok = false; else { @@ -767,13 +977,14 @@ PureResult pure_store_select(const char *path, const char *key, size_t offset, if (fd < 0) return PURE_STORE_ERROR; Tree tree = {0}; Head head; + struct stat file; Entry document; uint64_t index; int64_t revision; bool found; unsigned height; PureResult result = PURE_STORE_ERROR; - if (!read_head(fd, &head) || !map_tree(fd, &head, &tree) || + if (!read_head(fd, &file, false, &head) || !map_tree(fd, &file, &head, &tree) || !find(&tree, (Entry){head.catalog, head.catalog_bytes}, key, &index, &found, &document, &revision)) goto finish; if (!found) { result = PURE_NOT_FOUND; goto finish; } @@ -838,6 +1049,16 @@ typedef struct { bool turn; /* holds this process's writer turn */ } Session; +/* Keep written zeros after the published end, where commits and their + * unpublished leftovers go. A sync of blocks the file already has writes + * only them; a new size or new blocks add a journal commit (on an SD card + * 2.4 ms against 6 ms per commit). The zeros become durable with the next + * commit's sync. */ +static void prepare_tail(Session *session) { + uint64_t size = (uint64_t)session->file.st_size, end = session->head.end; + if (size < end + TAIL_MIN) zero_fill(session->fd, size > end ? size : end, tail_end(end)); +} + static void close_session(Session *session) { release_tree(&session->tree); if (session->fd >= 0) close(session->fd); @@ -859,19 +1080,22 @@ static PureResult open_session(const char *path, Session *session) { if (!lock_writer(session->lock, false)) return errno == EACCES || errno == EAGAIN ? PURE_BUSY : PURE_STORE_ERROR; session->fd = openat(session->directory, tree_name, O_RDWR | O_NOFOLLOW | O_CLOEXEC); - if (session->fd < 0 || !read_head(session->fd, &session->head) || - !map_tree(session->fd, &session->head, &session->tree) || fstat(session->fd, &session->file)) + if (session->fd < 0 || !read_head(session->fd, &session->file, true, &session->head)) return PURE_STORE_ERROR; - /* Bytes after the published head belong to a commit nobody acknowledged. */ - if ((uint64_t)session->file.st_size > session->head.end && - ftruncate(session->fd, (off_t)session->head.end)) return PURE_STORE_ERROR; - return PURE_OK; + /* A head this process did not make durable may be the last write of a + * writer that died before its sync: make it durable before building on it. */ + if (recall(&session->file, &session->head) != DURABLE) { + if (!sync_data(session->fd)) return PURE_STORE_ERROR; + remember(&session->file, &session->head, DURABLE); + } + prepare_tail(session); + return map_tree(session->fd, &session->file, &session->head, &session->tree) ? PURE_OK : PURE_STORE_ERROR; } -/* The documents' new nodes are already appended at writer. Record each in - * the catalog (revisions come in as the base and leave as the new one), - * then make the commit durable before its head becomes visible: every - * document becomes visible at once, or none does. */ +/* The documents' new nodes are already written at writer. Record each in + * the catalog (revisions come in as the base and leave as the new one), then + * write the head and sync once: every document becomes visible at once, or + * none does. Readers of this process see the head only once it is durable. */ #ifdef PURE_STORE_TEST static unsigned long publications; unsigned long pure_tree_test_publications(void) { return publications; } @@ -900,19 +1124,21 @@ static PureResult publish(Session *session, Writer *writer, size_t count, const catalog = updated; } written = written && write_root(&session->tree, writer, &catalog, false) && flush(writer); - if (!written || fault(PURE_AFTER_HEADER) || fault(PURE_BEFORE_FILE_SYNC) || - !sync_fd(session->fd) || fault(PURE_BEFORE_RENAME)) { - if (ftruncate(session->fd, (off_t)session->head.end) == 0) sync_fd(session->fd); + /* Unpublished bytes stay in the tail; the next commit overwrites them. */ + if (!written || fault(PURE_AFTER_HEADER) || fault(PURE_BEFORE_FILE_SYNC) || fault(PURE_BEFORE_RENAME)) return PURE_STORE_ERROR; - } - Head published = {session->head.seq + 1, writer->at, catalog.ref, catalog.bytes, session->head.live}; + Head published = {session->head.seq + 1, writer->at, catalog.ref, catalog.bytes, session->head.live, + session->head.end, digest_end(&writer->digest)}; unsigned char slot[SLOT]; encode_head(slot, &published); for (size_t i = 0; i < count; ++i) ++revisions[i]; /* Past this point the head may already be visible. */ - if (!write_at(session->fd, slot, SLOT, (published.seq % 2) * SLOT) || - fault(PURE_BEFORE_DIR_SYNC) || !sync_fd(session->fd)) return PURE_UNCERTAIN; - if (published.end - DATA > 4 * published.live + COMPACT_SLACK) - compact(session->directory, session->fd, &published); + remember(&session->file, &published, PUBLISHING); + bool durable = write_at(session->fd, slot, SLOT, (published.seq % 2) * BLOCK) && + !fault(PURE_BEFORE_DIR_SYNC) && sync_data(session->fd); + remember(&session->file, &published, durable ? DURABLE : CHECKED); + if (!durable) return PURE_UNCERTAIN; + if (published.end - DATA > 4 * published.live + COMPACT_SLACK && !fstat(session->fd, &session->file)) + compact(session->directory, session->fd, &session->file, &published); return PURE_OK; } @@ -942,10 +1168,7 @@ static PureResult mutate(const char *path, const char *key, int64_t expected, if (!writer || !rebuild(&session.tree, splice ? document : (Entry){0}, splice ? offset : 0, splice ? remove : 0, body, &next)) goto finish; *writer = (Writer){.fd = session.fd, .at = session.head.end}; - if (!write_root(&session.tree, writer, &next, false)) { - if (ftruncate(session.fd, (off_t)session.head.end) == 0) sync_fd(session.fd); - goto finish; - } + if (!write_root(&session.tree, writer, &next, false)) goto finish; result = publish(&session, writer, 1, &key, &next, revision); finish: free(writer); @@ -964,8 +1187,9 @@ PureResult pure_store_keys(const char *path, bool (*visit)(void *, const char *k if (fd < 0) return PURE_STORE_ERROR; Tree tree = {0}; Head head; + struct stat file; PureResult result = PURE_STORE_ERROR; - if (read_head(fd, &head) && map_tree(fd, &head, &tree)) { + if (read_head(fd, &file, false, &head) && map_tree(fd, &file, &head, &tree)) { Entry catalog = {head.catalog, head.catalog_bytes}, document; unsigned char record[RECORD]; char key[KEY + 1] = {0}; @@ -1023,7 +1247,7 @@ static PureResult load_draft(PureDraft *draft) { uint64_t index; int64_t revision; bool found; - bool ok = !fstat(fd, &info) && read_head(fd, &head) && map_tree(fd, &head, &tree) && + bool ok = read_head(fd, &info, false, &head) && map_tree(fd, &info, &head, &tree) && find(&tree, (Entry){head.catalog, head.catalog_bytes}, draft->key, &index, &found, &document, &revision); close(fd); @@ -1069,8 +1293,9 @@ void pure_draft_close(PureDraft *draft) { free(draft); } -/* Bring the handle to the newest published version: the same file and head - * cost two system calls; a new commit maps again, a compaction reopens. */ +/* Bring the handle to the newest published version: an unchanged store costs + * one system call (the mapping shows the head slots); a new commit maps + * again, a compaction reopens. */ static PureResult refresh(PureStore *store) { struct stat info; if (fstatat(store->directory, tree_name, &info, AT_SYMLINK_NOFOLLOW) || !S_ISREG(info.st_mode)) @@ -1084,13 +1309,23 @@ static PureResult refresh(PureStore *store) { store->inode = info.st_ino; store->head = (Head){0}; } + unsigned char slots[BLOCK + SLOT]; + if (store->head.seq) { + /* A newer head lands in the other slot first; a torn or older one there + * cannot be newer. */ + const unsigned char *map = store->mapping->map; + if (decode64(map + (store->head.seq + 1) % 2 * BLOCK + 8) <= store->head.seq) return PURE_OK; + memcpy(slots, map, SLOT); + memcpy(slots + BLOCK, map + BLOCK, SLOT); + } else if (!read_at(store->fd, slots, sizeof(slots), 0)) return PURE_STORE_ERROR; Head head; - if (!read_head(store->fd, &head)) return PURE_STORE_ERROR; - if (store->mapping && head.seq == store->head.seq && head.end == store->head.end) return PURE_OK; + if (!pick(store->fd, &info, (const unsigned char *[2]){slots, slots + BLOCK}, false, &head)) + return PURE_STORE_ERROR; + if (store->mapping && head.seq == store->head.seq && head.check == store->head.check) return PURE_OK; Tree tree; Mapping *mapping = malloc(sizeof(*mapping)); if (!mapping) return PURE_ALLOCATION; - if (!map_tree(store->fd, &head, &tree)) { free(mapping); return PURE_STORE_ERROR; } + if (!map_tree(store->fd, &info, &head, &tree)) { free(mapping); return PURE_STORE_ERROR; } *mapping = (Mapping){tree.map, tree.end, 1}; release_mapping(store->mapping); store->mapping = mapping; @@ -1358,11 +1593,7 @@ static void commit_group(const char *path, Waiter *first) { bool relocate = session.file.st_dev != drafts[i]->device || session.file.st_ino != drafts[i]->inode; written = write_root(&drafts[i]->tree, writer, &roots[i], relocate); } - if (written) result = publish(&session, writer, count, keys, roots, revisions); - else { - if (ftruncate(session.fd, (off_t)session.head.end) == 0) sync_fd(session.fd); - result = PURE_STORE_ERROR; - } + result = written ? publish(&session, writer, count, keys, roots, revisions) : PURE_STORE_ERROR; } } /* Rebase while still holding the writer lock; the arenas are no longer @@ -1507,12 +1738,13 @@ bool pure_tree_test_shape(const char *path, const char *key, unsigned *height, if (directory >= 0) close(directory); Tree tree = {0}; Head head; + struct stat file; Entry document; uint64_t index; int64_t revision; bool found = false; *leaves = *small = 0; - bool ok = fd >= 0 && read_head(fd, &head) && map_tree(fd, &head, &tree) && + bool ok = fd >= 0 && read_head(fd, &file, false, &head) && map_tree(fd, &file, &head, &tree) && find(&tree, (Entry){head.catalog, head.catalog_bytes}, key, &index, &found, &document, &revision) && found && root_height(&tree, document, height) && (!document.bytes || shape(&tree, document, *height, true, leaves, small)); diff --git a/prototypes/connections/test-tree-store.c b/prototypes/connections/test-tree-store.c index 9cb0701..bdb6573 100644 --- a/prototypes/connections/test-tree-store.c +++ b/prototypes/connections/test-tree-store.c @@ -18,12 +18,17 @@ static int fail_stage = -1; static int crash; static int slow_once; /* the next publication pauses before its data sync */ +static int held_sync; /* 1: the next publication waits after its head write; 2: it is waiting */ static void pause_ms(long ms) { struct timespec wait = {ms / 1000, (ms % 1000) * 1000000}; nanosleep(&wait, NULL); } int pure_store_test_fault(PureStoreStage stage) { if (stage == PURE_BEFORE_FILE_SYNC && __atomic_exchange_n(&slow_once, 0, __ATOMIC_SEQ_CST)) pause_ms(100); + int one = 1; + if (stage == PURE_BEFORE_DIR_SYNC && + __atomic_compare_exchange_n(&held_sync, &one, 2, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) + while (__atomic_load_n(&held_sync, __ATOMIC_SEQ_CST) == 2) pause_ms(1); if ((int)stage != fail_stage) return 0; if (crash) _exit(77); return 1; @@ -33,8 +38,12 @@ bool pure_tree_test_shape(const char *path, const char *key, unsigned *height, void pure_draft_test_arena(const PureDraft *draft, size_t *nodes, size_t *bytes); unsigned long pure_tree_test_publications(void); +unsigned long pure_tree_test_syncs(void); +void pure_tree_test_forget(void); -enum { KEYS = 4, CAPACITY = 1 << 20 }; +/* The file: head slots in blocks 0 and 1 (sequence at +8, end at +16, + * catalog at +24), commits from DATA_START. */ +enum { KEYS = 4, CAPACITY = 1 << 20, SLOT_BLOCK = 4096, DATA_START = 8192 }; typedef struct { unsigned char *bytes; size_t size; int64_t revision; } Oracle; static uint32_t random_value(uint32_t *state) { @@ -60,6 +69,22 @@ static uint64_t file_size(const char *path) { return (uint64_t)info.st_size; } +/* Where the newest commit ends. */ +static uint64_t published_end(const char *path, const char *key) { + unsigned height; + uint64_t leaves, small, end; + CHECK(pure_tree_test_shape(path, key, &height, &leaves, &small, &end)); + return end; +} + +static uint64_t slot_field(int fd, unsigned slot, unsigned offset) { + unsigned char bytes[8]; + CHECK(pread(fd, bytes, 8, (off_t)(slot * SLOT_BLOCK + offset)) == 8); + uint64_t value = 0; + for (unsigned i = 0; i < 8; ++i) value |= (uint64_t)bytes[i] << (8 * i); + return value; +} + static void fill(unsigned char *bytes, size_t size, uint32_t *state) { for (size_t i = 0; i < size; ++i) bytes[i] = (unsigned char)random_value(state); } @@ -148,7 +173,7 @@ static void amplification(const char *path) { const size_t edits[][3] = {{size / 2, 0, 1}, {size / 3, 1, 0}, {size - 1, 1, 1}, {0, 0, 1}}; uint64_t worst = 0; for (size_t i = 0; i < sizeof(edits) / sizeof(*edits); ++i) { - uint64_t before = file_size(path); + uint64_t before = published_end(path, "large"); size_t offset = edits[i][0], remove = edits[i][1], count = edits[i][2]; CHECK(pure_store_splice(path, "large", o.revision, offset, remove, (PureSpan){&byte, count}, PURE_OBJECT_LIMIT, &revision) == PURE_OK); @@ -156,7 +181,7 @@ static void amplification(const char *path) { memcpy(bytes + offset, &byte, count); o.size = o.size - remove + count; o.revision = revision; - uint64_t after = file_size(path); + uint64_t after = published_end(path, "large"); if (after > before && after - before > worst) worst = after - before; expect(path, "large", &o); } @@ -176,12 +201,13 @@ static void faults(const char *path) { int64_t revision; CHECK(pure_store_put(path, "fault", 0, (PureSpan){text, 10}, &revision) == PURE_OK); Oracle o = {text, 10, revision}; - uint64_t size = file_size(path); + uint64_t size = file_size(path), end = published_end(path, "fault"); for (fail_stage = PURE_AFTER_HEADER; fail_stage <= PURE_BEFORE_RENAME; ++fail_stage) { CHECK(pure_store_splice(path, "fault", o.revision, 1, 3, (PureSpan){text, 5}, PURE_OBJECT_LIMIT, &revision) == PURE_STORE_ERROR); expect(path, "fault", &o); - CHECK(file_size(path) == size); + /* The bytes stayed unpublished, in the written zeros after the end. */ + CHECK(published_end(path, "fault") == end && file_size(path) == size); } fail_stage = PURE_BEFORE_DIR_SYNC; CHECK(pure_store_splice(path, "fault", o.revision, 10, 0, (PureSpan){text, 2}, @@ -194,16 +220,16 @@ static void faults(const char *path) { } /* A process that dies at each publication stage leaves either the previous - * or the next complete version, and its unpublished bytes are discarded. */ + * or the next complete version; the next commit overwrites its unpublished + * bytes. A head it wrote but did not sync is made durable before the next + * writer builds on it. */ static void crashes(const char *path) { unsigned char text[] = "crash-base"; int64_t revision; CHECK(pure_store_put(path, "crash", 0, (PureSpan){text, 10}, &revision) == PURE_OK); Oracle o = {text, 10, revision}; for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) { - unsigned height; - uint64_t leaves, small, before, end; - CHECK(pure_tree_test_shape(path, "crash", &height, &leaves, &small, &before)); + uint64_t before = published_end(path, "crash"); pid_t child = fork(); CHECK(child >= 0); if (!child) { @@ -217,19 +243,20 @@ static void crashes(const char *path) { CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77); if (stage == PURE_BEFORE_DIR_SYNC) { text[0] = 'C'; ++o.revision; } expect(path, "crash", &o); - uint64_t garbage = file_size(path); + uint64_t published = published_end(path, "crash"); + CHECK(stage == PURE_BEFORE_DIR_SYNC ? published > before : published == before); + unsigned long syncs = pure_tree_test_syncs(); CHECK(pure_store_splice(path, "crash", o.revision, 10, 0, (PureSpan){(const unsigned char *)"", 0}, PURE_OBJECT_LIMIT, &revision) == PURE_OK); o.revision = revision; - CHECK(pure_tree_test_shape(path, "crash", &height, &leaves, &small, &end)); - CHECK(end == file_size(path)); - /* Same-sized commit: equal growth only if the dead writer's tail was cut. */ - if (stage != PURE_BEFORE_DIR_SYNC) CHECK(garbage > before && end - before == garbage - before); + expect(path, "crash", &o); + CHECK(pure_tree_test_syncs() - syncs == (stage == PURE_BEFORE_DIR_SYNC ? 2u : 1u)); + CHECK(published_end(path, "crash") > published); } } /* A torn newest head slot exposes the previous commit, and the next writer - * overwrites that slot and the unpublished tail. */ + * overwrites that slot and the commit's bytes. */ static void torn_head(const char *path) { unsigned char first[] = "first", second[] = "second"; int64_t revision; @@ -239,12 +266,11 @@ static void torn_head(const char *path) { char name[512]; snprintf(name, sizeof(name), "%s/pure.tree", path); int fd = open(name, O_RDWR); - unsigned char slots[128]; - CHECK(fd >= 0 && pread(fd, slots, 128, 0) == 128); - uint64_t a = 0, b = 0; - for (unsigned i = 0; i < 8; ++i) { a |= (uint64_t)slots[8 + i] << (8 * i); b |= (uint64_t)slots[72 + i] << (8 * i); } - off_t newest = a > b ? 0 : 64; - unsigned char torn = slots[newest + 20] ^ 0x5a; + CHECK(fd >= 0); + off_t newest = slot_field(fd, 0, 8) > slot_field(fd, 1, 8) ? 0 : SLOT_BLOCK; + unsigned char torn; + CHECK(pread(fd, &torn, 1, newest + 20) == 1); + torn ^= 0x5a; CHECK(pwrite(fd, &torn, 1, newest + 20) == 1 && close(fd) == 0); expect(path, "torn", &o); CHECK(pure_store_put(path, "torn", o.revision + 1, (PureSpan){first, 1}, &revision) == PURE_CONFLICT); @@ -298,12 +324,12 @@ static void corruption(const char *path) { CHECK(pure_store_put(path, "damaged", 0, (PureSpan){bytes, 200000}, &revision) == PURE_OK); char name[512]; snprintf(name, sizeof(name), "%s/pure.tree", path); - uint64_t size = file_size(path); + uint64_t end = published_end(path, "damaged"); int fd = open(name, O_RDWR); CHECK(fd >= 0); unsigned errors = 0; for (unsigned i = 0; i < 3000; ++i) { - off_t at = (off_t)(128 + random_value(&state) % (size - 128)); + off_t at = (off_t)(DATA_START + random_value(&state) % (end - DATA_START)); unsigned char original, damaged; CHECK(pread(fd, &original, 1, at) == 1); damaged = original ^ (unsigned char)(1u << (random_value(&state) % 8)); @@ -345,17 +371,14 @@ static void revision_limit(const char *path) { char name[512]; snprintf(name, sizeof(name), "%s/pure.tree", path); int fd = open(name, O_RDWR); - unsigned char slots[128]; - CHECK(fd >= 0 && pread(fd, slots, 128, 0) == 128); - uint64_t seq[2] = {0, 0}, catalog = 0; - for (unsigned s = 0; s < 2; ++s) - for (unsigned i = 0; i < 8; ++i) seq[s] |= (uint64_t)slots[64 * s + 8 + i] << (8 * i); - unsigned newest = seq[0] > seq[1] ? 0 : 1; - for (unsigned i = 0; i < 8; ++i) catalog |= (uint64_t)slots[64 * newest + 24 + i] << (8 * i); + CHECK(fd >= 0); + uint64_t catalog = slot_field(fd, slot_field(fd, 0, 8) > slot_field(fd, 1, 8) ? 0 : 1, 24); CHECK(catalog >> 63); const unsigned char maximum[] = {255,255,255,255,255,255,255,127}; off_t record = (off_t)(catalog & ((UINT64_C(1) << 62) - 1)); CHECK(pwrite(fd, maximum, 8, record + 80) == 8 && close(fd) == 0); + /* The forged commit no longer matches its digest; this process wrote it + * and does not read it again, a new process would pass over it. */ PureStoredView view; CHECK(pure_store_get(path, "work", 1, &view) == PURE_OK && view.revision == INT64_MAX); pure_store_release(&view); @@ -483,15 +506,16 @@ static void typing(const char *path) { /* Reachable: 10,000 net typed bytes, the leaves they landed in and one * path. Unreachable copies are reclaimed past max(reachable, 32 KiB). */ CHECK(peak < 2 * (10000 + 4 * 4096 + 2048) + 32768 + 8192); - uint64_t before = file_size(path); + uint64_t before = published_end(path, "typed"); int64_t revision; CHECK(pure_draft_commit(draft, &revision) == PURE_OK); o.revision = revision; - CHECK(file_size(path) - before < 10000 + 16384); + uint64_t appended = published_end(path, "typed") - before; + CHECK(appended < 10000 + 16384); expect(path, "typed", &o); pure_draft_close(draft); printf("{\"typing_keystrokes\":30000,\"peak_arena_bytes\":%zu,\"commit_appended_bytes\":%llu}\n", - peak, (unsigned long long)(file_size(path) - before)); + peak, (unsigned long long)appended); free(o.bytes); free(scratch); } @@ -927,6 +951,84 @@ static void grouped(const char *path) { } } +/* A crash during a commit's only sync can leave its head on the device + * without all of its bytes. A process that starts after it passes over that + * head to the previous commit, and the next writer overwrites both. */ +static void lost_commit(const char *path) { + unsigned char first[] = "first", second[] = "second"; + int64_t revision; + CHECK(pure_store_put(path, "lost", 0, (PureSpan){first, 5}, &revision) == PURE_OK); + Oracle o = {first, 5, revision}; + uint64_t start = published_end(path, "lost"); + CHECK(pure_store_put(path, "lost", o.revision, (PureSpan){second, 6}, &revision) == PURE_OK); + uint64_t end = published_end(path, "lost"); + char name[512]; + snprintf(name, sizeof(name), "%s/pure.tree", path); + int fd = open(name, O_RDWR); + off_t at = (off_t)(start + (end - start) / 2); + unsigned char byte; + CHECK(fd >= 0 && pread(fd, &byte, 1, at) == 1); + byte ^= 0xff; + CHECK(pwrite(fd, &byte, 1, at) == 1 && close(fd) == 0); + pure_tree_test_forget(); + expect(path, "lost", &o); + PureStore *store; + PureDraft *draft; + CHECK(pure_store_open(path, &store) == PURE_OK && pure_store_draft(store, "lost", &draft) == PURE_OK); + CHECK(pure_draft_revision(draft) == o.revision); + pure_draft_close(draft); + CHECK(pure_store_put(path, "lost", o.revision + 1, (PureSpan){first, 1}, &revision) == PURE_CONFLICT); + CHECK(pure_store_put(path, "lost", o.revision, (PureSpan){second, 3}, &revision) == PURE_OK); + o = (Oracle){second, 3, revision}; + CHECK(published_end(path, "lost") < end); /* it started where the lost commit did */ + CHECK(pure_store_draft(store, "lost", &draft) == PURE_OK && pure_draft_revision(draft) == o.revision); + pure_draft_close(draft); + pure_store_close(store); + pure_tree_test_forget(); + expect(path, "lost", &o); +} + +/* While a commit's sync is pending, this process keeps reading the previous + * version: a crash could still take the new one back. */ +static void *commit_held(void *context) { + int64_t revision; + CHECK(pure_store_put(context, "pending", 1, (PureSpan){(const unsigned char *)"new", 3}, &revision) == PURE_OK); + return NULL; +} + +static void pending_head(const char *path) { + int64_t revision; + CHECK(pure_store_put(path, "pending", 0, (PureSpan){(const unsigned char *)"old", 3}, &revision) == PURE_OK); + PureStore *store; + PureDraft *draft; + CHECK(pure_store_open(path, &store) == PURE_OK); + __atomic_store_n(&held_sync, 1, __ATOMIC_SEQ_CST); + pthread_t writer; + CHECK(pthread_create(&writer, NULL, commit_held, (void *)path) == 0); + while (__atomic_load_n(&held_sync, __ATOMIC_SEQ_CST) != 2) pause_ms(1); + CHECK(pure_store_draft(store, "pending", &draft) == PURE_OK && pure_draft_revision(draft) == 1); + pure_draft_close(draft); + expect(path, "pending", &(Oracle){(unsigned char *)"old", 3, 1}); + __atomic_store_n(&held_sync, 0, __ATOMIC_SEQ_CST); + CHECK(pthread_join(writer, NULL) == 0); + CHECK(pure_store_draft(store, "pending", &draft) == PURE_OK && pure_draft_revision(draft) == 2); + pure_draft_close(draft); + pure_store_close(store); + expect(path, "pending", &(Oracle){(unsigned char *)"new", 3, 2}); +} + +/* A commit costs one sync, and commits into the written zeros after the end + * leave the file size alone (no journal commit for a new size). */ +static void one_sync(const char *path) { + int64_t revision; + CHECK(pure_store_put(path, "sync", 0, (PureSpan){(const unsigned char *)"0", 1}, &revision) == PURE_OK); + uint64_t size = file_size(path); + unsigned long syncs = pure_tree_test_syncs(); + for (unsigned i = 0; i < 20; ++i) + CHECK(pure_store_put(path, "sync", revision, (PureSpan){(const unsigned char *)"x", 1}, &revision) == PURE_OK); + CHECK(pure_tree_test_syncs() - syncs == 20 && file_size(path) == size); +} + static void clean(const char *path) { DIR *entries = opendir(path); CHECK(entries); @@ -943,7 +1045,8 @@ int main(void) { void (*checks[])(const char *) = {oracle, amplification, faults, crashes, torn_head, isolation, corruption, many_keys, revision_limit, keys, drafts, typing, draft_conflict, draft_relocation, draft_faults, - draft_versions, anonymous_draft, batch, store_handle, threads, grouped}; + draft_versions, anonymous_draft, batch, store_handle, threads, grouped, + lost_commit, pending_head, one_sync}; for (size_t i = 0; i < sizeof(checks) / sizeof(*checks); ++i) { char path[] = "/tmp/pure-tree-check-XXXXXX"; CHECK(mkdtemp(path) && pure_store_init(path) == PURE_OK); @@ -954,6 +1057,6 @@ int main(void) { puts("{\"tree_store\":\"random-oracle-balance-held-views-one-byte-amplification-faults-" "process-crash-torn-head-lockless-isolation-compaction-corruption-1024-keys-revision-limit-key-listing-" "draft-oracle-typing-arena-conflict-relocation-commit-faults-held-versions-anonymous-" - "atomic-batch-runtime-keys-store-handle-threads-group-commit\"}"); + "atomic-batch-runtime-keys-store-handle-threads-group-commit-lost-commit-pending-head-one-sync\"}"); return 0; } diff --git a/research/experiments/010-one-sync-commit/2026-09-26-ext4-linux-aarch64.json b/research/experiments/010-one-sync-commit/2026-09-26-ext4-linux-aarch64.json new file mode 100644 index 0000000..a78f512 --- /dev/null +++ b/research/experiments/010-one-sync-commit/2026-09-26-ext4-linux-aarch64.json @@ -0,0 +1,148 @@ +{ + "rounds": [ + { + "probe": [ + { + "way": "append, fsync, head, fsync (two syncs)", + "p50_ms": 9.02, + "p90_ms": 11.11, + "max_ms": 64.61 + }, + { + "way": "append, head, fdatasync", + "p50_ms": 5.75, + "p90_ms": 8.26, + "max_ms": 60.96 + }, + { + "way": "append, head, fsync", + "p50_ms": 5.69, + "p90_ms": 8.39, + "max_ms": 13.53 + }, + { + "way": "overwrite written zeros, head, fdatasync", + "p50_ms": 2.44, + "p90_ms": 5.45, + "max_ms": 67.07 + }, + { + "way": "overwrite written zeros, head, fsync", + "p50_ms": 5.7, + "p90_ms": 8.33, + "max_ms": 88.51 + } + ], + "commits": { + "two syncs": { + "commits_per_second": 103, + "p50_ms": 9.19, + "p90_ms": 11.15, + "p99_ms": 62.85 + }, + "one sync": { + "commits_per_second": 292, + "p50_ms": 2.48, + "p90_ms": 4.69, + "p99_ms": 56.77 + } + } + }, + { + "probe": [ + { + "way": "append, fsync, head, fsync (two syncs)", + "p50_ms": 9.19, + "p90_ms": 10.95, + "max_ms": 62.65 + }, + { + "way": "append, head, fdatasync", + "p50_ms": 5.82, + "p90_ms": 8.55, + "max_ms": 58.98 + }, + { + "way": "append, head, fsync", + "p50_ms": 5.66, + "p90_ms": 8.67, + "max_ms": 59.21 + }, + { + "way": "overwrite written zeros, head, fdatasync", + "p50_ms": 2.53, + "p90_ms": 6.02, + "max_ms": 61.15 + }, + { + "way": "overwrite written zeros, head, fsync", + "p50_ms": 5.75, + "p90_ms": 8.68, + "max_ms": 8.87 + } + ], + "commits": { + "one sync": { + "commits_per_second": 289, + "p50_ms": 2.44, + "p90_ms": 5.49, + "p99_ms": 53.59 + }, + "two syncs": { + "commits_per_second": 109, + "p50_ms": 8.82, + "p90_ms": 11.42, + "p99_ms": 61.67 + } + } + }, + { + "probe": [ + { + "way": "append, fsync, head, fsync (two syncs)", + "p50_ms": 8.82, + "p90_ms": 11.09, + "max_ms": 63.3 + }, + { + "way": "append, head, fdatasync", + "p50_ms": 5.61, + "p90_ms": 9.51, + "max_ms": 78.07 + }, + { + "way": "append, head, fsync", + "p50_ms": 6.21, + "p90_ms": 8.28, + "max_ms": 8.43 + }, + { + "way": "overwrite written zeros, head, fdatasync", + "p50_ms": 2.41, + "p90_ms": 4.67, + "max_ms": 58.57 + }, + { + "way": "overwrite written zeros, head, fsync", + "p50_ms": 5.07, + "p90_ms": 7.71, + "max_ms": 14.71 + } + ], + "commits": { + "two syncs": { + "commits_per_second": 94, + "p50_ms": 9.09, + "p90_ms": 12.2, + "p99_ms": 63.22 + }, + "one sync": { + "commits_per_second": 304, + "p50_ms": 2.46, + "p90_ms": 5.15, + "p99_ms": 19.8 + } + } + } + ] +} diff --git a/research/experiments/010-one-sync-commit/commits.c b/research/experiments/010-one-sync-commit/commits.c new file mode 100644 index 0000000..8bc5878 --- /dev/null +++ b/research/experiments/010-one-sync-commit/commits.c @@ -0,0 +1,47 @@ +#define _POSIX_C_SOURCE 200809L +/* Commit latency of the tree store: COMMITS sequential 16-byte inserts at + * the start of one of 64 documents, as in the HTTP put benchmark. + * commits STORE [COMMITS] -> one JSON object. */ +#include "store.h" +#include <stdio.h> +#include <stdlib.h> +#include <time.h> + +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; +} +int main(int argc, char **argv) { + if (argc < 2) return 2; + const char *path = argv[1]; + int commits = argc > 2 ? atoi(argv[2]) : 200; + if (commits < 1 || pure_store_init(path) != PURE_OK) return 1; + int64_t revision[64] = {0}; + char key[16]; + const unsigned char body[16] = "0123456789abcdef"; + for (int k = 0; k < 64; ++k) { + snprintf(key, sizeof(key), "k%d", k); + if (pure_store_put(path, key, 0, (PureSpan){body, 16}, &revision[k]) != PURE_OK) return 1; + } + double *t = malloc(sizeof(double) * (size_t)commits), start = now_ms(); + if (!t) return 1; + for (int i = 0; i < commits; ++i) { + int k = i % 64; + snprintf(key, sizeof(key), "k%d", k); + double begun = now_ms(); + if (pure_store_splice(path, key, revision[k], 0, 0, (PureSpan){body, 16}, PURE_OBJECT_LIMIT, + &revision[k]) != PURE_OK) return 1; + t[i] = now_ms() - begun; + } + double total = now_ms() - start; + qsort(t, (size_t)commits, sizeof(double), order); + printf("{\"commits_per_second\":%.0f,\"p50_ms\":%.2f,\"p90_ms\":%.2f,\"p99_ms\":%.2f}\n", commits / total * 1e3, + t[commits / 2], t[commits * 9 / 10], t[commits * 99 / 100]); + free(t); + return 0; +} diff --git a/research/experiments/010-one-sync-commit/probe.c b/research/experiments/010-one-sync-commit/probe.c new file mode 100644 index 0000000..0569192 --- /dev/null +++ b/research/experiments/010-one-sync-commit/probe.c @@ -0,0 +1,70 @@ +#define _GNU_SOURCE +/* What each way of making a small commit durable costs on one device: an + * 8 KiB commit plus a 64-byte head, 40 times per way. The ways run + * round-robin so device drift reaches all of them alike. + * probe DIRECTORY -> one JSON object per way (milliseconds). */ +#include <fcntl.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <time.h> +#include <unistd.h> + +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; +} +enum { WAYS = 5, ROUNDS = 40, REGION = 8192 }; +static const char *names[WAYS] = { + "append, fsync, head, fsync (two syncs)", + "append, head, fdatasync", + "append, head, fsync", + "overwrite written zeros, head, fdatasync", + "overwrite written zeros, head, fsync", +}; +int main(int argc, char **argv) { + const char *directory = argc > 1 ? argv[1] : "/var/tmp"; + char path[WAYS][512]; + int fd[WAYS]; + off_t end[WAYS]; + static char region[REGION], head[64]; + memset(region, 'r', sizeof(region)); + for (int w = 0; w < WAYS; ++w) { + snprintf(path[w], sizeof(path[w]), "%s/pure-sync-probe-%d", directory, w); + fd[w] = open(path[w], O_RDWR | O_CREAT | O_TRUNC, 0600); + if (fd[w] < 0) return 1; + end[w] = 4096; + if (w >= 3) { + size_t bytes = (size_t)(ROUNDS + 2) * REGION + 4096; + char *zeros = calloc(1, bytes); + if (!zeros || pwrite(fd[w], zeros, bytes, 0) != (ssize_t)bytes) return 1; + free(zeros); + } else if (pwrite(fd[w], region, 4096, 0) != 4096) return 1; + fsync(fd[w]); + } + double t[WAYS][ROUNDS]; + for (int r = 0; r < ROUNDS; ++r) + for (int w = 0; w < WAYS; ++w) { + head[0] = (char)r; + double start = now_ms(); + if (pwrite(fd[w], region, REGION, end[w]) != REGION) return 1; + end[w] += REGION; + if (w == 0) fsync(fd[w]); + if (pwrite(fd[w], head, 64, (r % 2) * 64) != 64) return 1; + if (w == 1 || w == 3) fdatasync(fd[w]); else fsync(fd[w]); + t[w][r] = now_ms() - start; + } + for (int w = 0; w < WAYS; ++w) { + qsort(t[w], ROUNDS, sizeof(double), order); + printf("{\"way\":\"%s\",\"p50_ms\":%.2f,\"p90_ms\":%.2f,\"max_ms\":%.2f}\n", names[w], t[w][ROUNDS / 2], + t[w][ROUNDS * 9 / 10], t[w][ROUNDS - 1]); + close(fd[w]); + unlink(path[w]); + } + return 0; +} diff --git a/research/experiments/010-one-sync-commit/run.py b/research/experiments/010-one-sync-commit/run.py new file mode 100644 index 0000000..9a01c07 --- /dev/null +++ b/research/experiments/010-one-sync-commit/run.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Commit cost on the store's device (see docs/research/076-one-sync-commit.md). + +Runs probe.c (the ways of making a small commit durable) and commits.c (tree +store commit latency) built against the two-sync store of --before and the +current one, alternating their order per round so device drift reaches both +alike. Prints one JSON document. +""" +import argparse +import json +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parents[2] +PURE = ROOT / 'prototypes' / 'connections' + + +def build(output, sources): + subprocess.run(['cc', '-O2', '-std=c17', f'-I{PURE}', '-o', str(output)] + [str(s) for s in sources] + ['-lpthread'], + check=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('--before', default='ad2197c', help='a revision with the two-sync commit') + parser.add_argument('--rounds', type=int, default=3) + parser.add_argument('--commits', type=int, default=200) + args = parser.parse_args() + with tempfile.TemporaryDirectory() as work: + work = Path(work) + before = work / 'store-tree-before.c' + before.write_text(subprocess.run(['git', '-C', str(ROOT), 'show', f'{args.before}:prototypes/connections/store-tree.c'], + capture_output=True, text=True, check=True).stdout) + build(work / 'probe', [HERE / 'probe.c']) + builds = {'two syncs': work / 'commits-before', 'one sync': work / 'commits-after'} + build(builds['two syncs'], [HERE / 'commits.c', before]) + build(builds['one sync'], [HERE / 'commits.c', PURE / 'store-tree.c']) + results = {'rounds': []} + for round_ in range(args.rounds): + probe = subprocess.run([str(work / 'probe'), str(args.directory)], capture_output=True, text=True, check=True) + measured = {'probe': [json.loads(line) for line in probe.stdout.splitlines()], 'commits': {}} + names = list(builds) if round_ % 2 == 0 else list(builds)[::-1] + for name in names: + store = Path(tempfile.mkdtemp(dir=args.directory, prefix='pure-commit-cost-')) + shutil.rmtree(store) + try: + run = subprocess.run([str(builds[name]), str(store), str(args.commits)], capture_output=True, text=True, + check=True) + measured['commits'][name] = json.loads(run.stdout) + finally: + shutil.rmtree(store, ignore_errors=True) + 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()) |