aboutsummaryrefslogtreecommitdiffstats
path: root/prototypes
diff options
context:
space:
mode:
Diffstat (limited to 'prototypes')
-rw-r--r--prototypes/connections/module-store.c23
-rw-r--r--prototypes/connections/module-store.h3
-rw-r--r--prototypes/connections/state.h9
-rw-r--r--prototypes/connections/store-tree.c143
-rw-r--r--prototypes/connections/test-tree-store.c63
5 files changed, 233 insertions, 8 deletions
diff --git a/prototypes/connections/module-store.c b/prototypes/connections/module-store.c
index 02d6361..7e759fd 100644
--- a/prototypes/connections/module-store.c
+++ b/prototypes/connections/module-store.c
@@ -58,6 +58,29 @@ static PureResult save(void *context, void *const *documents, size_t document_co
return result;
}
+/* Through a store handle: documents and states share its mapping. */
+static PureResult open_handle_document(void *context, uint32_t collection, const char *key, void **document,
+ int64_t *revision) {
+ (void)collection;
+ PureDraft *draft;
+ PureResult result = pure_store_draft(context, key, &draft);
+ if (result != PURE_OK) return result;
+ pure_draft_hold(draft, true);
+ *document = draft;
+ *revision = pure_draft_revision(draft);
+ return PURE_OK;
+}
+
+static PureResult handle_state(void *context, const char *name, PureDraft **out) {
+ return pure_store_state(context, name, out);
+}
+
+PureData pure_store_data(PureStore *store) {
+ return (PureData){.context = store, .open = open_handle_document, .version = version, .read = read,
+ .splice = splice, .commit = commit, .close = close_document, .state = handle_state,
+ .save = save};
+}
+
PureData pure_tree_data(const char *path) {
return (PureData){.context = (void *)path, .open = open_document, .version = version,
.read = read, .splice = splice, .commit = commit, .close = close_document,
diff --git a/prototypes/connections/module-store.h b/prototypes/connections/module-store.h
index 0a4b494..d14ab3d 100644
--- a/prototypes/connections/module-store.h
+++ b/prototypes/connections/module-store.h
@@ -5,4 +5,7 @@
* one event opens the message key as a held draft and commits it together
* with the states the event changed. The path outlives the app. */
PureData pure_tree_data(const char *path);
+/* The same through a store handle (state.h): opening a document or state
+ * checks only whether the store changed. The handle outlives the app. */
+PureData pure_store_data(PureStore *);
#endif
diff --git a/prototypes/connections/state.h b/prototypes/connections/state.h
index 55ad257..25c0f61 100644
--- a/prototypes/connections/state.h
+++ b/prototypes/connections/state.h
@@ -21,6 +21,15 @@ PureResult pure_draft_open(const char *path, const char *key, PureDraft **out);
* under a name that is never a document key (hosts listing documents do not
* see it). name is a valid key of at most 63 bytes. */
PureResult pure_state_open(const char *path, const char *name, PureDraft **out);
+/* A store kept open by one owner thread. Drafts opened through it share one
+ * mapping of the newest version, and opening one checks only whether the file
+ * changed; commits of its drafts map the new version once. It must outlive
+ * its drafts. */
+typedef struct PureStore PureStore;
+PureResult pure_store_open(const char *path, PureStore **out);
+void pure_store_close(PureStore *);
+PureResult pure_store_draft(PureStore *, const char *key, PureDraft **out);
+PureResult pure_store_state(PureStore *, const char *name, PureDraft **out);
/* Memory only. Fails without change on a bad range or when memory runs out. */
PureResult pure_draft_splice(PureDraft *, size_t offset, size_t remove, PureSpan insert);
size_t pure_draft_size(const PureDraft *);
diff --git a/prototypes/connections/store-tree.c b/prototypes/connections/store-tree.c
index 96bd65f..c41393b 100644
--- a/prototypes/connections/store-tree.c
+++ b/prototypes/connections/store-tree.c
@@ -48,9 +48,12 @@ typedef struct {
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. */
+typedef struct { 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 */
@@ -182,8 +185,15 @@ static bool map_tree(int fd, const Head *head, Tree *tree) {
return true;
}
+static void release_mapping(Mapping *mapping) {
+ if (!mapping || --mapping->refs) return;
+ munmap((void *)mapping->map, (size_t)mapping->end);
+ free(mapping);
+}
+
static void release_tree(Tree *tree) {
- if (tree->map) munmap((void *)tree->map, (size_t)tree->end);
+ 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);
@@ -953,8 +963,20 @@ PureResult pure_store_keys(const char *path, bool (*visit)(void *, const char *k
return result;
}
-struct PureDraft {
+/* 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;
+};
+
+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 */
@@ -1021,10 +1043,110 @@ PureDraft *pure_draft_new(void) {
void pure_draft_close(PureDraft *draft) {
if (!draft) return;
release_tree(&draft->tree);
- free(draft->path);
+ if (!draft->store) free(draft->path);
free(draft);
}
+/* Bring the handle to the newest published version: the same file and head
+ * cost two system calls; a new commit maps again, a compaction reopens. */
+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};
+ }
+ Head head;
+ if (!read_head(store->fd, &head)) return PURE_STORE_ERROR;
+ if (store->mapping && head.seq == store->head.seq && head.end == store->head.end) return PURE_OK;
+ Tree tree;
+ Mapping *mapping = malloc(sizeof(*mapping));
+ if (!mapping) return PURE_ALLOCATION;
+ if (!map_tree(store->fd, &head, &tree)) { free(mapping); return PURE_STORE_ERROR; }
+ *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;
+ ++mapping->refs;
+ 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);
+}
+
+static PureResult open_through(PureStore *store, const char *key, PureDraft **out) {
+ *out = NULL;
+ PureResult result = 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; }
@@ -1151,9 +1273,18 @@ PureResult pure_draft_commit_all(PureDraft *const *drafts, size_t count, int64_t
}
result = publish(&session, writer, count, keys, roots, revisions);
/* Rebase while still holding the writer lock; the arenas are no longer
- * needed. If remapping fails a draft keeps equal content at its new revision. */
- for (size_t i = 0; result == PURE_OK && i < count; ++i)
- if (load_draft(drafts[i]) != PURE_OK) drafts[i]->revision = revisions[i];
+ * needed. Drafts of a store handle share one new mapping. If remapping
+ * fails a draft keeps equal content at its new revision. */
+ PureStore *refreshed = NULL;
+ for (size_t i = 0; result == PURE_OK && i < count; ++i) {
+ PureStore *store = drafts[i]->store;
+ bool rebased;
+ if (store) {
+ if (store != refreshed && refresh(store) == PURE_OK) refreshed = store;
+ rebased = store == refreshed && rebase(drafts[i]) == PURE_OK;
+ } else rebased = load_draft(drafts[i]) == PURE_OK;
+ if (!rebased) drafts[i]->revision = revisions[i];
+ }
finish:
free(writer);
free(roots);
diff --git a/prototypes/connections/test-tree-store.c b/prototypes/connections/test-tree-store.c
index f6a4b1b..3f6d2f0 100644
--- a/prototypes/connections/test-tree-store.c
+++ b/prototypes/connections/test-tree-store.c
@@ -765,6 +765,65 @@ static void batch(const char *path) {
}
}
+/* A store handle shares one mapping among its drafts, sees commits made by
+ * other writers when a draft opens, follows a compaction that replaced the
+ * file, and maps a batch it committed once. */
+static void store_handle(const char *path) {
+ unsigned char scratch[64];
+ int64_t revision, revisions[3];
+ CHECK(pure_store_put(path, "shared", 0, (PureSpan){(const unsigned char *)"one", 3}, &revision) == PURE_OK);
+ PureStore *store;
+ CHECK(pure_store_open(path, &store) == PURE_OK);
+ PureDraft *a, *b, *state, *bad;
+ CHECK(pure_store_draft(store, "shared", &a) == PURE_OK);
+ expect_draft(a, &(Oracle){(unsigned char *)"one", 3, 1}, scratch);
+ CHECK(pure_store_draft(store, "bad.key", &bad) == PURE_INVALID && !bad);
+ /* Another writer commits: the next draft sees it, the open one keeps its version. */
+ CHECK(pure_store_put(path, "shared", 1, (PureSpan){(const unsigned char *)"two", 3}, &revision) == PURE_OK);
+ CHECK(pure_store_draft(store, "shared", &b) == PURE_OK);
+ expect_draft(b, &(Oracle){(unsigned char *)"two", 3, 2}, scratch);
+ expect_draft(a, &(Oracle){(unsigned char *)"one", 3, 1}, scratch);
+ CHECK(pure_draft_splice(a, 3, 0, (PureSpan){(const unsigned char *)"!", 1}) == PURE_OK);
+ CHECK(pure_draft_commit(a, &revision) == PURE_CONFLICT && revision == 2);
+ pure_draft_close(a);
+ /* A batch of handle drafts, a runtime value among them. */
+ CHECK(pure_store_state(store, "values", &state) == PURE_OK && pure_draft_revision(state) == 0);
+ CHECK(pure_store_draft(store, "other", &a) == PURE_OK);
+ CHECK(pure_draft_splice(b, 3, 0, (PureSpan){(const unsigned char *)"+", 1}) == PURE_OK);
+ CHECK(pure_draft_splice(state, 0, 0, (PureSpan){(const unsigned char *)"v", 1}) == PURE_OK);
+ CHECK(pure_draft_splice(a, 0, 0, (PureSpan){(const unsigned char *)"o", 1}) == PURE_OK);
+ PureDraft *batch[3] = {b, state, a};
+ CHECK(pure_draft_commit_all(batch, 3, revisions) == PURE_OK && revisions[0] == 3 && revisions[1] == 1);
+ expect(path, "shared", &(Oracle){(unsigned char *)"two+", 4, 3});
+ expect_draft(state, &(Oracle){(unsigned char *)"v", 1, 1}, scratch);
+ /* Other writers grow the file past compaction: it is replaced under the handle. */
+ unsigned char *big = malloc(100000);
+ uint32_t seed = 5;
+ CHECK(big);
+ fill(big, 100000, &seed);
+ int64_t filler = 0;
+ char name[512];
+ struct stat before, replaced;
+ snprintf(name, sizeof(name), "%s/pure.tree", path);
+ CHECK(stat(name, &before) == 0);
+ for (int i = 0; i < 8; ++i) CHECK(pure_store_put(path, "filler", filler, (PureSpan){big, 100000}, &filler) == PURE_OK);
+ CHECK(stat(name, &replaced) == 0 && replaced.st_ino != before.st_ino);
+ free(big);
+ PureDraft *after;
+ CHECK(pure_store_draft(store, "shared", &after) == PURE_OK);
+ expect_draft(after, &(Oracle){(unsigned char *)"two+", 4, 3}, scratch);
+ /* Drafts of the old file still read and commit (their extents are copied). */
+ expect_draft(b, &(Oracle){(unsigned char *)"two+", 4, 3}, scratch);
+ CHECK(pure_draft_splice(b, 0, 0, (PureSpan){(const unsigned char *)">", 1}) == PURE_OK);
+ CHECK(pure_draft_commit(b, &revision) == PURE_OK && revision == 4);
+ expect(path, "shared", &(Oracle){(unsigned char *)">two+", 5, 4});
+ pure_draft_close(after);
+ pure_draft_close(a);
+ pure_draft_close(b);
+ pure_draft_close(state);
+ pure_store_close(store);
+}
+
static void clean(const char *path) {
DIR *entries = opendir(path);
CHECK(entries);
@@ -781,7 +840,7 @@ int main(void) {
void (*checks[])(const char *) = {oracle, amplification, faults, crashes, torn_head,
isolation, corruption, many_keys, revision_limit, keys,
drafts, typing, draft_conflict, draft_relocation, draft_faults,
- draft_versions, anonymous_draft, batch};
+ draft_versions, anonymous_draft, batch, store_handle};
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);
@@ -792,6 +851,6 @@ int main(void) {
puts("{\"tree_store\":\"random-oracle-balance-held-views-one-byte-amplification-faults-"
"process-crash-torn-head-lockless-isolation-compaction-corruption-1024-keys-revision-limit-key-listing-"
"draft-oracle-typing-arena-conflict-relocation-commit-faults-held-versions-anonymous-"
- "atomic-batch-runtime-keys\"}");
+ "atomic-batch-runtime-keys-store-handle\"}");
return 0;
}
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 14:01:05 +0000