diff options
| author | eunhhu <[email protected]> | 2026-09-27 13:14:38 +0100 |
|---|---|---|
| committer | eunhhu <[email protected]> | 2026-09-27 13:14:38 +0100 |
| commit | 37d3a1e152707d2da587440c4cea8601b9ef0fd2 (patch) | |
| tree | 1a4cbea9d8ca636deabd8fc24b0d73cdae2bdaef /store/store-tree.c | |
| parent | 5372d03b5968796515966e2cfc89628d6448f0c1 (diff) | |
| download | pure-37d3a1e152707d2da587440c4cea8601b9ef0fd2.tar.gz pure-37d3a1e152707d2da587440c4cea8601b9ef0fd2.zip | |
Lay Pure out as a core with the parts selected around it
The app model, the tree store and the HTTP host lived in
prototypes/connections among the earlier document runtime, the store was
known only by its format's name, and Vega compiled six Pure sources by path.
Now, as the README's principle has it:
core/ runtime, flow and the app model (module*.c), with what the core
asks of storage (state.h, store.h)
store/ the tree store (PURETRE3, PUREREC1)
serve/ pure-serve: HTTP host, outbound delivery, JSON in place
cli/ pure-run, pure-edit-module
A top-level make builds libpure.a (app model and store, for hosts to link)
and the hosts; make check runs the store's and the app model's tests. The
earlier runtime still builds from prototypes/connections against core/ and
store/ (check, check-tree and check-core pass). Design 009 records the rule:
parts by role behind libraries; git repositories split by coupling, so the
store stays here while its contract moves with the core, and Vega stays apart.
Diffstat (limited to 'store/store-tree.c')
| -rw-r--r-- | store/store-tree.c | 1856 |
1 files changed, 1856 insertions, 0 deletions
diff --git a/store/store-tree.c b/store/store-tree.c new file mode 100644 index 0000000..146e919 --- /dev/null +++ b/store/store-tree.c @@ -0,0 +1,1856 @@ +/* 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. A commit appends a record (its head: where its nodes + * are and their digest) and then its nodes, and syncs once: one contiguous + * range, which a flash device writes about twice as fast as two. Every few + * commits the head also goes into the older of two slots at the start, where + * a reader begins before following the records after it; 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 "state.h" +#include <errno.h> +#include <fcntl.h> +#include <inttypes.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <pthread.h> +#include <sys/mman.h> +#include <sys/stat.h> +#include <unistd.h> + +enum { FREE = 0xffff }; +/* 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 +enum { COMPACT_SLACK = 1048576 }; +#endif +#define LEAF_BIT (UINT64_C(1) << 63) +#define MEMORY_BIT (UINT64_C(1) << 62) +#define LOCATION (MEMORY_BIT - 1) +/* Slots of this format, and of the one before tail records (still read). */ +static const unsigned char head_magic[8] = {'P','U','R','E','T','R','E','3'}; +static const unsigned char earlier_magic[8] = {'P','U','R','E','T','R','E','2'}; +static const unsigned char record_magic[8] = {'P','U','R','E','R','E','C','1'}; +/* Commits between two heads written to a slot. */ +enum { CHECKPOINT = 16 }; +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 + * only between processes (it is the process's, and closing any descriptor of + * the lock file drops it). */ +static pthread_mutex_t writers = PTHREAD_MUTEX_INITIALIZER; +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, child count, or next free slot + 1 */ + unsigned height; /* FREE for a reclaimed slot */ +} Node; +/* One mapped version of the file, shared by the drafts of a store handle + * and by whoever still sends extents from it. */ +typedef struct PureMapping { const unsigned char *map; uint64_t end; unsigned refs; } Mapping; +typedef struct { + const unsigned char *map; + uint64_t end; + Mapping *shared; /* map belongs to this shared mapping, not to the tree */ + Node *nodes; + 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; +/* 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; + +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; +} + +/* Runtime-owned values (app states) live under '.' followed by a valid key. + * Document keys never start with '.', so the two never meet. */ +static bool runtime_key(const char *key) { + return key && key[0] == '.' && strlen(key) <= KEY && valid_key(key + 1); +} + +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)); +} + +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); + 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 unsigned char magic[8], const Head *head) { + memset(out, 0, SLOT); + memcpy(out, magic, 8); + 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 + 48, head->start); + encode64(out + 56, head->check); + encode64(out + SLOT - 8, digest_of(out, SLOT - 8)); +} + +/* A slot's head (record false) or a commit record's. */ +static bool decode_head(const unsigned char in[SLOT], bool record, Head *head) { + bool magic = record ? !memcmp(in, record_magic, 8) : !memcmp(in, head_magic, 8) || !memcmp(in, earlier_magic, 8); + if (!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), 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; +} + +/* ---- 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; +/* The newest head that counts per file, to follow records from. */ +typedef struct { dev_t device; ino_t inode; Head head; } Newest; +enum { NEWEST = 4 }; +static struct { + pthread_mutex_t lock; + Known heads[KNOWN]; + unsigned next; + Known publishing; /* written here, not durable yet */ + Newest newest[NEWEST]; + unsigned next_newest; +} 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 */ + } + if (state == CHECKED || state == DURABLE) { + Newest *at = NULL; + for (unsigned i = 0; i < NEWEST && !at; ++i) + if (known.newest[i].head.seq && known.newest[i].device == file->st_dev && + known.newest[i].inode == file->st_ino) at = &known.newest[i]; + if (!at) at = &known.newest[known.next_newest++ % NEWEST]; + else if (at->head.seq >= head->seq) at = NULL; + if (at) *at = (Newest){file->st_dev, file->st_ino, *head}; + } + pthread_mutex_unlock(&known.lock); +} + +static bool newest_known(const struct stat *file, Head *out) { + bool found = false; + pthread_mutex_lock(&known.lock); + for (unsigned i = 0; i < NEWEST && !found; ++i) + if (known.newest[i].head.seq && known.newest[i].device == file->st_dev && known.newest[i].inode == file->st_ino) { + *out = known.newest[i].head; + found = true; + } + pthread_mutex_unlock(&known.lock); + return found; +} + +#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)); + memset(known.newest, 0, sizeof(known.newest)); + 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, + uint64_t sequences[2]) { + Head heads[2]; + bool valid[2]; + for (unsigned i = 0; i < 2; ++i) { + valid[i] = decode_head(slots[i], false, &heads[i]) && heads[i].end <= (uint64_t)file->st_size; + if (sequences) sequences[i] = valid[i] ? heads[i].seq : 0; + } + 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; +} + +/* The commit whose record follows head's bytes, if its own bytes are whole. */ +static bool follow(int fd, const struct stat *file, bool writer, const Head *head, Head *out) { + unsigned char record[SLOT]; + uint64_t size = (uint64_t)file->st_size; + Head next; + if (head->end > size || size - head->end < SLOT || !read_at(fd, record, SLOT, head->end) || + !decode_head(record, true, &next) || next.seq != head->seq + 1 || next.start != head->end + SLOT || + next.end > size) return false; + int state = recall(file, &next); + if (state == BROKEN || (state == PUBLISHING && !writer)) return false; + if (state == UNKNOWN) { + bool counts = whole(fd, &next); + remember(file, &next, counts ? CHECKED : BROKEN); + if (!counts) return false; + } + *out = next; + return true; +} + +/* Whether head's record is in the file as it was made. */ +static bool recorded(int fd, const struct stat *file, const Head *head) { + unsigned char stored[SLOT], made[SLOT]; + if (head->start < DATA + SLOT || head->end > (uint64_t)file->st_size) return false; + encode_head(made, record_magic, head); + return read_at(fd, stored, SLOT, head->start - SLOT) && !memcmp(stored, made, SLOT); +} + +/* The newest commit: from the newer slot's head that counts, or from the + * newest this process has seen in the file when its record is still there, + * along the records that follow. sequences gets the slots' (0: none). */ +static bool read_head(int fd, struct stat *file, bool writer, Head *out, uint64_t sequences[2]) { + unsigned char slots[BLOCK + SLOT]; + Head head, seen; + if (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, &head, sequences)) return false; + if (newest_known(file, &seen) && seen.seq > head.seq && recorded(fd, file, &seen)) head = seen; + while (follow(fd, file, writer, &head, &head)) {} + *out = head; + return true; +} + +static bool map_tree(int fd, const struct stat *file, const Head *head, Tree *tree) { + *tree = (Tree){0}; + 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}; + return true; +} + +/* A lent mapping may be released on another thread (a host that answers on + * its connection threads): its count is atomic. */ +static void release_mapping(Mapping *mapping) { + if (!mapping || __atomic_sub_fetch(&mapping->refs, 1, __ATOMIC_ACQ_REL)) return; + munmap((void *)mapping->map, (size_t)mapping->end); + free(mapping); +} + +static void release_tree(Tree *tree) { + if (tree->shared) release_mapping(tree->shared); + else 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 size_t node_bytes(const Node *node) { + return node->bytes ? node->size : node->size * sizeof(Entry); +} + +static bool add_node(Tree *tree, Node node, uint64_t *index) { + tree->bytes += node_bytes(&node); + if (tree->free_head) { + *index = tree->free_head - 1; + tree->free_head = tree->nodes[*index].size; + --tree->free_count; + tree->nodes[*index] = node; + return true; + } + if (tree->count == tree->capacity) { + size_t capacity = tree->capacity ? 2 * tree->capacity : 64; + Node *nodes = realloc(tree->nodes, capacity * sizeof(*nodes)); + if (!nodes) { tree->bytes -= node_bytes(&node); 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) { + 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; + 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. The arena is not changed, so a failed commit leaves a + * draft intact. With relocate, stored nodes are copied too (compaction, or a + * draft whose base file was replaced by compaction). */ +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. 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_magic, 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; + 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 struct stat *file, const Head *head) { + Tree tree; + 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; + 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, 0, 0}; + 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. */ + pthread_mutex_lock(&writers); + 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; + struct stat file; + int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC); + if (fd >= 0) { + ok = read_head(fd, &file, true, &head, NULL); /* 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); + pthread_mutex_unlock(&writers); + 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; + struct stat file; + Entry document; + uint64_t index; + int64_t revision; + bool found; + unsigned height; + PureResult result = PURE_STORE_ERROR; + if (!read_head(fd, &file, false, &head, NULL) || !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; } + 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; + uint64_t slots[2]; /* the sequences of the slots' heads */ + Tree tree; + struct stat file; + 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); + if (session->lock >= 0) close(session->lock); + if (session->directory >= 0) close(session->directory); + if (session->turn) pthread_mutex_unlock(&writers); + session->turn = false; +} + +static PureResult open_session(const char *path, Session *session) { + *session = (Session){.directory = -1, .lock = -1, .fd = -1}; + pthread_mutex_lock(&writers); /* another thread of this process waits, not fails */ + session->turn = true; + 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->file, true, &session->head, session->slots)) + return PURE_STORE_ERROR; + /* 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; +} + +/* A commit's bytes start after room for its record at the published end. */ +static uint64_t commit_start(const Session *session) { return session->head.end + SLOT; } + +/* 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 as the record before them 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; } +#endif + +static PureResult publish(Session *session, Writer *writer, size_t count, const char *const *keys, + const Entry *documents, int64_t *revisions) { +#ifdef PURE_STORE_TEST + ++publications; +#endif + Entry catalog = {session->head.catalog, session->head.catalog_bytes}; + bool written = true; + for (size_t i = 0; written && i < count; ++i) { + unsigned char record[RECORD] = {0}; + uint64_t index; + bool found; + Entry current, updated; + int64_t stored; + memcpy(record, keys[i], strlen(keys[i])); + encode64(record + KEY, documents[i].ref); + encode64(record + KEY + 8, documents[i].bytes); + encode64(record + KEY + 16, (uint64_t)revisions[i] + 1); + written = find(&session->tree, catalog, keys[i], &index, &found, ¤t, &stored) && + rebuild(&session->tree, catalog, index * RECORD, found ? RECORD : 0, + (PureSpan){record, RECORD}, &updated); + catalog = updated; + } + written = written && write_root(&session->tree, writer, &catalog, false) && flush(writer); + /* 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, + commit_start(session), digest_end(&writer->digest)}; + unsigned char record[SLOT], slot[SLOT]; + encode_head(record, record_magic, &published); + encode_head(slot, head_magic, &published); + /* Every few commits the head goes into the older slot as well: one more + * range for that sync, and fewer records for a reader to follow. */ + uint64_t newer = session->slots[0] > session->slots[1] ? session->slots[0] : session->slots[1]; + unsigned older = session->slots[0] > session->slots[1]; + bool checkpoint = published.seq - newer >= CHECKPOINT; + for (size_t i = 0; i < count; ++i) ++revisions[i]; /* Past this point the head may already be visible. */ + remember(&session->file, &published, PUBLISHING); + bool durable = write_at(session->fd, record, SLOT, published.start - SLOT) && + (!checkpoint || write_at(session->fd, slot, SLOT, older * 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; +} + +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 = commit_start(&session)}; + if (!write_root(&session.tree, writer, &next, false)) goto finish; + result = publish(&session, writer, 1, &key, &next, revision); +finish: + free(writer); + close_session(&session); + return result; +} + +/* Keys come from the catalog in ascending byte order. Runtime values are not + * documents and are left out. */ +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; + struct stat file; + PureResult result = PURE_STORE_ERROR; + if (read_head(fd, &file, false, &head, NULL) && map_tree(fd, &file, &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 (runtime_key(key)) continue; + if (!valid_key(key)) { result = PURE_STORE_ERROR; break; } + if (!visit(context, key)) break; + } + } + release_tree(&tree); + close(fd); + return result; +} + +/* A store kept open by one owner thread: its current file and a mapping of + * its newest version, shared by the drafts opened through it. */ +struct PureStore { + char *path; + int directory, fd; + dev_t device; + ino_t inode; + Head head; + Mapping *mapping; + bool pinned; /* drafts open at the version of the last check */ +}; + +struct PureDraft { + PureStore *store; /* the handle it was opened through, or NULL */ + char *path; /* owned unless opened through a store handle */ + char key[KEY + 1]; + Tree tree; /* the base file's mapping plus this draft's node arena */ + dev_t device; /* identity of the base file; compaction replaces it */ + ino_t inode; + Entry root; + int64_t revision; + size_t live; /* arena bytes reachable at the last collection */ + bool held; /* a transaction still reads earlier versions */ +}; + +/* Rebase the draft on the committed version of its key. */ +static PureResult load_draft(PureDraft *draft) { + int directory = open(draft->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; + struct stat info; + Entry document; + uint64_t index; + int64_t revision; + bool found; + bool ok = read_head(fd, &info, false, &head, NULL) && map_tree(fd, &info, &head, &tree) && + find(&tree, (Entry){head.catalog, head.catalog_bytes}, draft->key, &index, &found, + &document, &revision); + close(fd); + if (!ok) { release_tree(&tree); return PURE_STORE_ERROR; } + release_tree(&draft->tree); + draft->tree = tree; + draft->device = info.st_dev; + draft->inode = info.st_ino; + draft->root = document; + draft->revision = revision; + draft->live = 0; + return PURE_OK; +} + +PureResult pure_draft_open(const char *path, const char *key, PureDraft **out) { + *out = NULL; + if (!path || !(valid_key(key) || runtime_key(key))) return PURE_INVALID; + PureDraft *draft = calloc(1, sizeof(*draft)); + if (!draft || !(draft->path = strdup(path))) { free(draft); return PURE_ALLOCATION; } + memcpy(draft->key, key, strlen(key) + 1); + PureResult result = load_draft(draft); + if (result != PURE_OK) { pure_draft_close(draft); return result; } + *out = draft; + return PURE_OK; +} + +PureResult pure_state_open(const char *path, const char *name, PureDraft **out) { + char key[KEY + 2]; + *out = NULL; + if (!valid_key(name) || strlen(name) >= KEY) return PURE_INVALID; + snprintf(key, sizeof(key), ".%s", name); + return pure_draft_open(path, key, out); +} + +PureDraft *pure_draft_new(void) { + return calloc(1, sizeof(PureDraft)); /* no path: nothing mapped, nothing to commit */ +} + +PureDraft *pure_draft_freeze(const PureDraft *draft) { + if (!draft->tree.shared || (draft->root.ref & MEMORY_BIT)) return NULL; + PureDraft *frozen = calloc(1, sizeof(*frozen)); + if (!frozen) return NULL; + Mapping *mapping = draft->tree.shared; + __atomic_add_fetch(&mapping->refs, 1, __ATOMIC_RELAXED); + *frozen = (PureDraft){.tree = {.map = mapping->map, .end = mapping->end, .shared = mapping}, + .device = draft->device, .inode = draft->inode, .root = draft->root, + .revision = draft->revision}; + memcpy(frozen->key, draft->key, sizeof(frozen->key)); + return frozen; +} + +void pure_draft_close(PureDraft *draft) { + if (!draft) return; + release_tree(&draft->tree); + if (!draft->store) free(draft->path); + free(draft); +} + +/* 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)) + return PURE_STORE_ERROR; + if (store->fd < 0 || info.st_dev != store->device || info.st_ino != store->inode) { + int fd = openat(store->directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC); + if (fd < 0 || fstat(fd, &info)) { if (fd >= 0) close(fd); return PURE_STORE_ERROR; } + if (store->fd >= 0) close(store->fd); + store->fd = fd; + store->device = info.st_dev; + store->inode = info.st_ino; + store->head = (Head){0}; + } + if (store->head.seq) { + /* The next commit's record follows the head's bytes; a slot shows a + * newer head only for a commit of the earlier format. */ + const unsigned char *map = store->mapping->map; + unsigned char next[16]; + bool newer = decode64(map + 8) > store->head.seq || decode64(map + BLOCK + 8) > store->head.seq || + (read_at(store->fd, next, sizeof(next), store->head.end) && !memcmp(next, record_magic, 8) && + decode64(next + 8) == store->head.seq + 1); + if (!newer) return PURE_OK; + } + Head head; + if (!read_head(store->fd, &info, false, &head, NULL)) 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, &info, &head, &tree)) { free(mapping); return PURE_STORE_ERROR; } + *mapping = (Mapping){tree.map, tree.end, 1}; + release_mapping(store->mapping); + store->mapping = mapping; + store->head = head; + return PURE_OK; +} + +/* Base a draft on the handle's newest version of its key, sharing the mapping. + * On failure the draft is unchanged. */ +static PureResult rebase(PureDraft *draft) { + PureStore *store = draft->store; + Mapping *mapping = store->mapping; + Tree tree = {.map = mapping->map, .end = mapping->end, .shared = mapping}; + Entry document; + uint64_t index; + int64_t revision; + bool found; + if (!find(&tree, (Entry){store->head.catalog, store->head.catalog_bytes}, draft->key, &index, &found, &document, + &revision)) return PURE_STORE_ERROR; + __atomic_add_fetch(&mapping->refs, 1, __ATOMIC_RELAXED); + release_tree(&draft->tree); + draft->tree = tree; + draft->device = store->device; + draft->inode = store->inode; + draft->root = document; + draft->revision = revision; + draft->live = 0; + return PURE_OK; +} + +PureResult pure_store_open(const char *path, PureStore **out) { + *out = NULL; + PureStore *store = calloc(1, sizeof(*store)); + if (!store || !path || !(store->path = strdup(path))) { free(store); return PURE_ALLOCATION; } + store->fd = -1; + store->directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + PureResult result = store->directory < 0 ? PURE_STORE_ERROR : refresh(store); + if (result != PURE_OK) { pure_store_close(store); return result; } + *out = store; + return PURE_OK; +} + +void pure_store_close(PureStore *store) { + if (!store) return; + release_mapping(store->mapping); + if (store->fd >= 0) close(store->fd); + if (store->directory >= 0) close(store->directory); + free(store->path); + free(store); +} + +PureMapping *pure_draft_mapping(const PureDraft *draft) { + Mapping *mapping = draft->tree.shared; + if (mapping) __atomic_add_fetch(&mapping->refs, 1, __ATOMIC_RELAXED); + return mapping; +} + +bool pure_mapping_holds(const PureMapping *mapping, PureSpan span) { + uintptr_t base = (uintptr_t)mapping->map, at = (uintptr_t)span.data; + return !span.size || (span.data && at >= base && at - base <= mapping->end && span.size <= mapping->end - (at - base)); +} + +void pure_mapping_release(PureMapping *mapping) { release_mapping(mapping); } + +PureResult pure_store_pin(PureStore *store, bool pinned) { + store->pinned = false; + PureResult result = pinned ? refresh(store) : PURE_OK; + store->pinned = pinned && result == PURE_OK; + return result; +} + +static PureResult open_through(PureStore *store, const char *key, PureDraft **out) { + *out = NULL; + PureResult result = store->pinned ? PURE_OK : refresh(store); + if (result != PURE_OK) return result; + PureDraft *draft = calloc(1, sizeof(*draft)); + if (!draft) return PURE_ALLOCATION; + *draft = (PureDraft){.store = store, .path = store->path}; + memcpy(draft->key, key, strlen(key) + 1); + result = rebase(draft); + if (result != PURE_OK) { free(draft); return result; } + *out = draft; + return PURE_OK; +} + +PureResult pure_store_draft(PureStore *store, const char *key, PureDraft **out) { + *out = NULL; + return store && valid_key(key) ? open_through(store, key, out) : PURE_INVALID; +} + +PureResult pure_store_state(PureStore *store, const char *name, PureDraft **out) { + char key[KEY + 2]; + *out = NULL; + if (!store || !valid_key(name) || strlen(name) >= KEY) return PURE_INVALID; + snprintf(key, sizeof(key), ".%s", name); + return open_through(store, key, out); +} + +size_t pure_draft_size(const PureDraft *draft) { return (size_t)draft->root.bytes; } + +int64_t pure_draft_revision(const PureDraft *draft) { return draft->revision; } + +static size_t mark(const Tree *tree, Entry entry, unsigned char *marks) { + if (!(entry.ref & MEMORY_BIT)) return 0; + size_t at = (size_t)(entry.ref & LOCATION); + if (marks[at]) return 0; + marks[at] = 1; + const Node *node = &tree->nodes[at]; + size_t bytes = node_bytes(node); + if (!node->bytes) + for (size_t i = 0; i < node->size; ++i) bytes += mark(tree, node->children[i], marks); + return bytes; +} + +/* Every edit copies a leaf and path, so reclaim arena nodes the root no + * longer reaches once they outweigh the reachable ones (and 32 KiB). */ +static void collect(PureDraft *draft) { + Tree *tree = &draft->tree; + if (draft->held) return; + size_t allowance = draft->live > 32768 ? draft->live : 32768; + if (tree->bytes <= draft->live || tree->bytes - draft->live <= allowance) return; + unsigned char *marks = calloc(tree->count, 1); + if (!marks) return; + draft->live = mark(tree, draft->root, marks); + for (size_t i = 0; i < tree->count; ++i) { + Node *node = &tree->nodes[i]; + if (marks[i] || node->height == FREE) continue; + tree->bytes -= node_bytes(node); + free(node->bytes); + free(node->children); + *node = (Node){.size = tree->free_head, .height = FREE}; + tree->free_head = i + 1; + ++tree->free_count; + } + free(marks); +} + +PureResult pure_draft_splice(PureDraft *draft, size_t offset, size_t remove, PureSpan insert) { + uint64_t size = draft->root.bytes; + if (insert.size && !insert.data) return PURE_INVALID; + if (offset > size || remove > size - offset || insert.size > PURE_OBJECT_LIMIT - (size - remove)) + return PURE_LIMIT; + if (!remove && !insert.size) return PURE_OK; + Entry next; + bool ok = rebuild(&draft->tree, draft->root, offset, remove, insert, &next); + if (ok) draft->root = next; + collect(draft); + return ok ? PURE_OK : PURE_STORE_ERROR; +} + +PureVersion pure_draft_version(const PureDraft *draft) { + return (PureVersion){draft->root.ref, draft->root.bytes}; +} + +PureResult pure_draft_read_version(const PureDraft *draft, PureVersion version, size_t offset, + size_t length, bool (*visit)(void *, PureSpan), void *context) { + Entry root = {version.ref, version.bytes}; + unsigned height; + if (offset > root.bytes || length > root.bytes - offset) return PURE_LIMIT; + if (!length) return PURE_OK; + if (!root_height(&draft->tree, root, &height)) return PURE_STORE_ERROR; + return visit_range(&draft->tree, root, height, offset, length, visit, context) ? + PURE_OK : PURE_STORE_ERROR; +} + +PureResult pure_draft_read(const PureDraft *draft, size_t offset, size_t length, + bool (*visit)(void *, PureSpan), void *context) { + return pure_draft_read_version(draft, pure_draft_version(draft), offset, length, visit, context); +} + +void pure_draft_restore(PureDraft *draft, PureVersion version) { + draft->root = (Entry){version.ref, version.bytes}; +} + +void pure_draft_hold(PureDraft *draft, bool held) { + draft->held = held; + if (!held) collect(draft); +} + +/* ---- Group commits ------------------------------------------------------ + * Threads that commit at the same time share one commit: they queue, the + * first becomes the leader and publishes every queued request of the same + * store with one head (one pair of syncs); the others wait for its result. + * Each request stays all or nothing on its own: a request with a stale draft + * is left out, and so is one whose key an earlier request of the same group + * already changes. The leader touches the waiters' drafts and handles only + * while they wait. */ +typedef struct Waiter { + PureDraft *const *drafts; + size_t count; + int64_t *revisions; + PureResult result; + bool accepted, done; + struct Waiter *next; +} Waiter; + +static struct { + pthread_mutex_t lock; + pthread_cond_t finished, joined; + Waiter *head, *tail; + size_t queued, expected; /* expected: how many requests groups have held lately */ + bool leading; + unsigned long groups, requests; +} group = {PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER, PTHREAD_COND_INITIALIZER, NULL, NULL, 0, 1, false, 0, 0}; + +void pure_store_group_counts(unsigned long *groups, unsigned long *requests) { + pthread_mutex_lock(&group.lock); + *groups = group.groups; + *requests = group.requests; + pthread_mutex_unlock(&group.lock); +} + +enum { GATHER_NS = 2000000 }; + +/* Publish the queued requests of the store at path in one session. */ +static void commit_group(const char *path, Waiter *first) { + size_t total = 0; + for (Waiter *w = first; w; w = w->next) + if (!strcmp(w->drafts[0]->path, path)) { total += w->count; w->accepted = false; } + Session session; + Writer *writer = NULL; + Entry *roots = malloc(total * sizeof(*roots)); + const char **keys = malloc(total * sizeof(*keys)); + int64_t *revisions = malloc(total * sizeof(*revisions)); + PureDraft **drafts = malloc(total * sizeof(*drafts)); + bool allocated = roots && keys && revisions && drafts, checked = false; + PureResult result = allocated ? open_session(path, &session) : PURE_ALLOCATION; + size_t count = 0; + if (result == PURE_OK) { + /* A request holds when each of its keys is at its draft's base + * revision and no earlier request of this group changes it. */ + for (Waiter *w = first; w; w = w->next) { + if (strcmp(w->drafts[0]->path, path)) continue; + w->result = PURE_OK; + for (size_t i = 0; i < w->count && w->result == PURE_OK; ++i) { + PureDraft *d = w->drafts[i]; + Entry current; + uint64_t index; + int64_t stored; + bool found; + if (!find(&session.tree, (Entry){session.head.catalog, session.head.catalog_bytes}, d->key, &index, + &found, ¤t, &stored)) { w->result = PURE_STORE_ERROR; break; } + for (size_t k = 0; k < count; ++k) + if (!strcmp(keys[k], d->key)) { ++stored; break; } /* it moves with the earlier request */ + if (stored != d->revision) { w->revisions[i] = stored; w->result = PURE_CONFLICT; } + else if (stored == INT64_MAX) w->result = PURE_LIMIT; + } + if (w->result != PURE_OK) continue; + w->accepted = true; + for (size_t i = 0; i < w->count; ++i) { + drafts[count] = w->drafts[i]; + roots[count] = w->drafts[i]->root; + keys[count] = w->drafts[i]->key; + revisions[count++] = w->drafts[i]->revision; + } + } + checked = true; + if (count) { + result = PURE_ALLOCATION; + if ((writer = malloc(sizeof(*writer)))) { + *writer = (Writer){.fd = session.fd, .at = commit_start(&session)}; + bool written = true; + for (size_t i = 0; i < count && written; ++i) { + /* After compaction a draft's stored references name the old file. */ + 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); + } + result = written ? publish(&session, writer, count, keys, roots, revisions) : PURE_STORE_ERROR; + } + } + /* Rebase while still holding the writer lock; the arenas are no longer + * needed. Drafts of a store handle share one new mapping. If remapping + * fails a draft keeps equal content at its new revision. */ + PureStore *refreshed[8] = {NULL}; + size_t fresh = 0; + for (size_t i = 0; result == PURE_OK && i < count; ++i) { + PureStore *store = drafts[i]->store; + bool rebased; + if (store) { + bool seen = false; + for (size_t k = 0; k < fresh && !seen; ++k) seen = refreshed[k] == store; + if (!seen && refresh(store) == PURE_OK) { + seen = true; + if (fresh < 8) refreshed[fresh++] = store; + } + rebased = seen && rebase(drafts[i]) == PURE_OK; + } else rebased = load_draft(drafts[i]) == PURE_OK; + if (!rebased) drafts[i]->revision = revisions[i]; + } + } + /* Before the checks every request shares the failure; after them an + * accepted request takes the publication's result and new revisions, and + * a refused one keeps its own result. */ + size_t at = 0; + for (Waiter *w = first; w; w = w->next) { + if (strcmp(w->drafts[0]->path, path)) continue; + if (!checked) { w->result = result; continue; } + if (!w->accepted) continue; + for (size_t i = 0; i < w->count; ++i) w->revisions[i] = revisions[at++]; + w->result = result; + } + free(writer); + free(roots); + free((void *)keys); + free(revisions); + free(drafts); + if (allocated) close_session(&session); +} + +PureResult pure_draft_commit_all(PureDraft *const *drafts, size_t count, int64_t *revisions) { + for (size_t i = 0; i < count; ++i) { + revisions[i] = drafts[i]->revision; + if (!drafts[i]->path || strcmp(drafts[i]->path, drafts[0]->path)) return PURE_INVALID; + for (size_t k = 0; k < i; ++k) if (!strcmp(drafts[k]->key, drafts[i]->key)) return PURE_INVALID; + } + if (!count) return PURE_OK; + Waiter me = {.drafts = drafts, .count = count, .revisions = revisions}; + pthread_mutex_lock(&group.lock); + if (group.tail) group.tail->next = &me; else group.head = &me; + group.tail = &me; + ++group.queued; + pthread_cond_signal(&group.joined); + while (!me.done) { + if (group.leading) { pthread_cond_wait(&group.finished, &group.lock); continue; } + group.leading = true; + /* When several threads have been committing together, give the others + * a moment (at most 2 ms) to join, or they would miss this group and + * wait for the next one: a sync costs far more. One thread never waits. */ + if (group.expected > 1 && group.queued < group.expected) { + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_nsec += GATHER_NS; + if (deadline.tv_nsec >= 1000000000) { ++deadline.tv_sec; deadline.tv_nsec -= 1000000000; } + while (group.queued < group.expected && + pthread_cond_timedwait(&group.joined, &group.lock, &deadline) == 0) {} + } + Waiter *batch = group.head; + size_t size = group.queued; + group.head = group.tail = NULL; + group.queued = 0; + ++group.groups; + group.requests += size; + /* Remember the largest recent group, forgetting one member per group. */ + group.expected = size > group.expected - 1 ? size : group.expected - 1; + if (!group.expected) group.expected = 1; + pthread_mutex_unlock(&group.lock); + /* One session per store among the queued requests. */ + for (Waiter *w = batch; w; w = w->next) { + bool earlier = false; + for (Waiter *v = batch; v != w && !earlier; v = v->next) earlier = !strcmp(v->drafts[0]->path, w->drafts[0]->path); + if (!earlier) commit_group(w->drafts[0]->path, batch); + } + pthread_mutex_lock(&group.lock); + for (Waiter *w = batch, *next; w; w = next) { next = w->next; w->done = true; } + group.leading = false; + pthread_cond_broadcast(&group.finished); + } + pthread_mutex_unlock(&group.lock); + return me.result; +} + +PureResult pure_draft_commit(PureDraft *draft, int64_t *revision) { + return pure_draft_commit_all(&draft, 1, revision); +} + +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: live arena nodes and their payload bytes. */ +void pure_draft_test_arena(const PureDraft *draft, size_t *nodes, size_t *bytes) { + *nodes = *bytes = 0; + for (size_t i = 0; i < draft->tree.count; ++i) { + const Node *node = &draft->tree.nodes[i]; + if (node->height == FREE) continue; + ++*nodes; + *bytes += node->bytes ? node->size : node->size * sizeof(Entry); + } +} + +/* 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; + struct stat file; + Entry document; + uint64_t index; + int64_t revision; + bool found = false; + *leaves = *small = 0; + bool ok = fd >= 0 && read_head(fd, &file, false, &head, NULL) && 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)); + if (ok) *file_end = head.end; + release_tree(&tree); + if (fd >= 0) close(fd); + return ok; +} +#endif |
