aboutsummaryrefslogtreecommitdiffstats
path: root/prototypes/connections
diff options
context:
space:
mode:
Diffstat (limited to 'prototypes/connections')
-rw-r--r--prototypes/connections/module-list.c6
-rw-r--r--prototypes/connections/module-list.h3
-rw-r--r--prototypes/connections/module-save.c17
-rw-r--r--prototypes/connections/module-save.h7
-rw-r--r--prototypes/connections/module-store.c22
-rw-r--r--prototypes/connections/module.c188
-rw-r--r--prototypes/connections/module.h41
-rw-r--r--prototypes/connections/pure-run.c17
-rw-r--r--prototypes/connections/test-module.c141
9 files changed, 344 insertions, 98 deletions
diff --git a/prototypes/connections/module-list.c b/prototypes/connections/module-list.c
index ba67143..374bc78 100644
--- a/prototypes/connections/module-list.c
+++ b/prototypes/connections/module-list.c
@@ -234,9 +234,9 @@ PureCell pure_list_get(const PureList *list, size_t at, unsigned field) {
return list->items[at][field];
}
-void pure_undo_rollback(PureUndo *undo) {
+void pure_undo_rollback(PureUndo *undo, size_t mark) {
Entry *entries = undo->entries;
- for (size_t i = undo->count; i-- > 0;) {
+ for (size_t i = undo->count; i-- > mark;) {
Entry *e = &entries[i];
if (e->change == INSERTED) free_item(e->list, take(e->list, e->at));
else if (e->change == REMOVED) place(e->list, e->at, e->item); /* capacity is still there */
@@ -245,7 +245,7 @@ void pure_undo_rollback(PureUndo *undo) {
e->list->items[e->at][e->field] = e->old;
}
}
- undo->count = 0;
+ undo->count = mark;
}
void pure_undo_commit(PureUndo *undo) {
diff --git a/prototypes/connections/module-list.h b/prototypes/connections/module-list.h
index e79ee46..b568a05 100644
--- a/prototypes/connections/module-list.h
+++ b/prototypes/connections/module-list.h
@@ -35,7 +35,8 @@ void pure_list_mirror(PureList *, PureDraft *, size_t header);
PureResult pure_list_write(PureList *);
/* Append the items stored in bytes; each field is checked. Before mirroring. */
PureResult pure_list_read(PureList *, const unsigned char *bytes, size_t size);
-void pure_undo_rollback(PureUndo *);
+/* Undo the changes logged after mark (a count taken earlier), newest first. */
+void pure_undo_rollback(PureUndo *, size_t mark);
void pure_undo_commit(PureUndo *);
void pure_undo_free(PureUndo *);
#endif
diff --git a/prototypes/connections/module-save.c b/prototypes/connections/module-save.c
index 40b3934..573ac3f 100644
--- a/prototypes/connections/module-save.c
+++ b/prototypes/connections/module-save.c
@@ -244,9 +244,15 @@ PureResult pure_save_open_values(PureApp *app, const PureModule *m, double *valu
static bool same_version(PureVersion a, PureVersion b) { return a.ref == b.ref && a.bytes == b.bytes; }
-PureResult pure_save_commit(PureApp *app, const double *values, void *document, int64_t *revision) {
- if (!pure_save_enabled(app))
- return document ? app->data.commit(app->data.context, document, revision) : PURE_OK;
+PureResult pure_save_commit(PureApp *app, const double *values, void *const *documents, size_t document_count,
+ int64_t *revisions) {
+ if (!pure_save_enabled(app)) {
+ /* Without a host that saves together, a save has one document at most. */
+ PureResult result = PURE_OK;
+ for (size_t i = 0; i < document_count && result == PURE_OK; ++i)
+ result = app->data.commit(app->data.context, documents[i], &revisions[i]);
+ return result;
+ }
const PureModule *m = app->module;
PureDraft *drafts[2 * PURE_MODULE_STATES + 1];
uint32_t states[2 * PURE_MODULE_STATES + 1];
@@ -278,9 +284,8 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *document,
drafts[count++] = app->stored_values;
}
}
- if (!count && !document) return PURE_OK;
- int64_t ignored;
- PureResult result = app->data.save(app->data.context, document, drafts, count, document ? revision : &ignored);
+ if (!count && !document_count) return PURE_OK;
+ PureResult result = app->data.save(app->data.context, documents, document_count, drafts, count, revisions);
for (size_t k = 0; result == PURE_OK && k < count; ++k)
if (states[k] != UINT32_MAX) app->saved[states[k]] = pure_draft_version(drafts[k]);
if (values_changed) {
diff --git a/prototypes/connections/module-save.h b/prototypes/connections/module-save.h
index 90e0701..b6db997 100644
--- a/prototypes/connections/module-save.h
+++ b/prototypes/connections/module-save.h
@@ -19,9 +19,10 @@ PureResult pure_save_open_list(PureApp *, const PureModule *m, uint32_t state, P
PureResult pure_save_open_values(PureApp *, const PureModule *m, double *values, bool load);
/* Store a list whose record changed in its draft, header and items anew. */
PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, PureList *list, PureDraft *draft);
-/* Publish every state changed since the last save (values are the pending
- * numbers), with document when not NULL; revision is the document's. */
-PureResult pure_save_commit(PureApp *, const double *values, void *document, int64_t *revision);
+/* Publish the documents and every state changed since the last save
+ * (values are the states' numbers); revisions are the documents'. */
+PureResult pure_save_commit(PureApp *, const double *values, void *const *documents, size_t document_count,
+ int64_t *revisions);
/* A list with record's fields, taking from `from` each field of the same name
* and kind and giving the others their defaults (module.c). */
diff --git a/prototypes/connections/module-store.c b/prototypes/connections/module-store.c
index 6335903..02d6361 100644
--- a/prototypes/connections/module-store.c
+++ b/prototypes/connections/module-store.c
@@ -42,17 +42,19 @@ static PureResult state(void *context, const char *name, PureDraft **out) {
return pure_state_open(context, name, out);
}
-/* The document and the app's changed states in one tree commit. */
-static PureResult save(void *context, void *document, PureDraft *const *states, size_t count, int64_t *revision) {
+enum { SAVED = PURE_APP_PENDING + 2 * PURE_MODULE_STATES + 1 };
+
+/* The documents and the app's changed states in one tree commit. */
+static PureResult save(void *context, void *const *documents, size_t document_count, PureDraft *const *states,
+ size_t count, int64_t *revisions) {
(void)context;
- PureDraft *drafts[2 * PURE_MODULE_STATES + 2] = {0};
- int64_t revisions[2 * PURE_MODULE_STATES + 2] = {0};
- size_t n = 0;
- if (count > 2 * PURE_MODULE_STATES + 1) return PURE_LIMIT;
- if (document) drafts[n++] = document;
- for (size_t i = 0; i < count; ++i) drafts[n++] = states[i];
- PureResult result = pure_draft_commit_all(drafts, n, revisions);
- if (document) *revision = revisions[0];
+ PureDraft *drafts[SAVED] = {0};
+ int64_t all[SAVED] = {0};
+ if (document_count > PURE_APP_PENDING || count > SAVED - document_count) return PURE_LIMIT;
+ for (size_t i = 0; i < document_count; ++i) drafts[i] = documents[i];
+ for (size_t i = 0; i < count; ++i) drafts[document_count + i] = states[i];
+ PureResult result = pure_draft_commit_all(drafts, document_count + count, all);
+ for (size_t i = 0; i < document_count; ++i) revisions[i] = all[i];
return result;
}
diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c
index 362919b..62f00c7 100644
--- a/prototypes/connections/module.c
+++ b/prototypes/connections/module.c
@@ -1075,13 +1075,14 @@ PureResult pure_app_start(PureApp *app, const PureModule *module, PureData data)
return result;
}
-PureResult pure_app_save(PureApp *app) {
- int64_t revision;
- return pure_save_commit(app, app->values, NULL, &revision);
+static bool pending(const PureApp *app, const void *document) {
+ for (unsigned i = 0; i < app->pending_count; ++i) if (app->pending[i].document == document) return true;
+ return false;
}
void pure_app_finish(PureApp *app) {
- if (app->document) app->data.close(app->data.context, app->document);
+ /* A document in the unsaved burst stays open until the save. */
+ if (app->document && !pending(app, app->document)) app->data.close(app->data.context, app->document);
free(app->copied);
app->document = NULL;
app->copied = NULL;
@@ -1090,8 +1091,75 @@ void pure_app_finish(PureApp *app) {
app->arena.used = 0;
}
+/* The start of a burst: what a failed save returns to. Text and stored list
+ * drafts stay held until the save, so every version since stays readable. */
+static void open_burst(PureApp *app) {
+ if (app->open) return;
+ const PureModule *m = app->module;
+ memcpy(app->durable_values, app->values, sizeof(app->values));
+ app->durable_ticks = app->ticks;
+ app->durable_events = app->events;
+ for (uint32_t i = 0; i < m->states; ++i) {
+ if (app->texts[i]) {
+ pure_draft_hold(app->texts[i], true);
+ app->durable_texts[i] = pure_draft_version(app->texts[i]);
+ }
+ if (app->list_drafts[i]) {
+ pure_draft_hold(app->list_drafts[i], true);
+ app->durable_lists[i] = pure_draft_version(app->list_drafts[i]);
+ }
+ }
+ app->open = true;
+}
+
+/* End a burst, keeping it (saved) or undoing all of it. The documents it held
+ * close, except the current event's, which pure_app_finish closes. */
+static void close_burst(PureApp *app, bool keep) {
+ for (unsigned i = 0; i < app->pending_count; ++i)
+ if (app->pending[i].document != app->document) app->data.close(app->data.context, app->pending[i].document);
+ app->pending_count = 0;
+ if (!app->open) return;
+ const PureModule *m = app->module;
+ for (uint32_t i = 0; i < m->states; ++i) {
+ if (app->texts[i]) {
+ if (!keep) pure_draft_restore(app->texts[i], app->durable_texts[i]);
+ pure_draft_hold(app->texts[i], false);
+ }
+ if (app->list_drafts[i]) {
+ if (!keep) pure_draft_restore(app->list_drafts[i], app->durable_lists[i]);
+ pure_draft_hold(app->list_drafts[i], false);
+ }
+ }
+ if (keep) pure_undo_commit(&app->undo);
+ else {
+ pure_undo_rollback(&app->undo, 0);
+ memcpy(app->values, app->durable_values, sizeof(app->values));
+ app->ticks = app->durable_ticks;
+ app->events = app->durable_events;
+ }
+ app->open = false;
+}
+
+PureResult pure_app_save(PureApp *app) {
+ void *documents[PURE_APP_PENDING];
+ int64_t revisions[PURE_APP_PENDING];
+ size_t count = 0;
+ for (unsigned i = 0; i < app->pending_count; ++i)
+ if (app->pending[i].edited) documents[count++] = app->pending[i].document;
+ PureResult result = pure_save_commit(app, app->values, documents, count, revisions);
+ close_burst(app, result == PURE_OK);
+ return result;
+}
+
+bool pure_app_defer(PureApp *app, bool deferred) {
+ if (deferred && !pure_save_enabled(app)) return false;
+ app->deferred = deferred;
+ return true;
+}
+
void pure_app_stop(PureApp *app) {
pure_app_finish(app);
+ close_burst(app, true); /* unsaved changes are dropped with the app */
for (uint32_t i = 0; i < PURE_MODULE_STATES; ++i) {
pure_draft_close(app->texts[i]);
pure_list_free(app->lists[i]);
@@ -1104,38 +1172,26 @@ void pure_app_stop(PureApp *app) {
memset(app, 0, sizeof(*app));
}
-/* Every text draft and stored list draft is held for the event; on failure
- * each returns to the version it had when the event began. */
-typedef struct { PureVersion before[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; } Texts;
+/* Where one event began inside the burst: a failed event returns there. */
+typedef struct { PureVersion texts[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; size_t undo; bool opened; } Mark;
-static void hold_texts(PureApp *app, Texts *texts) {
+static void mark_event(PureApp *app, Mark *mark) {
+ mark->opened = !app->open;
+ open_burst(app);
for (uint32_t i = 0; i < app->module->states; ++i) {
- if (app->texts[i]) {
- pure_draft_hold(app->texts[i], true);
- texts->before[i] = pure_draft_version(app->texts[i]);
- }
- if (app->list_drafts[i]) {
- pure_draft_hold(app->list_drafts[i], true);
- texts->lists[i] = pure_draft_version(app->list_drafts[i]);
- }
+ if (app->texts[i]) mark->texts[i] = pure_draft_version(app->texts[i]);
+ if (app->list_drafts[i]) mark->lists[i] = pure_draft_version(app->list_drafts[i]);
}
+ mark->undo = app->undo.count;
}
-/* Texts and stored lists return to their versions and lists undo their log,
- * or all keep what the event did. */
-static void settle(PureApp *app, const Texts *texts, bool keep) {
+static void undo_event(PureApp *app, const Mark *mark) {
for (uint32_t i = 0; i < app->module->states; ++i) {
- if (app->texts[i]) {
- if (!keep) pure_draft_restore(app->texts[i], texts->before[i]);
- pure_draft_hold(app->texts[i], false);
- }
- if (app->list_drafts[i]) {
- if (!keep) pure_draft_restore(app->list_drafts[i], texts->lists[i]);
- pure_draft_hold(app->list_drafts[i], false);
- }
+ if (app->texts[i]) pure_draft_restore(app->texts[i], mark->texts[i]);
+ if (app->list_drafts[i]) pure_draft_restore(app->list_drafts[i], mark->lists[i]);
}
- if (keep) pure_undo_commit(&app->undo);
- else pure_undo_rollback(&app->undo);
+ pure_undo_rollback(&app->undo, mark->undo);
+ if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */
}
/* All handlers of one event run in module order against pending states. */
@@ -1152,29 +1208,24 @@ static PureResult run_handlers(PureApp *app, uint32_t event, Event *ev, double *
return PURE_OK;
}
-/* A key or text event stores what it changed before it counts; a frame
- * leaves that to the next event (frames come 60 times a second). */
+/* A key or text event saves before it returns unless saving is deferred; a
+ * frame leaves that to the next event (frames come 60 times a second). */
static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
double pending[PURE_MODULE_STATES];
- Texts texts;
- hold_texts(app, &texts);
+ Mark mark;
+ mark_event(app, &mark);
PureResult result = run_handlers(app, event, ev, pending);
- int64_t revision;
- if (result == PURE_OK && event) result = pure_save_commit(app, pending, NULL, &revision);
- settle(app, &texts, result == PURE_OK);
- if (result == PURE_OK) {
- memcpy(app->values, pending, sizeof(pending));
- ++app->events;
- }
- return result;
+ if (result != PURE_OK) { undo_event(app, &mark); return result; }
+ memcpy(app->values, pending, sizeof(pending));
+ ++app->events;
+ if (!event) ++app->ticks;
+ return event && !app->deferred ? pure_app_save(app) : PURE_OK;
}
PureResult pure_app_frame(PureApp *app) {
pure_app_finish(app);
Event ev = {.app = app, .arena = &app->arena, .dt = 1.0 / 60.0};
- PureResult result = dispatch(app, 0, &ev);
- if (result == PURE_OK) ++app->ticks;
- return result;
+ return dispatch(app, 0, &ev);
}
PureResult pure_app_input(PureApp *app, uint32_t input, PureEdge edge) {
@@ -1220,32 +1271,47 @@ PureResult pure_app_message(PureApp *app, uint32_t input, const char *key, int64
return PURE_INVALID;
const PureInputShape *shape = &m->input[input];
if (payload.size < shape->prefix) return fault(app, PURE_FAULT_LIMIT, UINT32_MAX);
- app->payload = payload;
if (shape->keyed && m->collections) {
if (!app->data.open) return PURE_INVALID;
+ /* A key joins a burst once, so its revision moves once per save. */
+ bool seen = app->pending_count == PURE_APP_PENDING;
+ for (unsigned i = 0; i < app->pending_count && !seen; ++i) seen = !strcmp(app->pending[i].key, key);
+ if (seen) {
+ PureResult saved = pure_app_save(app);
+ if (saved != PURE_OK) return saved;
+ }
PureResult opened = app->data.open(app->data.context, 0, key, &app->document, &app->revision);
- if (opened != PURE_OK) { app->document = NULL; pure_app_finish(app); return opened; }
+ if (opened != PURE_OK) { app->document = NULL; return opened; }
if (base >= 0 && base != app->revision) { pure_app_finish(app); return PURE_CONFLICT; }
}
+ app->payload = payload;
Event ev = {.app = app, .shape = shape, .arena = &app->arena, .exists = app->revision > 0};
- double pending[PURE_MODULE_STATES];
- Texts texts;
- hold_texts(app, &texts);
- PureResult result = run_handlers(app, 1 + 2 * input, &ev, pending);
+ double pending_values[PURE_MODULE_STATES];
+ Mark mark;
+ mark_event(app, &mark);
+ PureResult result = run_handlers(app, 1 + 2 * input, &ev, pending_values);
if (result == PURE_OK && ev.modified && base < 0) result = PURE_INVALID;
- /* Commit rebases the working version and releasing texts may reclaim
+ /* A save rebases an edited document and releasing texts may reclaim
* earlier text versions: keep such a reply by copying it first. */
if (result == PURE_OK && app->has_reply &&
((app->reply.source == FROM_DOCUMENT && ev.modified) || app->reply.source == FROM_TEXT))
result = keep_reply(app);
- if (result == PURE_OK)
- result = pure_save_commit(app, pending, ev.modified ? app->document : NULL, &app->revision);
- /* States, texts and lists change only with a successful commit (or no document edit). */
- settle(app, &texts, result == PURE_OK);
- if (result == PURE_OK) {
- memcpy(app->values, pending, sizeof(pending));
- ++app->events;
- } else if (result != PURE_UNCERTAIN) pure_app_finish(app);
+ if (result != PURE_OK) {
+ undo_event(app, &mark);
+ pure_app_finish(app);
+ return result;
+ }
+ memcpy(app->values, pending_values, sizeof(pending_values));
+ ++app->events;
+ if (app->document) {
+ app->pending[app->pending_count].document = app->document;
+ app->pending[app->pending_count].edited = ev.modified;
+ memcpy(app->pending[app->pending_count++].key, key, strlen(key) + 1);
+ app->revision += ev.modified; /* a save moves an edited document one revision */
+ }
+ if (app->deferred) return PURE_OK;
+ result = pure_app_save(app);
+ if (result != PURE_OK) pure_app_finish(app);
return result;
}
@@ -1359,6 +1425,10 @@ static PureResult keep_list(PureList **out, PureList *from, const PureModule *ol
* a state the new module drops stays stored as it was. */
PureResult pure_app_replace(PureApp *app, const PureModule *module) {
pure_app_finish(app);
+ if (app->open || app->pending_count) {
+ PureResult saved = pure_app_save(app); /* the burst belongs to the module that made it */
+ if (saved != PURE_OK) return saved;
+ }
const PureModule *old = app->module;
bool stored = pure_save_enabled(app);
double values[PURE_MODULE_STATES] = {0};
diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h
index d2d4b7a..c5d4f52 100644
--- a/prototypes/connections/module.h
+++ b/prototypes/connections/module.h
@@ -19,7 +19,7 @@ enum {
PURE_MODULE_NAME = 32, PURE_MODULE_STRINGS = 32768, PURE_MODULE_HEADER = 40,
PURE_MODULE_FIELDS = 32, PURE_MODULE_COLLECTIONS = 1, PURE_MODULE_CONSTANTS = 64,
PURE_MODULE_CONSTANT_BYTES = 8192, PURE_MODULE_RECORDS = 8, PURE_MODULE_RECORD_FIELDS = 64,
- PURE_RECORD_FIELDS = 16, PURE_MODULE_JOINS = 8
+ PURE_RECORD_FIELDS = 16, PURE_MODULE_JOINS = 8, PURE_APP_PENDING = 16
};
typedef enum {
@@ -123,10 +123,11 @@ typedef struct {
void (*close)(void *context, void *document);
/* Optional: the host keeps the app's states. state opens a runtime-owned
* value by name as a draft (a missing one is empty at revision 0); save
- * publishes the given state drafts together with document (or NULL),
- * all or nothing, and names the document's new revision. */
+ * publishes the given documents and state drafts together, all or
+ * nothing, and names each document's new revision. */
PureResult (*state)(void *context, const char *name, PureDraft **out);
- PureResult (*save)(void *context, void *document, PureDraft *const *states, size_t count, int64_t *revision);
+ PureResult (*save)(void *context, void *const *documents, size_t document_count, PureDraft *const *states,
+ size_t count, int64_t *revisions);
} PureData;
/* A byte or text value inside one event or view: a range of the payload, a
@@ -138,10 +139,15 @@ typedef struct { unsigned char *bytes; size_t used, capacity; } PureArena;
* runtime-owned drafts; cursors are positions the runtime keeps on character
* boundaries; lists are runtime-owned and an index always names an item
* (or 0 in an empty list). Every event is atomic: on failure nothing changes.
- * When the data host keeps states, the app starts from the stored ones and
- * every key, text and message event stores what it changed with its own
- * commit (with the message's document); frame events are stored with the
- * next one, or by pure_app_save. */
+ *
+ * Saving. When the data host keeps states the app starts from the stored
+ * ones. A save publishes, in one commit, the documents edited by messages and
+ * the states changed since the last save; if it fails, every event since the
+ * last save is undone. By default each key, text and message event saves
+ * before it returns (a frame waits for the next one). Deferred, events apply
+ * at once and a host saves a burst with one commit: an event is durable once
+ * a later save succeeds, and each key's document joins a burst once (a second
+ * message for it saves the burst first). */
typedef struct {
const PureModule *module;
PureData data;
@@ -152,6 +158,14 @@ typedef struct {
PureDraft *list_drafts[PURE_MODULE_STATES]; /* stored lists, when kept */
PureDraft *stored_values; /* numbers, cursors and indices, when kept */
PureVersion saved[PURE_MODULE_STATES]; /* text and list versions at the last save */
+ /* Events since the last save: what a failed save returns to, and the
+ * documents messages opened (held until the save). */
+ bool deferred, open;
+ double durable_values[PURE_MODULE_STATES];
+ PureVersion durable_texts[PURE_MODULE_STATES], durable_lists[PURE_MODULE_STATES];
+ uint64_t durable_ticks, durable_events;
+ struct { void *document; char key[65]; bool edited; } pending[PURE_APP_PENDING];
+ unsigned pending_count;
PureFault fault;
uint32_t fault_pc;
uint64_t ticks, events;
@@ -166,8 +180,11 @@ typedef struct {
} PureApp;
PureResult pure_app_start(PureApp *, const PureModule *, PureData data);
-/* Store every state changed since the last save (after frame events). */
+/* Publish every change since the last save in one commit, or undo them all. */
PureResult pure_app_save(PureApp *);
+/* Defer saving to pure_app_save (false: save after each event). Needs a data
+ * host that saves documents together (PureData.save). */
+bool pure_app_defer(PureApp *, bool deferred);
/* Leaves unsaved frame changes unsaved: call pure_app_save first to keep them. */
void pure_app_stop(PureApp *);
PureResult pure_app_frame(PureApp *);
@@ -176,9 +193,11 @@ PureResult pure_app_input(PureApp *, uint32_t input, PureEdge);
PureResult pure_app_text(PureApp *, uint32_t input, PureSpan text);
/* Deliver a message. base is the revision the sender edited, or -1 for none;
* a handler that edits the document needs one. payload stays alive until
- * pure_app_finish. On return app->revision is the document's revision. */
+ * pure_app_finish. On return app->revision is the document's revision
+ * (deferred: the revision the next save gives it). */
PureResult pure_app_message(PureApp *, uint32_t input, const char *key, int64_t base, PureSpan payload);
-/* Visit the reply of the last message in order, without copying. */
+/* Visit the reply of the last message in order, without copying, until the
+ * next event (deferred: copy it before a later save too). */
PureResult pure_app_reply(const PureApp *, bool (*visit)(void *, PureSpan), void *context);
void pure_app_finish(PureApp *);
/* A number view; false for a text view or a failed evaluation. */
diff --git a/prototypes/connections/pure-run.c b/prototypes/connections/pure-run.c
index 7b6e253..6e4a6e1 100644
--- a/prototypes/connections/pure-run.c
+++ b/prototypes/connections/pure-run.c
@@ -8,7 +8,8 @@
/* Test host for a verified module: reads commands from stdin and prints one
* JSON snapshot per command. Not a product interface. With --store the app's
* states live in that tree store: it starts from them, and frame changes are
- * saved at the end of input or by save.
+ * saved at the end of input or by save. --batch N defers saving and saves
+ * every N commands (and at the end), so N events share one commit.
* down NAME | up NAME | frame [N] | type NAME HEX|- | replace PATH | save | show
* Text states and text views print as JSON strings; cursors as byte positions;
* lists as arrays of objects; indices as positions; a table view as an array
@@ -134,9 +135,11 @@ static int hex(char c) {
int main(int argc, char **argv) {
static PureModule modules[2];
- bool stored = argc == 4 && !strcmp(argv[2], "--store");
- if ((argc != 2 && !stored) || !load(argv[1], &modules[0])) {
- if (argc != 2 && !stored) fputs("pure-run MODULE [--store PATH] < commands\n", stderr);
+ bool stored = argc >= 4 && !strcmp(argv[2], "--store");
+ long batch = argc == 6 && stored && !strcmp(argv[4], "--batch") ? strtol(argv[5], NULL, 10) : 0;
+ bool usage = argc != 2 && !(stored && argc == 4) && !(batch > 0 && batch <= 100000);
+ if (usage || !load(argv[1], &modules[0])) {
+ if (usage) fputs("pure-run MODULE [--store PATH [--batch N]] < commands\n", stderr);
return 2;
}
if (stored && pure_store_init(argv[3]) != PURE_OK) { fprintf(stderr, "cannot open store %s\n", argv[3]); return 2; }
@@ -144,6 +147,8 @@ int main(int argc, char **argv) {
PureApp app;
PureResult started = pure_app_start(&app, &modules[active], stored ? pure_tree_data(argv[3]) : (PureData){0});
if (started != PURE_OK) { fprintf(stderr, "cannot start (%d)\n", (int)started); return 2; }
+ if (batch && !pure_app_defer(&app, true)) { fputs("cannot defer saving\n", stderr); return 2; }
+ long unsaved = 0;
static char line[1 << 16];
static unsigned char text[1 << 15];
int status = 0;
@@ -176,6 +181,10 @@ int main(int argc, char **argv) {
} else if (!strcmp(command, "save")) {
if (pure_app_save(&app) != PURE_OK) failure = "save failed";
} else if (strcmp(command, "show")) failure = "unknown command";
+ if (batch && strcmp(command, "show") && ++unsaved == batch) {
+ unsaved = 0;
+ if (pure_app_save(&app) != PURE_OK) failure = "save failed";
+ }
if (failure) status = 1;
show(&app, failure);
}
diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c
index 64409b2..e1d1ccf 100644
--- a/prototypes/connections/test-module.c
+++ b/prototypes/connections/test-module.c
@@ -1574,6 +1574,144 @@ static void stored_states(void) {
stored_migration();
}
+/* ---- Deferred saving ---------------------------------------------------- */
+
+typedef struct { uint32_t input; const char *word; } Step;
+
+static PureResult apply_step(PureApp *app, Step step) {
+ if (step.word) return pure_app_text(app, step.input, (PureSpan){(const unsigned char *)step.word, strlen(step.word)});
+ return pure_app_input(app, step.input, PURE_EDGE_DOWN);
+}
+
+/* Deferred events apply at once and last only through a save: an app that
+ * stops without saving comes back as it was at the last save. The oracle is
+ * an app in memory that replays exactly the saved events. */
+static void deferred_equivalence(Spec spec, const char *const *words, size_t word_count, uint32_t seed) {
+ char *path = new_store();
+ CHECK(load(spec, &module_a) == PURE_OK);
+ static Step history[2000];
+ size_t saved = 0, applied = 0;
+ PureApp memory, stored;
+ CHECK(pure_app_start(&memory, &module_a, (PureData){0}) == PURE_OK);
+ CHECK(pure_app_start(&stored, &module_a, pure_tree_data(path)) == PURE_OK);
+ CHECK(pure_app_defer(&stored, true) && !pure_app_defer(&memory, true));
+ unsigned lost = 0, saves = 0;
+ for (unsigned step = 0; step < 600; ++step) {
+ Step next = {next_random(&seed) % module_a.inputs, NULL};
+ if (module_a.input[next.input].kind == PURE_KIND_TEXT) next.word = words[next_random(&seed) % word_count];
+ PureResult a = apply_step(&memory, next), b = apply_step(&stored, next);
+ CHECK(a == b);
+ if (a == PURE_OK) history[applied++] = next;
+ same_states(&memory, &stored);
+ uint32_t roll = next_random(&seed) % 16;
+ if (roll < 3) {
+ CHECK(pure_app_save(&stored) == PURE_OK);
+ saved = applied;
+ ++saves;
+ } else if (roll == 3) {
+ /* Stop without saving: only saved events come back. */
+ pure_app_stop(&stored);
+ pure_app_stop(&memory);
+ CHECK(pure_app_start(&stored, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&stored, true));
+ CHECK(pure_app_start(&memory, &module_a, (PureData){0}) == PURE_OK);
+ for (size_t k = 0; k < saved; ++k) CHECK(apply_step(&memory, history[k]) == PURE_OK);
+ lost += (unsigned)(applied - saved);
+ applied = saved;
+ same_states(&memory, &stored);
+ }
+ }
+ CHECK(saves > 50 && lost > 0);
+ pure_app_stop(&memory);
+ pure_app_stop(&stored);
+ remove_store(path);
+}
+
+/* A burst of events is saved with one commit: when that commit fails every
+ * event of the burst is undone, and a process that dies while saving comes
+ * back with all of the burst or none of it. */
+static void deferred_bursts(void) {
+ char *path = new_store();
+ Spec spec = {crash_nums, 1, crash_inputs, 1, NULL, 0, crash_handlers, 1, crash_code, LEN(crash_code), NULL, 0,
+ crash_states, 4, crash_constants, 2, NULL, 0, crash_records, 1, NULL, 0};
+ CHECK(load(spec, &module_a) == PURE_OK);
+ PureApp app;
+ CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&app, true));
+ for (int i = 0; i < 5; ++i) CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK);
+ CHECK(consistent(&app) == 5 && app.events == 5);
+ for (fail_stage = PURE_AFTER_HEADER; fail_stage <= PURE_BEFORE_RENAME; fail_stage += 2) {
+ CHECK(pure_app_save(&app) == PURE_STORE_ERROR && consistent(&app) == 0 && app.events == 0);
+ for (int i = 0; i < 5; ++i) CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK);
+ }
+ fail_stage = -1;
+ CHECK(pure_app_save(&app) == PURE_OK && consistent(&app) == 5);
+ pure_app_stop(&app);
+ CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && consistent(&app) == 5);
+ pure_app_stop(&app);
+ unsigned expected = 5;
+ for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) {
+ pid_t child = fork();
+ CHECK(child >= 0);
+ if (!child) {
+ PureApp doomed;
+ if (pure_app_start(&doomed, &module_a, pure_tree_data(path)) != PURE_OK || !pure_app_defer(&doomed, true)) _exit(1);
+ for (int i = 0; i < 7; ++i) if (pure_app_input(&doomed, 0, PURE_EDGE_DOWN) != PURE_OK) _exit(2);
+ crash = 1;
+ fail_stage = stage;
+ pure_app_save(&doomed);
+ _exit(0);
+ }
+ int status;
+ CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77);
+ if (stage == PURE_BEFORE_DIR_SYNC) expected += 7;
+ CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && consistent(&app) == expected);
+ pure_app_stop(&app);
+ }
+ remove_store(path);
+}
+
+/* Messages in a burst: documents of different keys commit together with the
+ * states, each moving one revision; a second message for a key saves the
+ * burst first, and so does a full burst. */
+static void deferred_messages(void) {
+ char *path = new_store();
+ CHECK(load(editor_spec, &module_a) == PURE_OK);
+ PureApp app;
+ unsigned char buffer[64];
+ char key[8];
+ CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&app, true));
+ for (int i = 0; i < 3; ++i) {
+ snprintf(key, sizeof(key), "k%d", i);
+ CHECK(pure_app_message(&app, 1, key, 0, edit_payload(buffer, 0, 0, "ab")) == PURE_OK && app.revision == 1);
+ }
+ CHECK(app.values[0] == 3 && app.pending_count == 3 && app.open);
+ /* The same key again: the burst is saved first, then this edit joins a new one. */
+ CHECK(pure_app_message(&app, 1, "k0", 1, edit_payload(buffer, 2, 0, "c")) == PURE_OK && app.revision == 2);
+ CHECK(app.pending_count == 1 && app.values[0] == 4);
+ CHECK(pure_app_message(&app, 1, "k1", 0, edit_payload(buffer, 0, 0, "x")) == PURE_CONFLICT);
+ CHECK(pure_app_save(&app) == PURE_OK && !app.open && !app.pending_count);
+ for (int i = 0; i < PURE_APP_PENDING + 1; ++i) {
+ snprintf(key, sizeof(key), "m%d", i);
+ CHECK(pure_app_message(&app, 1, key, 0, edit_payload(buffer, 0, 0, "y")) == PURE_OK);
+ CHECK(app.pending_count == (i < PURE_APP_PENDING ? (unsigned)i + 1 : 1)); /* full: saved on the way */
+ }
+ pure_app_stop(&app); /* the last edit was never saved */
+ CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && app.values[0] == 4 + PURE_APP_PENDING);
+ CHECK(pure_app_message(&app, 0, "k0", -1, (PureSpan){0}) == PURE_OK && app.revision == 2 && replied(&app, "abc"));
+ CHECK(pure_app_message(&app, 0, "m16", -1, (PureSpan){0}) == PURE_NOT_FOUND);
+ CHECK(pure_app_message(&app, 0, "m15", -1, (PureSpan){0}) == PURE_OK && replied(&app, "y"));
+ pure_app_stop(&app);
+ remove_store(path);
+}
+
+static void deferred(void) {
+ static const char *const list_words[] = {"a", "\xea\xb0\x80", "", "\n", "\xf0\x9f\x98\x80"};
+ static const char *const text_words[] = {"a", "\xea\xb0\x80", "\n", "\xf0\x9f\x98\x80", "xy\nz", ""};
+ deferred_equivalence(list_spec, list_words, LEN(list_words), 0x2545f491u);
+ deferred_equivalence(text_spec, text_words, LEN(text_words), 0x1b873593u);
+ deferred_bursts();
+ deferred_messages();
+}
+
int main(void) {
behaviour();
faults();
@@ -1588,7 +1726,8 @@ int main(void) {
tables();
table_mutations();
stored_states();
+ deferred();
printf("{\"sizeof_module\":%zu,\"sizeof_app\":%zu}\n", sizeof(PureModule), sizeof(PureApp));
- puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration\"}");
+ puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts\"}");
return 0;
}
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 13:36:41 +0000