diff options
Diffstat (limited to 'prototypes')
| -rw-r--r-- | prototypes/connections/Makefile | 13 | ||||
| -rw-r--r-- | prototypes/connections/README.md | 10 | ||||
| -rw-r--r-- | prototypes/connections/check_tui.py | 14 | ||||
| -rw-r--r-- | prototypes/connections/store-tree.c | 971 | ||||
| -rw-r--r-- | prototypes/connections/test-splice.c | 3 | ||||
| -rw-r--r-- | prototypes/connections/test-tree-store.c | 394 |
6 files changed, 1399 insertions, 6 deletions
diff --git a/prototypes/connections/Makefile b/prototypes/connections/Makefile index 47a6530..661fbcd 100644 --- a/prototypes/connections/Makefile +++ b/prototypes/connections/Makefile @@ -3,7 +3,7 @@ CFLAGS ?= -O2 -g CFLAGS += -std=c17 -Wall -Wextra -Wpedantic -Werror BUILD ?= build -.PHONY: all core async-core editor fixed-editor fixed-server fixed-tui source-tui compiler mirror window server async-server document-peer document-server check check-policy check-policy-peer check-portable check-portable-tui check-paged-portable check-portable-peer check-native-checkpoint check-mirror check-core check-async check-async-server check-async-service-scale check-profile check-fixed-profile check-fixed-server check-service-scale check-tui-profile check-source-tui-profile check-tui check-source-tui check-release check-micro-memory check-hybrid check-flat-export check-copy-range check-peer-profile check-document-server-profile check-window-profile check-document-peer check-document-server FORCE +.PHONY: all core async-core editor fixed-editor fixed-server fixed-tui source-tui compiler mirror window server async-server document-peer document-server check check-policy check-policy-peer check-portable check-portable-tui check-paged-portable check-tree check-portable-peer check-native-checkpoint check-mirror check-core check-async check-async-server check-async-service-scale check-profile check-fixed-profile check-fixed-server check-service-scale check-tui-profile check-source-tui-profile check-tui check-source-tui check-release check-micro-memory check-hybrid check-flat-export check-copy-range check-peer-profile check-document-server-profile check-window-profile check-document-peer check-document-server FORCE CORE = runtime.c STORE ?= store.c FIXED_PROGRAM ?= examples/editor.pure @@ -115,6 +115,9 @@ $(BUILD)/test-async: test-async.c program.c $(CORE) $(HOST) $(HEADERS) | $(BUILD $(BUILD)/test-splice: test-splice.c program.c $(CORE) $(HOST) $(HEADERS) | $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_STORE_TEST test-splice.c program.c $(CORE) $(HOST) $(LDFLAGS) $(LDLIBS) -o $@ +$(BUILD)/test-tree-store: test-tree-store.c store-tree.c $(HEADERS) | $(BUILD) + $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_STORE_TEST test-tree-store.c store-tree.c $(LDFLAGS) $(LDLIBS) -o $@ + $(BUILD)/test-paged-store: test-paged-store.c store-paged.c $(HEADERS) | $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_STORE_TEST test-paged-store.c store-paged.c $(LDFLAGS) $(LDLIBS) -o $@ @@ -246,6 +249,14 @@ check-portable-tui: fixed-editor fixed-tui editor source-tui python3 check_tui.py --build $(BUILD) --portable $(if $(PORTABLE_FIXED_TUI_EVIDENCE),--output "$(PORTABLE_FIXED_TUI_EVIDENCE)") python3 check_tui.py --build $(BUILD) --source --portable $(if $(PORTABLE_SOURCE_TUI_EVIDENCE),--output "$(PORTABLE_SOURCE_TUI_EVIDENCE)") +# Selected single-file persistent-tree store: its own crash/oracle checks, then +# the unchanged runtime, splice, async and real-PTY editor checks on it. +check-tree: + $(MAKE) BUILD=build/tree STORE=store-tree.c CPPFLAGS="$(CPPFLAGS) -DPURE_STORE_TREE" check-portable build/tree/test-tree-store fixed-editor fixed-tui editor source-tui + build/tree/test-tree-store + python3 check_tui.py --build build/tree --portable --single-file-store $(if $(TREE_TUI_EVIDENCE),--output "$(TREE_TUI_EVIDENCE)") + python3 check_tui.py --build build/tree --source --portable --single-file-store + # Exercise the selected logical-page store without Linux-only size/RSS gates. # The CI runner separately asserts its real OS page size before this target. check-paged-portable: diff --git a/prototypes/connections/README.md b/prototypes/connections/README.md index 7da7eab..f0b041e 100644 --- a/prototypes/connections/README.md +++ b/prototypes/connections/README.md @@ -144,6 +144,16 @@ python3 prototypes/connections/export_flat_store.py \ [실제 큰·작은 문서 쓰기량과 전체 제한](../../docs/research/034-hybrid-document-storage.md), [기본 형식으로 내보내는 검증](../../docs/research/062-paged-store-rollback.md). +선택형 영속 트리 저장소(`store-tree.c`)는 모든 문서를 추가 전용 파일 하나에 불변 익스텐트 +트리로 둔다. 편집은 바뀐 leaf와 경로만 기록하고, 읽기는 잠그지 않는다. 기존 런타임·부분 수정· +비동기·실제 PTY 검사와 자체 고장·오라클 검사를 함께 실행한다. 형식이 달라 별도 `--store` +디렉터리를 쓴다. [설계·측정·한계](../../docs/research/073-single-file-persistent-tree.md). + +```sh +make -C prototypes/connections check-tree +prototypes/connections/build/tree/pure-edit-tui-fixed --store /tmp/pure-tree-documents note +``` + 같은 `.pure` 문서 편집기를 선택형 페이지 저장소와 함께 터미널에서 사용하려면 새 저장 디렉터리를 지정한다. 논리 페이지는 16 KiB이며 OS 페이지가 이를 나눌 때만 초기화한다. Linux 16 KiB 호스트에서 전체 선택형 검사가 통과했고, diff --git a/prototypes/connections/check_tui.py b/prototypes/connections/check_tui.py index 5328d02..adb0fef 100644 --- a/prototypes/connections/check_tui.py +++ b/prototypes/connections/check_tui.py @@ -128,7 +128,7 @@ def wait_stored(editor, store, key, expected, revision): raise AssertionError((key, expected[:100], shown(editor, store, key))) -def check(build, rss_limit_kib, source, program_path, portable=False): +def check(build, rss_limit_kib, source, program_path, portable=False, object_files=True): binary = (build / ('pure-edit-tui' if source else 'pure-edit-tui-fixed')).resolve() program = (program_path or HERE / 'examples/editor.pure').resolve() if source else None editor = [str((build / ('pure-edit' if source else 'pure-edit-fixed')).resolve())] @@ -171,9 +171,11 @@ def check(build, rss_limit_kib, source, program_path, portable=False): terminal.wait_for(b'Pure | note') terminal.type(b'\x0c') terminal.wait_for(b'Pure | documents') - (store / 'obj-second').unlink() - terminal.type(b'\x1b[B\r') - terminal.wait_for(b'Document disappeared; refresh the list') + if object_files: + # Only per-object file stores can lose one document underneath the list. + (store / 'obj-second').unlink() + terminal.type(b'\x1b[B\r') + terminal.wait_for(b'Document disappeared; refresh the list') terminal.type(b'\x1b') terminal.wait_for(b'Pure | note') terminal.type(b'!') @@ -335,12 +337,14 @@ def main(): parser.add_argument('--portable', action='store_true', help='run all PTY behavior without Linux /proc RSS measurement') parser.add_argument('--program', type=Path, help='definition for --source (including relocated bundles)') + parser.add_argument('--single-file-store', action='store_true', + help='store keeps all documents in one file (skips external deletion)') parser.add_argument('--output', type=Path) args = parser.parse_args() if args.output and args.output.exists(): parser.error('evidence file already exists') result = check(args.build, 0 if args.portable else args.rss_limit_kib, - args.source, args.program, args.portable) + args.source, args.program, args.portable, not args.single_file_store) body = json.dumps(result, indent=2) + '\n' if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) diff --git a/prototypes/connections/store-tree.c b/prototypes/connections/store-tree.c new file mode 100644 index 0000000..618028a --- /dev/null +++ b/prototypes/connections/store-tree.c @@ -0,0 +1,971 @@ +/* Selected storage experiment: every document lives in one append-only file as + * a persistent B-tree of byte extents. An edit writes only the leaves and the + * 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. */ +#define _POSIX_C_SOURCE 200809L +#ifdef __APPLE__ +#define _DARWIN_C_SOURCE +#endif +#include "store.h" +#include <errno.h> +#include <fcntl.h> +#include <inttypes.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/mman.h> +#include <sys/stat.h> +#include <unistd.h> + +enum { SLOT = 64, DATA = 2 * SLOT, LEAF_MAX = 4096, LEAF_MIN = 1024, + FAN_MAX = 32, FAN_MIN = 8, RECORD = 96, KEY = 64, HEIGHT_MAX = 12, + BUFFER = 65536 }; +#ifdef PURE_STORE_TEST +enum { COMPACT_SLACK = 65536 }; +#else +enum { COMPACT_SLACK = 1048576 }; +#endif +#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 branch_magic[4] = {'P','T','B','1'}; +static const char tree_name[] = "pure.tree"; +static uint64_t temporary_sequence; + +/* A reference is a file offset, or an index into this edit's node arena. + * bytes is the size of the whole subtree. Leaves hold raw document bytes. */ +typedef struct { uint64_t ref, bytes; } Entry; +typedef struct { Entry *items; size_t count, capacity; } List; +typedef struct { + unsigned char *bytes; /* leaf only */ + Entry *children; /* branch only */ + size_t size; /* leaf bytes or child count */ + unsigned height; +} Node; +typedef struct { + const unsigned char *map; + uint64_t end; + Node *nodes; + size_t count, capacity; +} Tree; +typedef struct { uint64_t seq, end, catalog, catalog_bytes, live; } Head; +typedef struct { + int fd; + uint64_t at; /* file offset of the next byte */ + size_t used; + bool failed; + unsigned char buffer[BUFFER]; +} Writer; +typedef struct { PureSpan part[3]; size_t index, offset; } Source; + +static bool fault(PureStoreStage stage) { +#ifdef PURE_STORE_TEST + if (pure_store_test_fault(stage)) { errno = EIO; return true; } +#else + (void)stage; +#endif + return false; +} + +static bool valid_key(const char *key) { + if (!key) return false; + size_t length = strlen(key); + if (!length || length > KEY) return false; + for (size_t i = 0; i < length; ++i) { + char b = key[i]; + if (!((b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z') || + (b >= '0' && b <= '9') || b == '-' || b == '_')) return false; + } + return true; +} + +static uint64_t decode64(const unsigned char *bytes) { + uint64_t value = 0; + for (unsigned i = 0; i < 8; ++i) value |= (uint64_t)bytes[i] << (8 * i); + return value; +} + +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 bool sync_fd(int fd) { + int rc; + do { rc = fsync(fd); } while (rc && errno == EINTR); + return rc == 0; +} + +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); + if (n < 0 && errno == EINTR) continue; + if (n <= 0) return false; + bytes += (size_t)n; + length -= (size_t)n; + offset += (uint64_t)n; + } + return true; +} + +static bool read_at(int fd, unsigned char *bytes, size_t length, uint64_t offset) { + while (length) { + ssize_t n = pread(fd, bytes, length, (off_t)offset); + if (n < 0 && errno == EINTR) continue; + if (n <= 0) return false; + bytes += (size_t)n; + length -= (size_t)n; + offset += (uint64_t)n; + } + return true; +} + +static void encode_head(unsigned char out[SLOT], const Head *head) { + memset(out, 0, SLOT); + memcpy(out, head_magic, sizeof(head_magic)); + encode64(out + 8, head->seq); + encode64(out + 16, head->end); + encode64(out + 24, head->catalog); + encode64(out + 32, head->catalog_bytes); + encode64(out + 40, head->live); + encode64(out + 56, checksum(out, 56)); +} + +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)) + 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->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; +} + +static bool map_tree(int fd, const Head *head, Tree *tree) { + struct stat info; + *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; + 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}; + return true; +} + +static void release_tree(Tree *tree) { + if (tree->map) munmap((void *)tree->map, (size_t)tree->end); + for (size_t i = 0; i < tree->count; ++i) { + free(tree->nodes[i].bytes); + free(tree->nodes[i].children); + } + free(tree->nodes); + *tree = (Tree){0}; +} + +static const unsigned char *leaf_bytes(const Tree *tree, Entry entry) { + if (!(entry.ref & LEAF_BIT) || !entry.bytes || entry.bytes > LEAF_MAX) return NULL; + uint64_t at = entry.ref & LOCATION; + if (entry.ref & MEMORY_BIT) { + if (at >= tree->count) return NULL; + const Node *node = &tree->nodes[at]; + return node->bytes && node->size == entry.bytes ? node->bytes : NULL; + } + if (at < DATA || at > tree->end || entry.bytes > tree->end - at) return NULL; + return tree->map + at; +} + +/* Returns the child count of a checked branch at height, or 0. */ +static size_t children(const Tree *tree, Entry entry, unsigned height, Entry out[FAN_MAX]) { + if ((entry.ref & LEAF_BIT) || !height) return 0; + uint64_t at = entry.ref & LOCATION; + size_t count; + if (entry.ref & MEMORY_BIT) { + if (at >= tree->count) return 0; + const Node *node = &tree->nodes[at]; + if (node->bytes || node->height != height || !node->size || node->size > FAN_MAX) return 0; + count = node->size; + memcpy(out, node->children, count * sizeof(*out)); + } else { + if (at < DATA || at % 8 || at > tree->end || tree->end - at < 8) return 0; + const unsigned char *p = tree->map + at; + count = (size_t)p[4] | (size_t)p[5] << 8; + unsigned stored = (unsigned)p[6] | (unsigned)p[7] << 8; + if (memcmp(p, branch_magic, sizeof(branch_magic)) || stored != height || !count || + count > FAN_MAX || (tree->end - at - 8) / 16 < count) return 0; + for (size_t i = 0; i < count; ++i) + out[i] = (Entry){decode64(p + 8 + 16 * i), decode64(p + 16 + 16 * i)}; + } + uint64_t sum = 0; + for (size_t i = 0; i < count; ++i) { + bool leaf = (out[i].ref & LEAF_BIT) != 0; + if (!out[i].bytes || out[i].bytes > UINT64_MAX - sum || leaf != (height == 1)) return 0; + sum += out[i].bytes; + } + return sum == entry.bytes ? count : 0; +} + +static bool root_height(const Tree *tree, Entry root, unsigned *height) { + *height = 0; + if (!root.bytes) return !root.ref; + if (root.ref & LEAF_BIT) return true; + uint64_t at = root.ref & LOCATION; + if (root.ref & MEMORY_BIT) { + if (at >= tree->count || tree->nodes[at].bytes) return false; + *height = tree->nodes[at].height; + } else { + if (at < DATA || at > tree->end || tree->end - at < 8) return false; + *height = (unsigned)tree->map[at + 6] | (unsigned)tree->map[at + 7] << 8; + } + return *height >= 1 && *height <= HEIGHT_MAX; +} + +/* Visit a range as borrowed extents: 1 done, 0 damaged, -1 stopped by visit. */ +static int visit_range(const Tree *tree, Entry entry, unsigned height, uint64_t offset, + uint64_t length, bool (*visit)(void *, PureSpan), void *context) { + if (!height) { + const unsigned char *bytes = leaf_bytes(tree, entry); + if (!bytes || offset > entry.bytes || length > entry.bytes - offset) return 0; + return visit(context, (PureSpan){bytes + offset, (size_t)length}) ? 1 : -1; + } + Entry kids[FAN_MAX]; + size_t count = children(tree, entry, height, kids); + if (!count) return 0; + uint64_t start = 0; + for (size_t i = 0; i < count && length; ++i) { + uint64_t end = start + kids[i].bytes; + if (offset < end) { + uint64_t inside = offset - start, take = kids[i].bytes - inside; + if (take > length) take = length; + int result = visit_range(tree, kids[i], height - 1, inside, take, visit, context); + if (result != 1) return result; + offset += take; + length -= take; + } + start = end; + } + return length ? 0 : 1; +} + +static bool copy_span(void *context, PureSpan span) { + unsigned char **out = context; + memcpy(*out, span.data, span.size); + *out += span.size; + return true; +} + +static bool copy_range(const Tree *tree, Entry entry, unsigned height, uint64_t offset, + uint64_t length, unsigned char *out) { + return !length || visit_range(tree, entry, height, offset, length, copy_span, &out) == 1; +} + +/* The bytes of a range held by one leaf, or NULL when it spans leaves. */ +static const unsigned char *single_leaf(const Tree *tree, Entry entry, unsigned height, + uint64_t offset, uint64_t length) { + while (height) { + Entry kids[FAN_MAX]; + size_t count = children(tree, entry, height, kids), i = 0; + uint64_t start = 0; + if (!count) return NULL; + while (i + 1 < count && offset >= start + kids[i].bytes) start += kids[i++].bytes; + if (offset - start > kids[i].bytes || length > kids[i].bytes - (offset - start)) return NULL; + entry = kids[i]; + offset -= start; + --height; + } + const unsigned char *bytes = leaf_bytes(tree, entry); + return bytes && offset <= entry.bytes && length <= entry.bytes - offset ? bytes + offset : NULL; +} + +static bool push(List *list, Entry entry) { + if (list->count == list->capacity) { + size_t capacity = list->capacity ? 2 * list->capacity : 8; + Entry *items = realloc(list->items, capacity * sizeof(*items)); + if (!items) return false; + list->items = items; + list->capacity = capacity; + } + list->items[list->count++] = entry; + return true; +} + +static bool add_node(Tree *tree, Node node, uint64_t *index) { + if (tree->count == tree->capacity) { + size_t capacity = tree->capacity ? 2 * tree->capacity : 64; + Node *nodes = realloc(tree->nodes, capacity * sizeof(*nodes)); + if (!nodes) return false; + tree->nodes = nodes; + tree->capacity = capacity; + } + tree->nodes[tree->count] = node; + *index = tree->count++; + return true; +} + +static void take(Source *source, unsigned char *out, size_t length) { + while (length) { + PureSpan part = source->part[source->index]; + size_t left = part.size - source->offset; + if (!left) { ++source->index; source->offset = 0; continue; } + size_t n = left < length ? left : length; + memcpy(out, part.data + source->offset, n); + out += n; + length -= n; + source->offset += n; + } +} + +/* Balanced leaves: every leaf of a multi-leaf result holds >= LEAF_MAX / 2. */ +static bool chunk_bytes(Tree *tree, Source *source, uint64_t total, List *out) { + uint64_t pieces = (total + LEAF_MAX - 1) / LEAF_MAX; + for (uint64_t i = 0; i < pieces; ++i) { + size_t size = (size_t)(total / pieces + (i < total % pieces)); + unsigned char *bytes = malloc(size); + uint64_t index; + if (!bytes) return false; + take(source, bytes, size); + if (!add_node(tree, (Node){.bytes = bytes, .size = size}, &index)) { free(bytes); return false; } + if (!push(out, (Entry){LEAF_BIT | MEMORY_BIT | index, size})) return false; + } + return true; +} + +static bool chunk_entries(Tree *tree, const Entry *items, size_t count, unsigned height, List *out) { + size_t pieces = (count + FAN_MAX - 1) / FAN_MAX; + for (size_t i = 0, at = 0; i < pieces; ++i) { + size_t size = count / pieces + (i < count % pieces); + Entry *kids = malloc(size * sizeof(*kids)); + uint64_t sum = 0, index; + if (!kids) return false; + memcpy(kids, items + at, size * sizeof(*kids)); + for (size_t k = 0; k < size; ++k) sum += kids[k].bytes; + if (!add_node(tree, (Node){.children = kids, .size = size, .height = height}, &index)) { + free(kids); + return false; + } + if (!push(out, (Entry){MEMORY_BIT | index, sum})) return false; + at += size; + } + return true; +} + +/* Only nodes created by this edit can be underfull; stored siblings are left as they are. */ +static bool underfull(const Tree *tree, Entry entry, unsigned height) { + if (!(entry.ref & MEMORY_BIT)) return false; + return height ? tree->nodes[entry.ref & LOCATION].size < FAN_MIN : entry.bytes < LEAF_MIN; +} + +static bool replace(List *list, size_t at, size_t removed, const List *with) { + size_t count = list->count - removed + with->count; + if (count > list->capacity) { + Entry *items = realloc(list->items, count * sizeof(*items)); + if (!items) return false; + list->items = items; + list->capacity = count; + } + memmove(list->items + at + with->count, list->items + at + removed, + (list->count - at - removed) * sizeof(*list->items)); + memcpy(list->items + at, with->items, with->count * sizeof(*list->items)); + list->count = count; + return true; +} + +static bool merge(Tree *tree, Entry a, Entry b, unsigned height, List *out); + +/* Merge each new underfull node with a neighbour. A merge removes a node or + * yields two nodes of at least half capacity, so the loop terminates. */ +static bool repair(Tree *tree, List *all, unsigned height) { + bool ok = true; + for (size_t i = 0; ok && all->count > 1 && i < all->count;) { + if (!underfull(tree, all->items[i], height)) { ++i; continue; } + size_t at = i + 1 < all->count ? i : i - 1; + List merged = {0}; + ok = merge(tree, all->items[at], all->items[at + 1], height, &merged) && + replace(all, at, 2, &merged); + free(merged.items); + i = at; + } + return ok; +} + +static bool merge(Tree *tree, Entry a, Entry b, unsigned height, List *out) { + if (!height) { + const unsigned char *x = leaf_bytes(tree, a), *y = leaf_bytes(tree, b); + if (!x || !y) return false; + Source source = {.part = {{x, (size_t)a.bytes}, {y, (size_t)b.bytes}}}; + return chunk_bytes(tree, &source, a.bytes + b.bytes, out); + } + Entry kids[FAN_MAX]; + List all = {0}; + size_t count = children(tree, a, height, kids); + bool ok = count != 0; + for (size_t i = 0; ok && i < count; ++i) ok = push(&all, kids[i]); + count = ok ? children(tree, b, height, kids) : 0; + ok = count != 0; + for (size_t i = 0; ok && i < count; ++i) ok = push(&all, kids[i]); + /* A small child carried in from a one-child parent is repaired here too. */ + ok = ok && repair(tree, &all, height - 1) && chunk_entries(tree, all.items, all.count, height, out); + free(all.items); + return ok; +} + +/* Replace [offset, offset + remove) of entry with insert. Appends the + * replacement nodes, all at entry's height, to out (possibly none). */ +static bool edit(Tree *tree, Entry entry, unsigned height, uint64_t offset, uint64_t remove, + PureSpan insert, List *out) { + if (offset > entry.bytes || remove > entry.bytes - offset) return false; + if (!height) { + const unsigned char *bytes = leaf_bytes(tree, entry); + if (!bytes) return false; + Source source = {.part = {{bytes, (size_t)offset}, insert, + {bytes + offset + remove, (size_t)(entry.bytes - offset - remove)}}}; + return chunk_bytes(tree, &source, entry.bytes - remove + insert.size, out); + } + Entry kids[FAN_MAX]; + size_t count = children(tree, entry, height, kids), first = count, final = count; + if (!count) return false; + uint64_t last = remove ? offset + remove - 1 : offset, start = 0, first_start = 0, final_start = 0; + for (size_t i = 0; i < count; ++i) { + uint64_t end = start + kids[i].bytes; + bool tail = i + 1 == count; + if (first == count && (offset < end || tail)) { first = i; first_start = start; } + if (first != count && (last < end || tail)) { final = i; final_start = start; break; } + start = end; + } + List middle = {0}, all = {0}; + bool ok = first == final ? + edit(tree, kids[first], height - 1, offset - first_start, remove, insert, &middle) : + edit(tree, kids[first], height - 1, offset - first_start, + first_start + kids[first].bytes - offset, insert, &middle) && + edit(tree, kids[final], height - 1, 0, offset + remove - final_start, (PureSpan){0}, &middle); + for (size_t i = 0; ok && i < first; ++i) ok = push(&all, kids[i]); + for (size_t i = 0; ok && i < middle.count; ++i) ok = push(&all, middle.items[i]); + for (size_t i = final + 1; ok && i < count; ++i) ok = push(&all, kids[i]); + free(middle.items); + ok = ok && repair(tree, &all, height - 1) && chunk_entries(tree, all.items, all.count, height, out); + free(all.items); + return ok; +} + +/* The new root of root with [offset, offset + remove) replaced by insert. */ +static bool rebuild(Tree *tree, Entry root, uint64_t offset, uint64_t remove, + PureSpan insert, Entry *out) { + unsigned height; + List list = {0}; + bool ok = root_height(tree, root, &height); + if (ok && !root.bytes) { + Source source = {.part = {insert}}; + ok = !offset && !remove && chunk_bytes(tree, &source, insert.size, &list); + } else if (ok) ok = edit(tree, root, height, offset, remove, insert, &list); + while (ok && list.count > 1) { + List up = {0}; + ok = height < HEIGHT_MAX && chunk_entries(tree, list.items, list.count, ++height, &up); + free(list.items); + list = up; + } + Entry result = ok && list.count ? list.items[0] : (Entry){0}; + free(list.items); + while (ok && height && result.bytes) { + Entry kids[FAN_MAX]; + size_t count = children(tree, result, height, kids); + if (count != 1) { ok = count > 1; break; } + result = kids[0]; + --height; + } + *out = result; + return ok; +} + +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; + writer->used = 0; + return !writer->failed; +} + +static void put_bytes(Writer *writer, const unsigned char *bytes, size_t length) { + while (length && !writer->failed) { + if (writer->used == BUFFER && !flush(writer)) return; + size_t n = BUFFER - writer->used < length ? BUFFER - writer->used : length; + memcpy(writer->buffer + writer->used, bytes, n); + writer->used += n; + writer->at += n; + bytes += n; + length -= n; + } +} + +static uint64_t put_branch(Writer *writer, const Entry *kids, size_t count, unsigned height) { + static const unsigned char zero[8]; + put_bytes(writer, zero, (size_t)((8 - writer->at % 8) % 8)); + uint64_t at = writer->at; + unsigned char header[8]; + memcpy(header, branch_magic, sizeof(branch_magic)); + header[4] = (unsigned char)count; + header[5] = (unsigned char)(count >> 8); + header[6] = (unsigned char)height; + header[7] = (unsigned char)(height >> 8); + put_bytes(writer, header, sizeof(header)); + for (size_t i = 0; i < count; ++i) { + unsigned char pair[16]; + encode64(pair, kids[i].ref); + encode64(pair + 8, kids[i].bytes); + put_bytes(writer, pair, sizeof(pair)); + } + return at; +} + +/* Append a subtree's arena nodes, children first, and set entry to the + * stored reference without changing the arena. With relocate, stored nodes + * are copied too; compaction uses this to rewrite reachable nodes. */ +static bool write_subtree(const Tree *tree, Writer *writer, Entry *entry, unsigned height, + bool relocate) { + if (!entry->bytes || (!(entry->ref & MEMORY_BIT) && !relocate)) return true; + if (!height) { + const unsigned char *bytes = leaf_bytes(tree, *entry); + if (!bytes) return false; + entry->ref = LEAF_BIT | writer->at; + put_bytes(writer, bytes, (size_t)entry->bytes); + return !writer->failed; + } + Entry kids[FAN_MAX]; + size_t count = children(tree, *entry, height, kids); + if (!count) return false; + for (size_t i = 0; i < count; ++i) + if (!write_subtree(tree, writer, &kids[i], height - 1, relocate)) return false; + entry->ref = put_branch(writer, kids, count, height); + return !writer->failed; +} + +static bool write_root(const Tree *tree, Writer *writer, Entry *root, bool relocate) { + unsigned height; + return root_height(tree, *root, &height) && write_subtree(tree, writer, root, height, relocate); +} + +static bool read_record(const Tree *tree, Entry catalog, uint64_t index, + unsigned char record[RECORD], Entry *document, int64_t *revision) { + unsigned height; + if (!root_height(tree, catalog, &height) || + !copy_range(tree, catalog, height, index * RECORD, RECORD, record)) return false; + *document = (Entry){decode64(record + KEY), decode64(record + KEY + 8)}; + uint64_t stored = decode64(record + KEY + 16); + *revision = (int64_t)stored; + return stored && stored <= (uint64_t)INT64_MAX && !document->ref == !document->bytes && + document->bytes <= PURE_OBJECT_LIMIT; +} + +/* Binary search over records ordered by their zero-padded key bytes. */ +static bool find(const Tree *tree, Entry catalog, const char *key, uint64_t *index, + bool *found, Entry *document, int64_t *revision) { + unsigned char wanted[KEY] = {0}, record[RECORD]; + memcpy(wanted, key, strlen(key)); + uint64_t low = 0, high = catalog.bytes / RECORD; + *found = false; + *document = (Entry){0}; + *revision = 0; + while (low < high) { + uint64_t middle = low + (high - low) / 2; + if (!read_record(tree, catalog, middle, record, document, revision)) return false; + int order = memcmp(wanted, record, KEY); + if (!order) { *index = middle; *found = true; return true; } + if (order < 0) high = middle; else low = middle + 1; + } + *document = (Entry){0}; + *revision = 0; + *index = low; + return true; +} + +static int create_temporary(int directory, char name[96]) { + for (unsigned attempt = 0; attempt < 64; ++attempt) { + if (temporary_sequence == UINT64_MAX) break; + snprintf(name, 96, ".pending-%ld-%" PRIu64, (long)getpid(), ++temporary_sequence); + int fd = openat(directory, name, O_RDWR | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0600); + if (fd >= 0) return fd; + if (errno != EEXIST) break; + } + name[0] = 0; /* Never remove an existing file on a name collision. */ + 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); + if (close(fd)) ok = false; + ok = ok && !renameat(directory, temporary, directory, tree_name); + if (ok) temporary[0] = 0; + return ok && sync_fd(directory); +} + +/* 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) { + Tree tree; + if (!map_tree(source, head, &tree)) return; + char temporary[96]; + int fd = create_temporary(directory, temporary); + Writer *writer = fd >= 0 ? malloc(sizeof(*writer)) : NULL; + uint64_t count = head->catalog_bytes / RECORD; + unsigned char *records = count ? malloc((size_t)(count * RECORD)) : NULL; + Entry catalog = {head->catalog, head->catalog_bytes}; + unsigned height; + bool ok = writer && (!count || records) && root_height(&tree, catalog, &height) && + (!count || copy_range(&tree, catalog, height, 0, count * RECORD, records)); + if (writer) *writer = (Writer){.fd = fd, .at = DATA}; + for (uint64_t i = 0; ok && i < count; ++i) { + unsigned char *record = records + i * RECORD; + Entry document = {decode64(record + KEY), decode64(record + KEY + 8)}; + unsigned document_height; + ok = root_height(&tree, document, &document_height) && + write_subtree(&tree, writer, &document, document_height, true); + encode64(record + KEY, document.ref); + } + Entry fresh = {0}; + 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}; + ok = install(directory, temporary, fd, &compacted); + fd = -1; + } + if (fd >= 0) close(fd); + if (*temporary) unlinkat(directory, temporary, 0); + free(records); + free(writer); + release_tree(&tree); +} + +static bool lock_writer(int lock, bool wait) { + struct flock hold = {.l_type = F_WRLCK, .l_whence = SEEK_SET, .l_start = 0, .l_len = 0}; + int rc; + do { rc = fcntl(lock, wait ? F_SETLKW : F_SETLK, &hold); } while (rc && wait && errno == EINTR); + return rc == 0; +} + +PureResult pure_store_init(const char *path) { + if (!path || !*path) return PURE_STORE_ERROR; + bool created = mkdir(path, 0700) == 0; + if (!created && errno != EEXIST) return PURE_STORE_ERROR; + int directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (directory < 0) return PURE_STORE_ERROR; /* Never reinterpret an old DB file. */ + int lock = openat(directory, ".writer.lock", O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600); + struct stat info; + bool ok = lock >= 0 && fstat(lock, &info) == 0 && S_ISREG(info.st_mode) && lock_writer(lock, true); + if (ok) { + Head head; + 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. */ + close(fd); + } else if (errno != ENOENT) ok = false; + else { + char temporary[96]; + int created_fd = create_temporary(directory, temporary); + head = (Head){.seq = 1, .end = DATA}; + ok = created_fd >= 0 && install(directory, temporary, created_fd, &head); + if (*temporary) unlinkat(directory, temporary, 0); + } + } + ok = ok && sync_fd(lock) && sync_fd(directory); + if (lock >= 0) close(lock); + close(directory); + if (ok && created) { + char *parent = strdup(path); + if (!parent) return PURE_STORE_ERROR; + size_t length = strlen(parent); + while (length > 1 && parent[length - 1] == '/') parent[--length] = 0; + char *slash = strrchr(parent, '/'); + if (!slash) strcpy(parent, "."); + else if (slash == parent) parent[1] = 0; + else *slash = 0; + int fd = open(parent, O_RDONLY | O_DIRECTORY | O_CLOEXEC); + free(parent); + ok = fd >= 0 && sync_fd(fd); + if (fd >= 0) close(fd); + } + return ok ? PURE_OK : PURE_STORE_ERROR; +} + +void pure_store_release(PureStoredView *view) { + if (view->mapped_size) munmap(view->mapping, view->mapped_size); + *view = (PureStoredView){0}; +} + +PureResult pure_store_select(const char *path, const char *key, size_t offset, + size_t length, bool whole, size_t minimum, + size_t available, PureStoredView *out) { + *out = (PureStoredView){0}; + if (!valid_key(key)) return PURE_STORE_ERROR; + int directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (directory < 0) return PURE_STORE_ERROR; + int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC); + close(directory); + if (fd < 0) return PURE_STORE_ERROR; + Tree tree = {0}; + Head head; + 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) || + !find(&tree, (Entry){head.catalog, head.catalog_bytes}, key, &index, &found, + &document, &revision)) goto finish; + if (!found) { result = PURE_NOT_FOUND; goto finish; } + size_t size = (size_t)document.bytes; + if (size < minimum || offset > size) { result = PURE_LIMIT; goto finish; } + if (whole) length = size - offset; + if (length > size - offset) { result = PURE_LIMIT; goto finish; } + if (length > available) { result = PURE_BUDGET; goto finish; } + if (length) { + if (!root_height(&tree, document, &height)) goto finish; + const unsigned char *direct = single_leaf(&tree, document, height, offset, length); + void *mapping; + size_t mapped_size, prefix = 0; + if (direct) { + /* One extent: map it in place, as the flat store does. */ + long page = sysconf(_SC_PAGESIZE); + if (page <= 0) goto finish; + size_t start = (size_t)(direct - tree.map), aligned = start - start % (size_t)page; + prefix = start - aligned; + mapped_size = prefix + length; + mapping = mmap(NULL, mapped_size, PROT_READ, MAP_PRIVATE, fd, (off_t)aligned); + if (mapping == MAP_FAILED) goto finish; + } else { + /* A contiguous view of several extents is assembled for this API. */ + mapped_size = length; + mapping = mmap(NULL, mapped_size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (mapping == MAP_FAILED) goto finish; + if (!copy_range(&tree, document, height, offset, length, mapping) || + mprotect(mapping, mapped_size, PROT_READ)) { + munmap(mapping, mapped_size); + goto finish; + } + } + out->mapping = mapping; + out->mapped_size = mapped_size; + out->body = (PureSpan){(const unsigned char *)mapping + prefix, length}; + } + out->revision = revision; + out->object_size = size; + result = PURE_OK; +finish: + release_tree(&tree); + close(fd); + return result; +} + +PureResult pure_store_get(const char *path, const char *key, size_t available, PureStoredView *out) { + return pure_store_select(path, key, 0, 0, true, 0, available, out); +} + +PureResult pure_store_get_range(const char *path, const char *key, size_t offset, + size_t length, size_t available, PureStoredView *out) { + return pure_store_select(path, key, offset, length, false, 0, available, out); +} + +/* A locked writer over the current published file. */ +typedef struct { + int directory, lock, fd; + Head head; + Tree tree; + struct stat file; +} Session; + +static void close_session(Session *session) { + release_tree(&session->tree); + if (session->fd >= 0) close(session->fd); + if (session->lock >= 0) close(session->lock); + if (session->directory >= 0) close(session->directory); +} + +static PureResult open_session(const char *path, Session *session) { + *session = (Session){.directory = -1, .lock = -1, .fd = -1}; + session->directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (session->directory < 0) return PURE_STORE_ERROR; + session->lock = openat(session->directory, ".writer.lock", O_RDWR | O_NOFOLLOW | O_CLOEXEC); + struct stat info; + if (session->lock < 0 || fstat(session->lock, &info) || !S_ISREG(info.st_mode)) return PURE_STORE_ERROR; + 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)) + 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; +} + +/* The document's new nodes are already appended at writer. Record it in the + * catalog, then make the commit durable before its head becomes visible. */ +static PureResult publish(Session *session, Writer *writer, const char *key, uint64_t index, + bool found, Entry document, int64_t *revision) { + unsigned char record[RECORD] = {0}; + Entry catalog = {session->head.catalog, session->head.catalog_bytes}, updated; + memcpy(record, key, strlen(key)); + encode64(record + KEY, document.ref); + encode64(record + KEY + 8, document.bytes); + encode64(record + KEY + 16, (uint64_t)*revision + 1); + bool written = rebuild(&session->tree, catalog, index * RECORD, found ? RECORD : 0, + (PureSpan){record, RECORD}, &updated) && + write_root(&session->tree, writer, &updated, 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); + return PURE_STORE_ERROR; + } + Head published = {session->head.seq + 1, writer->at, updated.ref, updated.bytes, session->head.live}; + unsigned char slot[SLOT]; + encode_head(slot, &published); + ++*revision; /* 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); + return PURE_OK; +} + +static PureResult mutate(const char *path, const char *key, int64_t expected, + PureSpan body, bool splice, size_t offset, size_t remove, + size_t max_size, int64_t *revision) { + *revision = 0; + if (!valid_key(key) || (body.size && !body.data)) return PURE_STORE_ERROR; + if (body.size > PURE_OBJECT_LIMIT || max_size > PURE_OBJECT_LIMIT || expected < 0) return PURE_LIMIT; + Session session; + Writer *writer = NULL; + Entry document, next; + uint64_t index; + bool found; + PureResult result = open_session(path, &session); + if (result != PURE_OK) goto finish; + result = PURE_STORE_ERROR; + if (!find(&session.tree, (Entry){session.head.catalog, session.head.catalog_bytes}, key, + &index, &found, &document, revision)) goto finish; + if (*revision != expected) { result = PURE_CONFLICT; goto finish; } + if (*revision == INT64_MAX) { result = PURE_LIMIT; goto finish; } + uint64_t old_size = document.bytes; + if (splice && (offset > old_size || remove > old_size - offset || + old_size - remove > max_size || + body.size > max_size - (old_size - remove))) { result = PURE_LIMIT; goto finish; } + writer = malloc(sizeof(*writer)); + 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; + } + result = publish(&session, writer, key, index, found, next, revision); +finish: + free(writer); + close_session(&session); + return result; +} + +/* Keys come from the catalog in ascending byte order. */ +PureResult pure_store_keys(const char *path, bool (*visit)(void *, const char *key), + void *context) { + int directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (directory < 0) return PURE_STORE_ERROR; + int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC); + close(directory); + if (fd < 0) return PURE_STORE_ERROR; + Tree tree = {0}; + Head head; + PureResult result = PURE_STORE_ERROR; + if (read_head(fd, &head) && map_tree(fd, &head, &tree)) { + Entry catalog = {head.catalog, head.catalog_bytes}, document; + unsigned char record[RECORD]; + char key[KEY + 1] = {0}; + int64_t revision; + result = PURE_OK; + for (uint64_t i = 0; i < catalog.bytes / RECORD; ++i) { + if (!read_record(&tree, catalog, i, record, &document, &revision)) { result = PURE_STORE_ERROR; break; } + memcpy(key, record, KEY); + if (!valid_key(key)) { result = PURE_STORE_ERROR; break; } + if (!visit(context, key)) break; + } + } + release_tree(&tree); + close(fd); + return result; +} + +PureResult pure_store_put(const char *path, const char *key, int64_t expected, + PureSpan body, int64_t *revision) { + return mutate(path, key, expected, body, false, 0, 0, PURE_OBJECT_LIMIT, revision); +} + +PureResult pure_store_splice(const char *path, const char *key, int64_t expected, + size_t offset, size_t remove, PureSpan insert, + size_t max_size, int64_t *revision) { + return mutate(path, key, expected, insert, true, offset, remove, max_size, revision); +} + +#ifdef PURE_STORE_TEST +static bool shape(const Tree *tree, Entry entry, unsigned height, bool root, + uint64_t *leaves, uint64_t *small) { + if (!height) { + ++*leaves; + if (!root && entry.bytes < LEAF_MIN) ++*small; + return leaf_bytes(tree, entry) != NULL; + } + Entry kids[FAN_MAX]; + size_t count = children(tree, entry, height, kids); + if (!count || (root && count < 2)) return false; + if (!root && count < FAN_MIN) ++*small; + for (size_t i = 0; i < count; ++i) + if (!shape(tree, kids[i], height - 1, false, leaves, small)) return false; + return true; +} + +/* Test-only: the stored height, leaf count and number of underfull nodes. */ +bool pure_tree_test_shape(const char *path, const char *key, unsigned *height, + uint64_t *leaves, uint64_t *small, uint64_t *file_end) { + int directory = open(path, O_RDONLY | O_DIRECTORY | O_CLOEXEC); + int fd = directory >= 0 ? openat(directory, tree_name, O_RDONLY | O_CLOEXEC) : -1; + if (directory >= 0) close(directory); + Tree tree = {0}; + Head head; + 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) && + 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)); + if (ok) *file_end = head.end; + release_tree(&tree); + if (fd >= 0) close(fd); + return ok; +} +#endif diff --git a/prototypes/connections/test-splice.c b/prototypes/connections/test-splice.c index bf3c0a9..51970d3 100644 --- a/prototypes/connections/test-splice.c +++ b/prototypes/connections/test-splice.c @@ -126,12 +126,15 @@ static void storage(const char *path) { expect(path, "work", revision, (PureSpan){oracle, length}); CHECK(pure_store_splice(path, "work", revision, 0, length, (PureSpan){0}, PURE_OBJECT_LIMIT, &revision) == PURE_OK); expect(path, "work", revision, (PureSpan){0}); +#ifndef PURE_STORE_TREE + /* Per-object file formats; test-tree-store covers the same limit. */ char file[256]; snprintf(file, sizeof(file), "%s/obj-work", path); int fd = open(file, O_WRONLY); const unsigned char maximum[] = {255,255,255,255,255,255,255,127}; CHECK(fd >= 0 && pwrite(fd, maximum, 8, 8) == 8 && close(fd) == 0); CHECK(pure_store_splice(path, "work", INT64_MAX, 0, 0, (PureSpan){0}, PURE_OBJECT_LIMIT, &revision) == PURE_LIMIT); +#endif } static void runtime(const char *path) { diff --git a/prototypes/connections/test-tree-store.c b/prototypes/connections/test-tree-store.c new file mode 100644 index 0000000..3a1e8f0 --- /dev/null +++ b/prototypes/connections/test-tree-store.c @@ -0,0 +1,394 @@ +#define _POSIX_C_SOURCE 200809L +#include "store.h" +#include <assert.h> +#include <dirent.h> +#include <fcntl.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/stat.h> +#include <sys/wait.h> +#include <unistd.h> + +#define CHECK(x) do { if (!(x)) { fprintf(stderr, "%s:%d: %s\n", __FILE__, __LINE__, #x); abort(); } } while (0) + +static int fail_stage = -1; +static int crash; +int pure_store_test_fault(PureStoreStage stage) { + if ((int)stage != fail_stage) return 0; + if (crash) _exit(77); + return 1; +} +bool pure_tree_test_shape(const char *path, const char *key, unsigned *height, + uint64_t *leaves, uint64_t *small, uint64_t *file_end); + +enum { KEYS = 4, CAPACITY = 1 << 20 }; +typedef struct { unsigned char *bytes; size_t size; int64_t revision; } Oracle; + +static uint32_t random_value(uint32_t *state) { + *state ^= *state << 13; + *state ^= *state >> 17; + *state ^= *state << 5; + return *state; +} + +static void expect(const char *path, const char *key, const Oracle *o) { + PureStoredView view; + CHECK(pure_store_get(path, key, PURE_OBJECT_LIMIT, &view) == PURE_OK); + CHECK(view.revision == o->revision && view.body.size == o->size); + CHECK(!o->size || !memcmp(view.body.data, o->bytes, o->size)); + pure_store_release(&view); +} + +static uint64_t file_size(const char *path) { + char name[512]; + snprintf(name, sizeof(name), "%s/pure.tree", path); + struct stat info; + CHECK(stat(name, &info) == 0); + return (uint64_t)info.st_size; +} + +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); +} + +/* Random edits against a flat oracle, including views held across later + * commits and compactions. Every stored node must stay balanced. */ +static void oracle(const char *path) { + static const char *keys[KEYS] = {"alpha", "b", "gamma-long_key-0123456789", "z"}; + Oracle docs[KEYS] = {{0}}; + unsigned char *insert = malloc(65536); + for (unsigned k = 0; k < KEYS; ++k) CHECK((docs[k].bytes = malloc(CAPACITY))); + CHECK(insert); + uint32_t state = 0x9e3779b9u; + PureStoredView held = {0}; + unsigned char *held_copy = malloc(CAPACITY); + size_t held_size = 0; + unsigned max_height = 0; + size_t max_size = 0, splices = 0; + for (unsigned step = 0; step < 6000; ++step) { + unsigned k = random_value(&state) % KEYS; + Oracle *o = &docs[k]; + uint32_t kind = random_value(&state) % 100; + size_t count = kind < 70 ? random_value(&state) % 24 : + kind < 95 ? random_value(&state) % 6000 : random_value(&state) % 65536; + int64_t revision; + fill(insert, count, &state); + if (kind >= 98 || !o->revision) { + CHECK(pure_store_put(path, keys[k], o->revision, (PureSpan){insert, count}, &revision) == PURE_OK); + memcpy(o->bytes, insert, count); + o->size = count; + } else { + size_t offset = o->size ? random_value(&state) % (o->size + 1) : 0; + size_t remove = o->size - offset ? random_value(&state) % (o->size - offset + 1) : 0; + if (random_value(&state) % 24) remove = remove % 64; + if (o->size - remove + count > CAPACITY) { count = 0; remove = o->size - offset; } + CHECK(pure_store_splice(path, keys[k], o->revision, offset, remove, + (PureSpan){insert, count}, PURE_OBJECT_LIMIT, &revision) == PURE_OK); + memmove(o->bytes + offset + count, o->bytes + offset + remove, o->size - offset - remove); + memcpy(o->bytes + offset, insert, count); + o->size = o->size - remove + count; + ++splices; + } + if (o->size > max_size) max_size = o->size; + CHECK(revision == o->revision + 1); + o->revision = revision; + expect(path, keys[k], o); + unsigned height; + uint64_t leaves, small, end; + CHECK(pure_tree_test_shape(path, keys[k], &height, &leaves, &small, &end)); + CHECK(small == 0); + if (height > max_height) max_height = height; + if (step % 97 == 0) { + if (held.mapped_size) { + CHECK(held.body.size == held_size && !memcmp(held.body.data, held_copy, held_size)); + pure_store_release(&held); + } + CHECK(pure_store_get(path, keys[k], PURE_OBJECT_LIMIT, &held) == PURE_OK); + held_size = held.body.size; + if (held_size) memcpy(held_copy, held.body.data, held_size); + } + } + if (held.mapped_size) { + CHECK(held.body.size == held_size && !memcmp(held.body.data, held_copy, held_size)); + pure_store_release(&held); + } + for (unsigned k = 0; k < KEYS; ++k) { expect(path, keys[k], &docs[k]); free(docs[k].bytes); } + CHECK(max_height >= 2 && max_height <= 4 && max_size > 262144); + printf("{\"oracle_steps\":6000,\"splices\":%zu,\"max_document_bytes\":%zu,\"max_height\":%u}\n", + splices, max_size, max_height); + free(insert); + free(held_copy); +} + +/* A one-byte insert, delete or replacement in a 1 MiB document appends only + * its changed leaf and path, not another copy of the document. */ +static void amplification(const char *path) { + size_t size = 1048576; + unsigned char *bytes = malloc(size + 1); + uint32_t state = 7; + CHECK(bytes); + fill(bytes, size, &state); + int64_t revision; + CHECK(pure_store_put(path, "large", 0, (PureSpan){bytes, size}, &revision) == PURE_OK); + Oracle o = {bytes, size, revision}; + unsigned char byte = 'Q'; + 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); + 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); + memmove(bytes + offset + count, bytes + offset + remove, o.size - offset - remove); + memcpy(bytes + offset, &byte, count); + o.size = o.size - remove + count; + o.revision = revision; + uint64_t after = file_size(path); + if (after > before && after - before > worst) worst = after - before; + expect(path, "large", &o); + } + /* Compaction may shrink the file; growth per edit is the claim. */ + CHECK(worst > 0 && worst < 16384); + unsigned height; + uint64_t leaves, small, end; + CHECK(pure_tree_test_shape(path, "large", &height, &leaves, &small, &end)); + CHECK(small == 0 && height == 2 && leaves >= 256 && leaves <= 512); + printf("{\"one_byte_edit_max_appended_bytes\":%llu,\"height\":%u,\"leaves\":%llu}\n", + (unsigned long long)worst, height, (unsigned long long)leaves); + free(bytes); +} + +static void faults(const char *path) { + unsigned char text[] = "fault-free"; + 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); + 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); + } + fail_stage = PURE_BEFORE_DIR_SYNC; + CHECK(pure_store_splice(path, "fault", o.revision, 10, 0, (PureSpan){text, 2}, + PURE_OBJECT_LIMIT, &revision) == PURE_UNCERTAIN); + fail_stage = -1; + unsigned char next[] = "fault-freefa"; + o = (Oracle){next, 12, o.revision + 1}; + CHECK(revision == o.revision); + expect(path, "fault", &o); +} + +/* A process that dies at each publication stage leaves either the previous + * or the next complete version, and its unpublished bytes are discarded. */ +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)); + pid_t child = fork(); + CHECK(child >= 0); + if (!child) { + crash = 1; + fail_stage = stage; + pure_store_splice(path, "crash", o.revision, 0, 1, (PureSpan){(const unsigned char *)"C", 1}, + PURE_OBJECT_LIMIT, &revision); + _exit(0); + } + int status; + 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); + 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); + } +} + +/* A torn newest head slot exposes the previous commit, and the next writer + * overwrites that slot and the unpublished tail. */ +static void torn_head(const char *path) { + unsigned char first[] = "first", second[] = "second"; + int64_t revision; + CHECK(pure_store_put(path, "torn", 0, (PureSpan){first, 5}, &revision) == PURE_OK); + Oracle o = {first, 5, revision}; + CHECK(pure_store_put(path, "torn", o.revision, (PureSpan){second, 6}, &revision) == PURE_OK); + 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(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); + CHECK(pure_store_put(path, "torn", o.revision, (PureSpan){second, 3}, &revision) == PURE_OK); + o = (Oracle){second, 3, revision}; + expect(path, "torn", &o); +} + +/* A reader's view stays byte-identical while another process commits and + * compacts; readers take no lock. */ +static void isolation(const char *path) { + size_t size = 300000; + unsigned char *bytes = malloc(size); + uint32_t state = 99; + CHECK(bytes); + fill(bytes, size, &state); + int64_t revision; + CHECK(pure_store_put(path, "shared", 0, (PureSpan){bytes, size}, &revision) == PURE_OK); + PureStoredView view; + CHECK(pure_store_get(path, "shared", size, &view) == PURE_OK); + uint64_t before = file_size(path); + pid_t child = fork(); + CHECK(child >= 0); + if (!child) { + int64_t at = revision; + unsigned char byte = 'x'; + for (unsigned i = 0; i < 400; ++i) + if (pure_store_splice(path, "shared", at, (i * 7919u) % size, 1, (PureSpan){&byte, 1}, + PURE_OBJECT_LIMIT, &at) != PURE_OK) _exit(1); + _exit(0); + } + int status; + CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 0); + CHECK(view.revision == revision && view.body.size == size && !memcmp(view.body.data, bytes, size)); + pure_store_release(&view); + PureStoredView latest; + CHECK(pure_store_get(path, "shared", size, &latest) == PURE_OK && latest.revision == revision + 400); + pure_store_release(&latest); + /* About 2 MB of path copies without compaction would exceed this. */ + CHECK(file_size(path) < before + 3 * 524288); + free(bytes); +} + +/* Damaged branch bytes are reported as storage errors, never trusted. */ +static void corruption(const char *path) { + unsigned char *bytes = malloc(200000); + uint32_t state = 3; + CHECK(bytes); + fill(bytes, 200000, &state); + int64_t revision; + 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); + 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)); + unsigned char original, damaged; + CHECK(pread(fd, &original, 1, at) == 1); + damaged = original ^ (unsigned char)(1u << (random_value(&state) % 8)); + CHECK(pwrite(fd, &damaged, 1, at) == 1); + PureStoredView view; + PureResult result = pure_store_get(path, "damaged", PURE_OBJECT_LIMIT, &view); + CHECK(result == PURE_OK || result == PURE_STORE_ERROR || result == PURE_NOT_FOUND); + if (result == PURE_OK) pure_store_release(&view); else ++errors; + CHECK(pwrite(fd, &original, 1, at) == 1); + } + CHECK(close(fd) == 0 && errors > 0); + free(bytes); +} + +static void many_keys(const char *path) { + int64_t revision; + char key[32]; + for (unsigned i = 0; i < 1024; ++i) { + unsigned id = (i * 617u) % 1024u; + snprintf(key, sizeof(key), "doc-%04u", id); + CHECK(pure_store_put(path, key, 0, (PureSpan){(const unsigned char *)key, strlen(key)}, &revision) == PURE_OK); + } + for (unsigned id = 0; id < 1024; ++id) { + snprintf(key, sizeof(key), "doc-%04u", id); + Oracle o = {(unsigned char *)key, strlen(key), 1}; + expect(path, key, &o); + } + PureStoredView view; + CHECK(pure_store_get(path, "doc-1024", 64, &view) == PURE_NOT_FOUND); + CHECK(pure_store_get(path, "doc-0001", 3, &view) == PURE_BUDGET); + CHECK(pure_store_splice(path, "doc-0002", 1, 9, 0, (PureSpan){0}, PURE_OBJECT_LIMIT, &revision) == PURE_LIMIT); + CHECK(pure_store_put(path, "doc-0003", 0, (PureSpan){0}, &revision) == PURE_CONFLICT && revision == 1); +} + +/* The newest catalog of a one-key store is one leaf; forge its revision. */ +static void revision_limit(const char *path) { + int64_t revision; + CHECK(pure_store_put(path, "work", 0, (PureSpan){(const unsigned char *)"w", 1}, &revision) == PURE_OK); + 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(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); + PureStoredView view; + CHECK(pure_store_get(path, "work", 1, &view) == PURE_OK && view.revision == INT64_MAX); + pure_store_release(&view); + CHECK(pure_store_splice(path, "work", INT64_MAX, 0, 0, (PureSpan){0}, PURE_OBJECT_LIMIT, &revision) == PURE_LIMIT); +} + +static bool count_key(void *context, const char *key) { + (void)key; + ++*(unsigned *)context; + return true; +} + +static void keys(const char *path) { + unsigned count = 0; + CHECK(pure_store_keys(path, count_key, &count) == PURE_OK && count == 0); + int64_t revision; + const char *names[] = {"m", "a", "zz", "a-b", "A"}; + for (unsigned i = 0; i < 5; ++i) + CHECK(pure_store_put(path, names[i], 0, (PureSpan){0}, &revision) == PURE_OK); + CHECK(pure_store_keys(path, count_key, &count) == PURE_OK && count == 5); +} + +static void clean(const char *path) { + DIR *entries = opendir(path); + CHECK(entries); + struct dirent *entry; + while ((entry = readdir(entries))) { + if (!strcmp(entry->d_name, ".") || !strcmp(entry->d_name, "..")) continue; + CHECK(strncmp(entry->d_name, ".pending-", 9)); + CHECK(unlinkat(dirfd(entries), entry->d_name, 0) == 0); + } + CHECK(closedir(entries) == 0 && rmdir(path) == 0); +} + +int main(void) { + void (*checks[])(const char *) = {oracle, amplification, faults, crashes, torn_head, + isolation, corruption, many_keys, revision_limit, keys}; + 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); + checks[i](path); + CHECK(pure_store_init(path) == PURE_OK); + clean(path); + } + 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\"}"); + return 0; +} |