aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authoreunhhu <[email protected]>2026-09-28 19:08:09 +0100
committereunhhu <[email protected]>2026-09-28 19:08:09 +0100
commitec210a8aed62c301e9ae384d4d157e379fa58f54 (patch)
tree49ce69c1ffcdb50a9d9ccf9d6cb8fcbf46e34a5d
parent745efcd4f19b59bcfb131df2d127a3387a544e84 (diff)
downloadpure-ec210a8aed62c301e9ae384d4d157e379fa58f54.tar.gz
pure-ec210a8aed62c301e9ae384d4d157e379fa58f54.zip
Keep the numbers heard as a list: a save logs only the ones it changed
The last number heard from each source was written whole into the links value at every save, and counted in two bytes: a node that heard from more than 65,535 sources stored a wrong count, and each save cost as many bytes as there were sources. The numbers are now a runtime-owned list {source, seq} in order of source, stored as a list state is (its image "0heard" and its changes in the log), changed with the event that delivers, so a failed save takes them back with it. Links stored before (PLK1) move their numbers to the list at a start and are stored again without them (PLK2).
-rw-r--r--core/module-save.c191
-rw-r--r--core/module-save.h17
-rw-r--r--core/module.c112
-rw-r--r--core/module.h32
-rw-r--r--core/test-module.c61
5 files changed, 290 insertions, 123 deletions
diff --git a/core/module-save.c b/core/module-save.c
index ddc2797..968e65b 100644
--- a/core/module-save.c
+++ b/core/module-save.c
@@ -42,12 +42,21 @@ unsigned long pure_save_test_checkpoints;
enum { JOURNAL_MIN = 1 << 20 };
#define LOG_KEPT (UINT64_C(64) << 20)
#endif
-/* links: "PLK1"; u8 outputs, each {u8 name length, name, u8 field count, the
+/* links: "PLK2"; u8 outputs, each {u8 name length, name, u8 field count, the
* fields' types, u64le last number} (the module's outputs, then those it
- * dropped, without fields); u16le sources, each {u8 length, the name, u64le
- * last number heard}; then the sent messages to the end, each {u8 output (in
- * the table), u64le number, u32le size, payload}. */
-static const unsigned char links_magic[4] = {'P', 'L', 'K', '1'};
+ * dropped, without fields); then the sent messages to the end, each {u8
+ * output (in the table), u64le number, u32le size, payload}. Before the
+ * heard list, "PLK1" kept the numbers heard between the two: u16le sources,
+ * each {u8 length, the name, u64le last number heard}; a start moves them to
+ * the heard list.
+ *
+ * The heard list (module.h, PureApp) is stored as a list state is, by a name
+ * no state has (a state's is an identifier), with the fields source (text)
+ * and seq (number): the numbers of every source, which only grow with the
+ * sources a node hears from, cost a save only the records of their changes. */
+static const unsigned char links_magic[4] = {'P', 'L', 'K', '2'}, links_magic_1[4] = {'P', 'L', 'K', '1'};
+static const char heard_name[] = "0heard", heard_names[2][PURE_MODULE_NAME] = {"source", "seq"};
+static const uint8_t heard_text[2] = {1, 0};
static const PureVersion unsaved = {UINT64_MAX, UINT64_MAX};
static unsigned char *put_le(unsigned char *at, uint64_t value, unsigned width) {
@@ -135,22 +144,38 @@ PURE_COLD PureResult pure_save_open_text(PureApp *app, const PureModule *m, uint
enum { IMAGE_HEADER = 5 + PURE_RECORD_FIELDS * (2 + PURE_MODULE_NAME) + 8 };
-static size_t image_header(const PureModule *m, uint32_t record, uint64_t seq, unsigned char *out) {
- const PureRecord *r = &m->record[record];
+static size_t header_of(unsigned count, const uint8_t *text, const char (*names)[PURE_MODULE_NAME], uint64_t seq,
+ unsigned char *out) {
size_t at = 0;
memcpy(out, image_magic, 4);
at = 4;
- out[at++] = (unsigned char)r->count;
- for (unsigned f = 0; f < r->count; ++f) {
- const char *name = m->record_field_names[r->first + f];
- out[at++] = m->record_field_text[r->first + f];
- out[at++] = (unsigned char)strlen(name);
- memcpy(out + at, name, strlen(name));
- at += strlen(name);
+ out[at++] = (unsigned char)count;
+ for (unsigned f = 0; f < count; ++f) {
+ out[at++] = text[f];
+ out[at++] = (unsigned char)strlen(names[f]);
+ memcpy(out + at, names[f], strlen(names[f]));
+ at += strlen(names[f]);
}
return (size_t)(put_le(out + at, seq, 8) - out);
}
+static size_t image_header(const PureModule *m, uint32_t record, uint64_t seq, unsigned char *out) {
+ const PureRecord *r = &m->record[record];
+ return header_of(r->count, m->record_field_text + r->first, m->record_field_names + r->first, seq, out);
+}
+
+PureList *pure_heard_new(void) { return pure_list_new(2, heard_text); }
+
+/* A stored list by its place: a list state, or the heard list (PURE_LIST_HEARD). */
+static PureList *stored_list(const PureApp *app, uint32_t i) {
+ if (i == PURE_LIST_HEARD) return app->heard;
+ return i < app->module->states && app->module->state_type[i] == PURE_STATE_LIST ? app->lists[i] : NULL;
+}
+
+static const char *stored_name(const PureApp *app, uint32_t i) {
+ return i == PURE_LIST_HEARD ? heard_name : app->module->state_names[i];
+}
+
static const char *directory_of(const PureApp *app) {
return app->data.directory ? app->data.directory(app->data.context) : NULL;
}
@@ -257,7 +282,7 @@ PURE_COLD PureResult pure_save_open_log(PureApp *app) {
void pure_save_release(PureApp *app) {
if (!app->log) return;
uint64_t kept = app->seq + 1;
- for (uint32_t i = 0; i < PURE_MODULE_STATES; ++i)
+ for (uint32_t i = 0; i <= PURE_LIST_HEARD; ++i)
if (app->logged_since[i] && app->logged_since[i] < kept) kept = app->logged_since[i];
pure_log_release(app->log, kept - 1);
}
@@ -408,6 +433,49 @@ PURE_COLD PureResult pure_save_open_list(PureApp *app, const PureModule *m, uint
return result;
}
+PURE_COLD PureResult pure_save_open_heard(PureApp *app, PureList **out, PureSaved *saved) {
+ *saved = (PureSaved){0};
+ *out = NULL;
+ if (!directory_of(app)) return PURE_STORE_ERROR;
+ unsigned char *bytes = NULL, header[IMAGE_HEADER];
+ size_t size = 0, length = header_of(2, heard_text, heard_names, 0, header) - 8;
+ PureResult result = pure_image_read(directory_of(app), heard_name, &bytes, &size);
+ bool image = result == PURE_OK;
+ uint64_t holds = image && size >= length + 8 && !memcmp(bytes, header, length) ? get_le(bytes + length, 8) : 0;
+ if (image && (size < length + 8 || memcmp(bytes, header, length))) result = PURE_STORE_ERROR;
+ else if (image || result == PURE_NOT_FOUND) {
+ result = (*out = pure_heard_new()) ? PURE_OK : PURE_ALLOCATION;
+ if (result == PURE_OK && image) result = pure_list_read(*out, bytes + length + 8, size - length - 8);
+ }
+ free(bytes);
+ /* Then the changes logged after the save it holds (an image is written
+ * before any change of the list is logged). */
+ struct Named *logged = app->start_log ? named_list(app->start_log, heard_name, strlen(heard_name), false) : NULL;
+ for (size_t k = 0; result == PURE_OK && logged && k < logged->count; ++k) {
+ const Logged *change = &logged->changes[k];
+ if (change->seq <= holds) continue;
+ if (!image) result = PURE_STORE_ERROR;
+ else if ((result = pure_list_replay(*out, change->records.data, change->records.size)) == PURE_OK) {
+ if (!saved->since) saved->since = change->seq;
+ saved->logged += change->records.size;
+ }
+ }
+ if (result != PURE_OK) {
+ pure_list_free(*out);
+ *out = NULL;
+ return result;
+ }
+ if (image) saved->image = holds;
+ else saved->rewrite = true;
+ return PURE_OK;
+}
+
+PureResult pure_save_heard_image(PureApp *app) {
+ unsigned char header[IMAGE_HEADER];
+ size_t header_size = header_of(2, heard_text, heard_names, app->seq, header);
+ return write_image(directory_of(app), heard_name, header, header_size, app->heard);
+}
+
/* ---- Checkpoints */
struct PureCheckpoint {
@@ -425,32 +493,33 @@ PureCheckpoint *pure_app_checkpoint(PureApp *app) {
const PureModule *m = app->module;
if (!app->log || app->checkpoint || app->open || app->uncertain || app->reading) return NULL;
/* A list whose logged changes outgrew its items, or, when the log grew
- * past LOG_KEPT, the one that holds it back most. */
+ * past LOG_KEPT, the one that holds it back most: the heard list is one. */
int due = -1, oldest = -1;
- for (uint32_t i = 0; i < m->states && due < 0; ++i) {
- if (m->state_type[i] != PURE_STATE_LIST || !app->logged_since[i]) continue;
- size_t items = pure_list_size(app->lists[i]);
- if (app->logged[i] > (items > JOURNAL_MIN ? items : JOURNAL_MIN)) due = (int)i;
+ uint64_t items = 0;
+ for (uint32_t i = 0; i <= PURE_LIST_HEARD && due < 0; ++i) {
+ const PureList *list = stored_list(app, i);
+ size_t size = list ? pure_list_size(list) : 0;
+ items += size;
+ if (!list || !app->logged_since[i]) continue;
+ if (app->logged[i] > (size > JOURNAL_MIN ? size : JOURNAL_MIN)) due = (int)i;
else if (oldest < 0 || app->logged_since[i] < app->logged_since[oldest]) oldest = (int)i;
}
- uint64_t items = 0;
- for (uint32_t i = 0; due < 0 && i < m->states; ++i)
- if (m->state_type[i] == PURE_STATE_LIST && app->lists[i]) items += pure_list_size(app->lists[i]);
int pick = due >= 0 ? due : pure_log_bytes(app->log) > (items > LOG_KEPT ? items : LOG_KEPT) ? oldest : -1;
if (pick < 0) return NULL;
PureCheckpoint *c = calloc(1, sizeof(*c));
const char *directory = directory_of(app);
size_t length = strlen(directory) + 1;
if (c && (c->directory = malloc(length))) memcpy(c->directory, directory, length);
- if (!c || !c->directory || !(c->snapshot = pure_app_snapshot(app))) {
+ if (!c || !c->directory || !(c->snapshot = pure_save_snapshot(app, pick == PURE_LIST_HEARD))) {
if (c) free(c->directory);
free(c);
return NULL;
}
c->state = (uint32_t)pick;
- snprintf(c->name, sizeof(c->name), "%s", m->state_names[pick]);
+ snprintf(c->name, sizeof(c->name), "%s", stored_name(app, (uint32_t)pick));
c->seq = app->seq;
- c->header_size = image_header(m, m->state_ref[pick], c->seq, c->header);
+ c->header_size = pick == PURE_LIST_HEARD ? header_of(2, heard_text, heard_names, c->seq, c->header) :
+ image_header(m, m->state_ref[pick], c->seq, c->header);
c->logged = app->logged[pick];
c->since = app->logged_since[pick];
app->checkpoint = c;
@@ -563,14 +632,13 @@ static unsigned char *encode_links(const PureApp *app, size_t *size) {
const PureModule *m = app->module;
unsigned outputs = 0;
uint8_t table[PURE_MODULE_INPUTS];
- size_t total = 4 + 1 + 2;
+ size_t total = 4 + 1;
for (uint32_t i = 0; i < m->inputs; ++i)
if (m->input[i].kind == PURE_KIND_OUTPUT) {
table[i] = (uint8_t)outputs++;
total += 2 + strlen(m->input_names[i]) + m->input[i].field_count + 8;
}
for (unsigned c = 0; c < app->carried_count; ++c) total += 2 + strlen(app->carried[c].name) + 8;
- for (unsigned h = 0; h < app->heard_count; ++h) total += 1 + strlen(app->heard[h].source) + 8;
for (size_t s = 0; s < app->sent_count; ++s)
if (!app->sent[s].trimmed) total += 1 + 8 + 4 + app->sent[s].size;
unsigned char *bytes = malloc(total), *at = bytes;
@@ -596,13 +664,6 @@ static unsigned char *encode_links(const PureApp *app, size_t *size) {
*at++ = 0;
at = put_le(at, app->carried[c].last, 8);
}
- at = put_le(at, app->heard_count, 2);
- for (unsigned h = 0; h < app->heard_count; ++h) {
- size_t length = strlen(app->heard[h].source);
- *at++ = (unsigned char)length;
- memcpy(at, app->heard[h].source, length);
- at = put_le(at + length, app->heard[h].seq, 8);
- }
for (size_t s = 0; s < app->sent_count; ++s) {
const struct PureSent *sent = &app->sent[s];
if (sent->trimmed) continue;
@@ -616,11 +677,22 @@ static unsigned char *encode_links(const PureApp *app, size_t *size) {
return bytes;
}
+/* A number heard, as links kept it before the heard list. */
+PURE_COLD static PureResult heard_before(PureApp *app, const unsigned char *source, size_t length, uint64_t seq) {
+ PureCell key = {.text = (unsigned char *)source, .size = length};
+ bool found;
+ size_t at = pure_list_seek(app->heard, 0, key, &found);
+ if (!found) return pure_list_insert(app->heard, at, (PureCell[2]){key, {.num = (double)seq}}, NULL);
+ if (pure_list_get(app->heard, at, 1).num >= (double)seq) return PURE_OK;
+ return pure_list_set(app->heard, at, 1, (PureCell){.num = (double)seq}, NULL);
+}
+
/* The stored links of a starting app: outputs by name (an output whose
* fields changed or that the module lacks refuses the start while it has
* sends left: CONFLICT; one without sends keeps its number for later),
- * sources and sent messages. */
-PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
+ * sources kept before the heard list, and sent messages. */
+PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, bool *moved) {
+ *moved = false;
PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links);
if (result != PURE_OK || !pure_draft_revision(app->stored_links)) return result;
unsigned char *bytes;
@@ -629,8 +701,9 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
int output[256];
bool fits[256];
size_t at = 5;
- unsigned outputs = size >= 5 && !memcmp(bytes, links_magic, 4) ? bytes[4] : 0;
- result = size >= 5 && !memcmp(bytes, links_magic, 4) ? PURE_OK : PURE_STORE_ERROR;
+ bool first = size >= 5 && !memcmp(bytes, links_magic_1, 4), known = first || (size >= 5 && !memcmp(bytes, links_magic, 4));
+ unsigned outputs = known ? bytes[4] : 0;
+ result = known ? PURE_OK : PURE_STORE_ERROR;
for (unsigned t = 0; t < outputs && result == PURE_OK; ++t) {
char name[PURE_MODULE_NAME];
size_t length = at < size ? bytes[at] : 0;
@@ -655,18 +728,18 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
grown[app->carried_count++].last = last;
}
}
- unsigned sources = result == PURE_OK && size - at >= 2 ? (unsigned)get_le(bytes + at, 2) : 0;
- if (result == PURE_OK && size - at < 2) result = PURE_STORE_ERROR;
- at += 2;
- if (result == PURE_OK && sources && !(app->heard = calloc(sources, sizeof(*app->heard)))) result = PURE_ALLOCATION;
- app->heard_capacity = sources;
+ unsigned sources = result == PURE_OK && first && size - at >= 2 ? (unsigned)get_le(bytes + at, 2) : 0;
+ if (result == PURE_OK && first && size - at < 2) result = PURE_STORE_ERROR;
+ at += first ? 2 : 0;
for (unsigned h = 0; h < sources && result == PURE_OK; ++h) {
size_t length = at < size ? bytes[at] : 0;
if (!length || length > PURE_SOURCE || size - at < 1 + length + 8) { result = PURE_STORE_ERROR; break; }
- struct PureHeard *heard = &app->heard[app->heard_count++];
- memcpy(heard->source, bytes + at + 1, length);
- heard->seq = heard->saved = get_le(bytes + at + 1 + length, 8);
+ uint64_t seq = get_le(bytes + at + 1 + length, 8);
+ if (seq > PURE_SEQ_MOST) { result = PURE_STORE_ERROR; break; }
+ result = heard_before(app, bytes + at + 1, length, seq);
at += 1 + length + 8;
+ *moved = true;
+ app->links_changed = true; /* stored again without them */
}
while (result == PURE_OK && at < size) {
if (size - at < 13 || bytes[at] >= outputs) { result = PURE_STORE_ERROR; break; }
@@ -698,18 +771,19 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
/* The lists' changes since the last save, as one log entry: per list changed,
* its name and its records. NULL with size 0 when no list changed. */
static unsigned char *list_changes(const PureApp *app, size_t *size) {
- const PureModule *m = app->module;
*size = 0;
- for (uint32_t i = 0; i < m->states; ++i)
- if (m->state_type[i] == PURE_STATE_LIST && app->lists[i] && pure_list_changes(app->lists[i]).size)
- *size += 1 + strlen(m->state_names[i]) + 4 + pure_list_changes(app->lists[i]).size;
+ for (uint32_t i = 0; i <= PURE_LIST_HEARD; ++i) {
+ const PureList *list = stored_list(app, i);
+ if (list && pure_list_changes(list).size) *size += 1 + strlen(stored_name(app, i)) + 4 + pure_list_changes(list).size;
+ }
unsigned char *payload = *size ? malloc(*size) : NULL, *at = payload;
- for (uint32_t i = 0; payload && i < m->states; ++i) {
- PureSpan changes = m->state_type[i] == PURE_STATE_LIST && app->lists[i] ? pure_list_changes(app->lists[i]) : (PureSpan){0};
+ for (uint32_t i = 0; payload && i <= PURE_LIST_HEARD; ++i) {
+ const PureList *list = stored_list(app, i);
+ PureSpan changes = list ? pure_list_changes(list) : (PureSpan){0};
if (!changes.size) continue;
- size_t length = strlen(m->state_names[i]);
+ size_t length = strlen(stored_name(app, i));
*at++ = (unsigned char)length;
- memcpy(at, m->state_names[i], length);
+ memcpy(at, stored_name(app, i), length);
at = put_le(at + length, changes.size, 4);
memcpy(at, changes.data, changes.size);
at += changes.size;
@@ -814,12 +888,13 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc
if (result == PURE_OK) {
for (size_t k = 0; k < count; ++k)
if (states[k] != UINT32_MAX) app->saved[states[k]] = pure_draft_version(drafts[k]);
- for (uint32_t i = 0; i < m->states; ++i) {
- PureSpan changes = m->state_type[i] == PURE_STATE_LIST && app->lists[i] ? pure_list_changes(app->lists[i]) : (PureSpan){0};
+ for (uint32_t i = 0; i <= PURE_LIST_HEARD; ++i) {
+ PureList *list = stored_list(app, i);
+ PureSpan changes = list ? pure_list_changes(list) : (PureSpan){0};
if (!changes.size) continue;
app->logged[i] += changes.size;
if (!app->logged_since[i]) app->logged_since[i] = seq;
- pure_list_forget(app->lists[i], 0);
+ pure_list_forget(list, 0);
}
if (tree || payload_size) app->seq = seq;
}
diff --git a/core/module-save.h b/core/module-save.h
index 819f948..32e68c7 100644
--- a/core/module-save.h
+++ b/core/module-save.h
@@ -24,16 +24,27 @@ PureResult pure_save_image(PureApp *, const PureModule *m, uint32_t state, const
void pure_save_close_start(PureApp *);
/* The log keeps only the entries some list's image does not hold yet. */
void pure_save_release(PureApp *);
-/* A snapshot's list (module.c). */
+/* A snapshot's list (module.c): a state's, or the heard list's
+ * (PURE_LIST_HEARD) in a snapshot made with it (pure_save_snapshot). */
const PureList *pure_snapshot_list(const PureSnapshot *, uint32_t state);
+PureSnapshot *pure_save_snapshot(PureApp *, bool heard);
+/* The heard list of a starting app (module.h, PureApp): from its image with
+ * its logged changes made again, or new (saved.rewrite: it needs an image,
+ * which pure_save_heard_image writes). */
+PureResult pure_save_open_heard(PureApp *, PureList **out, PureSaved *saved);
+PureResult pure_save_heard_image(PureApp *);
+/* An empty heard list. */
+PureList *pure_heard_new(void);
/* Open the stored numbers, cursors and indices, and put each stored one of the
* same name and kind that still fits into values, for the states of m that
* kept marks as new (-1); texts and lists are m's. */
PureResult pure_save_open_values(PureApp *, const PureModule *m, double *values, const int *kept,
PureDraft *const *texts, PureList *const *lists);
/* The stored links of a starting app (module.h, PureApp); CONFLICT when an
- * output with sends left is missing or has other fields. */
-PureResult pure_save_open_links(PureApp *, const PureModule *m);
+ * output with sends left is missing or has other fields. Links stored before
+ * the heard list kept the numbers heard: they go to it (moved: it needs an
+ * image). */
+PureResult pure_save_open_links(PureApp *, const PureModule *m, bool *moved);
/* 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,
diff --git a/core/module.c b/core/module.c
index 45f67f7..3a694a1 100644
--- a/core/module.c
+++ b/core/module.c
@@ -2442,6 +2442,7 @@ static void open_burst(PureApp *app) {
}
if (app->lists[i]) app->durable_lists[i] = pure_list_changes(app->lists[i]).size;
}
+ if (app->heard) app->durable_heard = pure_list_changes(app->heard).size;
app->open = true;
}
@@ -2462,6 +2463,7 @@ static void close_burst(PureApp *app, bool keep) {
}
if (app->lists[i] && !keep) pure_list_forget(app->lists[i], app->durable_lists[i]);
}
+ if (app->heard && !keep) pure_list_forget(app->heard, app->durable_heard);
if (keep) pure_undo_commit(&app->undo);
else {
pure_undo_rollback(&app->undo, 0);
@@ -2478,7 +2480,8 @@ static void drop_sent(PureApp *app, size_t count) {
}
/* After a save: what it stored is the links' saved state (trimmed sends go),
- * or, when it failed, the links return to the last saved state. */
+ * or, when it failed, the links return to the last saved state (the numbers
+ * heard go back with the burst's list changes). */
static void settle_links(PureApp *app, bool saved) {
if (saved) {
size_t kept = 0;
@@ -2488,18 +2491,10 @@ static void settle_links(PureApp *app, bool saved) {
}
app->sent_count = app->sent_durable = kept;
memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent));
- for (unsigned i = 0; i < app->heard_count; ++i) app->heard[i].saved = app->heard[i].seq;
} else {
drop_sent(app, app->sent_durable);
for (size_t i = 0; i < app->sent_count; ++i) app->sent[i].trimmed = false;
memcpy(app->last_sent, app->saved_sent, sizeof(app->last_sent));
- unsigned kept = 0;
- for (unsigned i = 0; i < app->heard_count; ++i)
- if (app->heard[i].saved) {
- app->heard[kept] = app->heard[i];
- app->heard[kept++].seq = app->heard[i].saved;
- }
- app->heard_count = kept;
}
app->links_changed = false;
}
@@ -2536,13 +2531,16 @@ struct PureSnapshot {
double values[PURE_MODULE_STATES];
PureDraft *texts[PURE_MODULE_STATES];
PureList *lists[PURE_MODULE_STATES];
+ PureList *heard; /* only in a checkpoint's of the heard list */
PureRetired retired;
struct PureSnapshot *newer;
};
void pure_snapshot_hold(PureSnapshot *snapshot) { __atomic_add_fetch(&snapshot->refs, 1, __ATOMIC_RELAXED); }
-const PureList *pure_snapshot_list(const PureSnapshot *snapshot, uint32_t state) { return snapshot->lists[state]; }
+const PureList *pure_snapshot_list(const PureSnapshot *snapshot, uint32_t state) {
+ return state == PURE_LIST_HEARD ? snapshot->heard : snapshot->lists[state];
+}
void pure_snapshot_release(PureSnapshot *snapshot) {
while (snapshot && !__atomic_sub_fetch(&snapshot->refs, 1, __ATOMIC_ACQ_REL)) {
@@ -2551,6 +2549,7 @@ void pure_snapshot_release(PureSnapshot *snapshot) {
pure_draft_close(snapshot->texts[i]);
pure_list_free(snapshot->lists[i]);
}
+ pure_list_free(snapshot->heard);
pure_retired_free(&snapshot->retired);
free(snapshot);
snapshot = newer;
@@ -2574,7 +2573,7 @@ void pure_app_stop(PureApp *app) {
pure_draft_close(app->stored_values);
drop_sent(app, 0);
free(app->sent);
- free(app->heard);
+ pure_list_free(app->heard);
free(app->carried);
pure_draft_close(app->stored_links);
pure_draft_close(app->stored_seq);
@@ -2593,7 +2592,7 @@ void pure_app_stop(PureApp *app) {
/* Where one event began inside the burst: a failed event returns there. */
typedef struct {
PureVersion texts[PURE_MODULE_STATES];
- size_t lists[PURE_MODULE_STATES]; /* stored lists' journal lengths */
+ size_t lists[PURE_MODULE_STATES], heard; /* stored lists' journal lengths */
size_t undo, sent;
uint64_t last_sent[PURE_MODULE_INPUTS];
bool opened;
@@ -2606,6 +2605,7 @@ static void mark_event(PureApp *app, Mark *mark) {
if (app->texts[i]) mark->texts[i] = pure_draft_version(app->texts[i]);
if (app->lists[i]) mark->lists[i] = pure_list_changes(app->lists[i]).size;
}
+ mark->heard = app->heard ? pure_list_changes(app->heard).size : 0;
mark->undo = app->undo.count;
mark->sent = app->sent_count;
memcpy(mark->last_sent, app->last_sent, sizeof(mark->last_sent));
@@ -2616,29 +2616,32 @@ static void undo_event(PureApp *app, const Mark *mark) {
if (app->texts[i]) pure_draft_restore(app->texts[i], mark->texts[i]);
if (app->lists[i]) pure_list_forget(app->lists[i], mark->lists[i]);
}
+ if (app->heard) pure_list_forget(app->heard, mark->heard);
pure_undo_rollback(&app->undo, mark->undo);
drop_sent(app, mark->sent);
memcpy(app->last_sent, mark->last_sent, sizeof(app->last_sent));
if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */
}
-/* The last number heard from source, or NULL. */
-static struct PureHeard *heard_from(PureApp *app, const char *source) {
- for (unsigned i = 0; i < app->heard_count; ++i)
- if (!strcmp(app->heard[i].source, source)) return &app->heard[i];
- return NULL;
+/* Where source's item is in the heard list, or would go; found says which. */
+static size_t heard_at(const PureApp *app, const char *source, bool *found) {
+ *found = false;
+ return app->heard ? pure_list_seek(app->heard, 0, (PureCell){.text = (unsigned char *)source, .size = strlen(source)}, found) : 0;
}
-/* Hear seq from source (room for a new source was made before). */
-static void hear(PureApp *app, const char *source, uint64_t seq) {
- struct PureHeard *heard = heard_from(app, source);
- if (!heard) {
- heard = &app->heard[app->heard_count++];
- *heard = (struct PureHeard){.saved = 0};
- memcpy(heard->source, source, strlen(source) + 1);
- }
- heard->seq = seq;
- app->links_changed = true;
+uint64_t pure_app_heard(const PureApp *app, const char *source) {
+ bool found;
+ size_t at = heard_at(app, source, &found);
+ return found ? (uint64_t)pure_list_get(app->heard, at, 1).num : 0;
+}
+
+/* Hear seq from source: a change of the heard list, undone with the burst. */
+static PureResult hear(PureApp *app, const char *source, uint64_t seq) {
+ bool found;
+ size_t at = heard_at(app, source, &found);
+ if (found) return pure_list_set(app->heard, at, 1, (PureCell){.num = (double)seq}, &app->undo);
+ PureCell item[2] = {{.text = (unsigned char *)source, .size = strlen(source)}, {.num = (double)seq}};
+ return pure_list_insert(app->heard, at, item, &app->undo);
}
static PureResult keep_reply(PureApp *app);
@@ -2673,6 +2676,8 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
if (result == PURE_OK && app->has_reply &&
((app->reply.source == FROM_DOCUMENT && ev->modified) || app->reply.source == FROM_TEXT))
result = keep_reply(app);
+ /* A delivery's number is heard with its changes. */
+ if (result == PURE_OK && app->delivering.pending) result = hear(app, app->delivering.source, app->delivering.seq);
if (result != PURE_OK) {
undo_event(app, &mark);
clear_rows(app);
@@ -2691,10 +2696,7 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
app->document_pending = true;
app->revision += ev->modified; /* a save moves an edited document one revision */
}
- if (app->delivering.pending) {
- hear(app, app->delivering.source, app->delivering.seq);
- app->delivering.pending = false;
- }
+ app->delivering.pending = false;
return event && !app->deferred ? pure_app_save(app) : PURE_OK;
}
@@ -2889,23 +2891,17 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui
const PureModule *m = app->module;
size_t length = source ? strlen(source) : 0;
app->duplicate = false;
+ /* Kept states without a directory would not keep the numbers heard. */
if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed || !length ||
- length > PURE_SOURCE || !seq) return PURE_INVALID;
+ length > PURE_SOURCE || !seq || seq > PURE_SEQ_MOST || !app->heard || (pure_save_enabled(app) && !app->log))
+ return PURE_INVALID;
for (size_t i = 0; i < length; ++i)
if (source[i] <= ' ' || source[i] > '~') return PURE_INVALID;
- struct PureHeard *heard = heard_from(app, source);
- if (heard && seq <= heard->seq) {
+ if (seq <= pure_app_heard(app, source)) {
pure_app_finish(app);
app->duplicate = true;
return PURE_OK;
}
- if (!heard && app->heard_count == app->heard_capacity) {
- unsigned capacity = app->heard_capacity ? 2 * app->heard_capacity : 4;
- struct PureHeard *grown = realloc(app->heard, capacity * sizeof(*grown));
- if (!grown) return PURE_ALLOCATION;
- app->heard = grown;
- app->heard_capacity = capacity;
- }
app->delivering.source = source;
app->delivering.seq = seq;
app->delivering.pending = true;
@@ -2917,17 +2913,15 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui
if (app->fault == PURE_FAULT_NONE) return result;
Mark mark;
mark_event(app, &mark);
- hear(app, source, seq);
+ PureResult heard = hear(app, source, seq);
+ if (heard != PURE_OK) {
+ undo_event(app, &mark);
+ return heard;
+ }
PureResult saved = app->deferred ? PURE_OK : pure_app_save(app);
return saved == PURE_OK ? result : saved;
}
-uint64_t pure_app_heard(const PureApp *app, const char *source) {
- for (unsigned i = 0; i < app->heard_count; ++i)
- if (!strcmp(app->heard[i].source, source)) return app->heard[i].seq;
- return 0;
-}
-
PureResult pure_app_reply(const PureApp *app, bool (*visit)(void *, PureSpan), void *context) {
if (!app->has_reply) return PURE_OK;
return visit_value(app, &app->arena, app->reply, visit, context);
@@ -3099,7 +3093,9 @@ PureResult pure_app_cell_text(PureApp *app, uint32_t column, size_t row, bool (*
return result == PURE_OK ? visit_value(app, &app->view_arena, slot.bytes, visit, context) : result;
}
-PureSnapshot *pure_app_snapshot(PureApp *app) {
+PureSnapshot *pure_app_snapshot(PureApp *app) { return pure_save_snapshot(app, false); }
+
+PureSnapshot *pure_save_snapshot(PureApp *app, bool heard) {
if (app->open || app->reading) return NULL;
const PureModule *m = app->module;
PureSnapshot *made = calloc(1, sizeof(*made)), *last = app->snapshot;
@@ -3118,6 +3114,7 @@ PureSnapshot *pure_app_snapshot(PureApp *app) {
} else
ok = (made->lists[i] = pure_list_share(app->lists[i], &app->retired)) != NULL;
}
+ if (ok && heard) ok = (made->heard = pure_list_share(app->heard, &app->retired)) != NULL;
if (!ok) {
pure_snapshot_release(made);
return NULL;
@@ -3256,6 +3253,11 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
bool lists_stored = stored && app->data.directory && app->data.directory(app->data.context);
if (old && app->checkpoint) return PURE_CONFLICT; /* the host has a checkpoint of the running module out */
PureResult result = !old && lists_stored ? pure_save_open_log(app) : PURE_OK;
+ /* The numbers heard belong to the app, whatever its module. */
+ PureSaved heard = {0};
+ if (!old && result == PURE_OK)
+ result = lists_stored ? pure_save_open_heard(app, &app->heard, &heard) :
+ (app->heard = pure_heard_new()) ? PURE_OK : PURE_ALLOCATION;
for (uint32_t i = 0; i < module->states; ++i) {
int previous = old ? pure_module_state(old, module->state_names[i]) : -1;
kept[i] = previous >= 0 && old->state_type[previous] == module->state_type[i] ? previous : -1;
@@ -3310,7 +3312,9 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
}
if (result == PURE_OK && stored) result = pure_save_open_values(app, module, values, kept, texts, lists);
/* Outputs keep their sends and numbers by name. */
- if (result == PURE_OK) result = old ? remap_outputs(app, old, module, false) : stored ? pure_save_open_links(app, module) : PURE_OK;
+ bool heard_moved = false;
+ if (result == PURE_OK) result = old ? remap_outputs(app, old, module, false) : stored ? pure_save_open_links(app, module, &heard_moved) : PURE_OK;
+ heard.rewrite = heard.rewrite || (heard_moved && lists_stored);
for (uint32_t i = 0; i < module->states; ++i) {
uint32_t list = module->state_ref[i];
if (module->state_type[i] == PURE_STATE_INDEX && moved[list] && values[i] < (double)pure_list_count(lists[list]))
@@ -3332,6 +3336,8 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
for (uint32_t i = 0; i < module->states && result == PURE_OK; ++i)
if (rewrite[i] && (result = pure_save_image(app, module, i, lists[i])) == PURE_OK)
listed[i] = (PureSaved){.image = app->seq};
+ if (result == PURE_OK && heard.rewrite && (result = pure_save_heard_image(app)) == PURE_OK)
+ heard = (PureSaved){.image = app->seq};
if (imaged && result != PURE_OK && old) app->uncertain = true;
if (result != PURE_OK) {
for (uint32_t i = 0; i < module->states; ++i) {
@@ -3367,6 +3373,12 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
app->logged_since[i] = listed[i].since;
app->logged[i] = listed[i].logged;
}
+ if (!old) {
+ app->image_seq[PURE_LIST_HEARD] = heard.image;
+ app->logged_since[PURE_LIST_HEARD] = heard.since;
+ app->logged[PURE_LIST_HEARD] = heard.logged;
+ if (lists_stored) pure_list_journal(app->heard, true);
+ }
pure_save_release(app);
for (uint32_t i = 0; i < module->states; ++i) {
/* A stored list's changes are recorded from now on. */
diff --git a/core/module.h b/core/module.h
index e9e5dd3..8176bea 100644
--- a/core/module.h
+++ b/core/module.h
@@ -80,6 +80,8 @@ typedef enum { PURE_LESS, PURE_LESS_EQUAL, PURE_GREATER, PURE_GREATER_EQUAL, PUR
typedef enum { PURE_KIND_KEY = 1, PURE_KIND_MESSAGE = 2, PURE_KIND_TEXT = 3, PURE_KIND_TIMER = 4,
PURE_KIND_OUTPUT = 5, PURE_KIND_ARGUMENTS = 6 } PureInputKind;
enum { PURE_TIMER_LONGEST = 86400000, PURE_SOURCE = 64 };
+/* The largest number a delivery may have: it is kept as a binary64. */
+#define PURE_SEQ_MOST (UINT64_C(1) << 53)
/* A message field: a u32le or a binary64 (fixed, at its offset in the
* payload's prefix), or bytes, text or rows of any length. The prefix also
* holds each variable field's u32le length but the last one's (its offset is
@@ -261,6 +263,8 @@ typedef struct { unsigned char *bytes; size_t used, capacity; } PureArena;
* 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, or one past PURE_APP_PENDING keys, saves the burst first). */
+/* The heard list's place among the stored lists' records (PureApp). */
+enum { PURE_LIST_HEARD = PURE_MODULE_STATES };
typedef struct {
const PureModule *module;
PureData data;
@@ -269,12 +273,13 @@ typedef struct {
PureList *lists[PURE_MODULE_STATES];
PureUndo undo; /* list changes of the current event */
PureDraft *list_drafts[PURE_MODULE_STATES]; /* lists the tree kept before images, until emptied */
- /* Stored lists (module-save.c): the save each one's image holds, the
- * first save logged since that changed it (0: none), and the bytes of its
- * changes logged since; the log, the tree's number of its last save, the
- * last save's number, and a checkpoint being written. */
- uint64_t image_seq[PURE_MODULE_STATES], logged_since[PURE_MODULE_STATES];
- size_t logged[PURE_MODULE_STATES];
+ /* Stored lists (module-save.c), by state, and the heard list's last
+ * (PURE_LIST_HEARD): the save each one's image holds, the first save
+ * logged since that changed it (0: none), and the bytes of its changes
+ * logged since; the log, the tree's number of its last save, the last
+ * save's number, and a checkpoint being written. */
+ uint64_t image_seq[PURE_MODULE_STATES + 1], logged_since[PURE_MODULE_STATES + 1];
+ size_t logged[PURE_MODULE_STATES + 1];
struct PureLog *log;
PureDraft *stored_seq;
uint64_t seq;
@@ -318,15 +323,18 @@ typedef struct {
/* Links (runtime-owned, saved with the states): messages sent on outputs
* and not trimmed, in the order sent (the first sent_durable are saved;
* a trimmed one goes at the next save), each output's last number (and
- * as saved), those of outputs a module dropped, and the last number
- * delivered from each source (and as saved). */
+ * as saved), and those of outputs a module dropped. */
struct PureSent { uint32_t output; uint64_t seq; unsigned char *payload; size_t size; bool trimmed; } *sent;
size_t sent_count, sent_capacity, sent_durable;
uint64_t last_sent[PURE_MODULE_INPUTS], saved_sent[PURE_MODULE_INPUTS];
struct PureCarried { char name[PURE_MODULE_NAME]; uint64_t last; } *carried;
unsigned carried_count;
- struct PureHeard { char source[PURE_SOURCE + 1]; uint64_t seq, saved; } *heard;
- unsigned heard_count, heard_capacity;
+ /* The last number delivered from each source: a list of {source (text),
+ * number}, in order of source, stored as a list state is (module-save.c)
+ * and changed with the events that deliver, so a failed save takes its
+ * numbers back with them; its journal's length when the burst began. */
+ PureList *heard;
+ size_t durable_heard;
PureDraft *stored_links;
bool links_changed; /* since the last save */
bool duplicate; /* the last delivery was heard before and not applied */
@@ -421,7 +429,9 @@ void pure_app_outbox_newest(const PureApp *, PureSentVisit visit, void *context)
/* Forget output's saved sends numbered up to through (saved with the next save). */
void pure_app_trim(PureApp *, uint32_t output, uint64_t through);
/* Deliver a message event from source (a sender's name, 1 to 64 printable
- * ASCII bytes) numbered seq (from 1, growing per source). A number heard
+ * ASCII bytes) numbered seq (from 1 to PURE_SEQ_MOST, growing per source).
+ * An app whose states are kept takes deliveries only where its lists are
+ * (a directory: the numbers heard are a list). A number heard
* before is not applied again: it succeeds with app->duplicate set. The
* number counts as heard with the event's changes, and also when the app
* refuses the message (a fault: it was delivered, and refused); not when a
diff --git a/core/test-module.c b/core/test-module.c
index b2024b5..9a1f8a4 100644
--- a/core/test-module.c
+++ b/core/test-module.c
@@ -727,6 +727,15 @@ static const char *state_text(const PureApp *app, uint32_t state) {
return t.text;
}
+/* A stored value's first four bytes (or fewer). */
+static const char *read_state(const PureDraft *draft) {
+ static Text t;
+ t.size = 0;
+ size_t size = pure_draft_size(draft) < 4 ? pure_draft_size(draft) : 4;
+ CHECK(pure_draft_read(draft, 0, size, text_span, &t) == PURE_OK);
+ return t.text;
+}
+
static double number_view(PureApp *app, uint32_t view) {
double value;
CHECK(pure_app_view(app, view, &value));
@@ -3918,6 +3927,56 @@ static void links(void) {
CHECK(deliver(&b, "orders.ordered", 2, "ink", 3) == PURE_OK && !b.duplicate && b.values[0] == 5);
CHECK(pure_app_save(&b) == PURE_OK);
pure_app_stop(&b);
+ /* Links stored before the heard list kept its numbers: a start moves them
+ * there (the larger number of a source wins) and stores the links without. */
+ PureDraft *stored_links;
+ int64_t revision;
+ unsigned char old_links[64] = "PLK1";
+ size_t at = 5;
+ old_links[4] = 0; /* no outputs */
+ old_links[at++] = 2, old_links[at++] = 0;
+ const char *old_sources[] = {"legacy", "orders.ordered"};
+ uint64_t old_numbers[] = {5, 1};
+ for (unsigned k = 0; k < 2; ++k) {
+ old_links[at++] = (unsigned char)strlen(old_sources[k]);
+ memcpy(old_links + at, old_sources[k], strlen(old_sources[k]));
+ at += strlen(old_sources[k]);
+ for (unsigned i = 0; i < 8; ++i) old_links[at++] = (unsigned char)(old_numbers[k] >> (8 * i));
+ }
+ CHECK(pure_state_open(path_b, "links", &stored_links) == PURE_OK);
+ CHECK(pure_draft_splice(stored_links, 0, pure_draft_size(stored_links), (PureSpan){old_links, at}) == PURE_OK);
+ CHECK(pure_draft_commit(stored_links, &revision) == PURE_OK);
+ pure_draft_close(stored_links);
+ CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK);
+ CHECK(pure_app_heard(&b, "legacy") == 5 && pure_app_heard(&b, "orders.ordered") == 2 && pure_app_heard(&b, "x") == 1);
+ CHECK(deliver(&b, "legacy", 5, "old", 1) == PURE_OK && b.duplicate && deliver(&b, "legacy", 6, "new", 1) == PURE_OK && !b.duplicate);
+ pure_app_stop(&b);
+ CHECK(pure_state_open(path_b, "links", &stored_links) == PURE_OK && !memcmp(read_state(stored_links), "PLK2", 4));
+ pure_draft_close(stored_links);
+ /* Numbers heard from more sources than two bytes count stay heard across
+ * a start; a save logs only the numbers it changed, and the heard list's
+ * image is written again once they outgrow it (from a snapshot: through a
+ * store handle). */
+ PureStore *store;
+ CHECK(pure_store_open(path_b, &store) == PURE_OK);
+ CHECK(pure_app_start(&b, &module_b, pure_store_data(store)) == PURE_OK && pure_app_heard(&b, "legacy") == 6);
+ unsigned long checkpoints = pure_save_test_checkpoints;
+ for (unsigned k = 0; k < 400; ++k) CHECK(deliver(&b, "s1", 1 + k, "more", 1) == PURE_OK);
+ CHECK(pure_save_test_checkpoints > checkpoints && pure_app_heard(&b, "s1") == 400);
+ CHECK(pure_app_defer(&b, true));
+ char many[16];
+ for (unsigned k = 0; k < 70000; ++k) {
+ snprintf(many, sizeof(many), "s%u", k);
+ CHECK(deliver(&b, many, 400 + k % 3, "one", 1) == PURE_OK && b.duplicate == (k == 1 && k % 3 == 0));
+ if (k % 5000 == 4999) CHECK(pure_app_save(&b) == PURE_OK);
+ }
+ CHECK(pure_app_save(&b) == PURE_OK);
+ pure_app_stop(&b);
+ CHECK(pure_app_start(&b, &module_b, pure_store_data(store)) == PURE_OK && pure_list_count(b.heard) == 70003);
+ CHECK(pure_app_heard(&b, "s0") == 400 && pure_app_heard(&b, "s1") == 401 && pure_app_heard(&b, "s65537") == 402);
+ CHECK(deliver(&b, "s65537", 402, "again", 1) == PURE_OK && b.duplicate);
+ pure_app_stop(&b);
+ pure_store_close(store);
/* Replacing: a module without the output is refused while sends are left
* on it; once they are trimmed it replaces, and the output's number is
@@ -3970,7 +4029,7 @@ static void links(void) {
CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK);
o = outbox_of(&a);
CHECK(o.count == (size_t)a.values[0] && (!o.count || o.items[o.count - 1].seq == o.count));
- heard = b.heard_count ? b.heard[0].seq : 0;
+ heard = pure_app_heard(&b, "orders.ordered");
CHECK(b.values[0] == (double)heard);
kept[0] = (size_t)heard;
kept[1] = o.count;