From ec210a8aed62c301e9ae384d4d157e379fa58f54 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Mon, 28 Sep 2026 19:08:09 +0100 Subject: 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). --- core/module.c | 112 ++++++++++++++++++++++++++++++++-------------------------- 1 file changed, 62 insertions(+), 50 deletions(-) (limited to 'core/module.c') 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. */ -- cgit v1.2.3