Let a held snapshot keep the states of its time, not all that came after

An app's snapshots formed a chain, each holding the next, and what the
app's lists let go while a snapshot was the newest went with it. So one
snapshot held for long held every later one and every node and text let
go since: a surface holding the snapshot of 20,000 items while their
notes were set again grew the node from 6.5 MB to 696 MB in 30 s, 3.4 KB
an update, without end.

A snapshot reads only what was made before it. Snapshots are now
numbered (their epoch, given to the lists' copies and so to each node
and text made after), and the ones held are linked under a register
they share. What the app lets go while a snapshot is its newest goes
with it, as before; when that one goes, what the newest older one still
held may read (made before its epoch) goes to it, and the rest back to
the system at once. The links move under a spin the register keeps; what
is freed is freed outside it. The same held snapshot now costs its
states once: 6.6 MB to 10.6 MB, flat after the notes diverge.

test-module's snapshot_held holds a snapshot through 40,000 events and
snapshots: none of the second 20,000's allocations stays (0 B; with the
chain's keeping, it fails). The kept-snapshots tests read eight held in
random order, and the module tests pass under AddressSanitizer too.

4 files changed, 174 insertions, 37 deletions

diff --git a/core/module-list.c b/core/module-list.c
index f32fd05..861798e 100644
--- a/core/module-list.c
+++ b/core/module-list.c
@@ -67,7 +67,7 @@ struct PureList {
     unsigned char *journal;
     size_t journal_size, journal_capacity;
     uint64_t id, version; /* which list, and a number every change moves on */
-    uint64_t shares;      /* copies made so far: a node or text made before the last one is shared */
+    uint64_t shares;      /* the last copy's epoch: a node or text made before it is shared */
     PureRetired *retired; /* where shared nodes and texts go when the list lets them go */
     Node *spare[2];       /* inner nodes and leaves let go that no copy read, kept for the next ones (linked by first) */
     unsigned spares[2];
@@ -195,6 +195,36 @@ void pure_retired_free(PureRetired *retired) {
     *retired = (PureRetired){0};
 }
 
+static uint64_t made_at(const Retired *r) {
+    return r->text ? ((const uint64_t *)r->pointer)[-1] : ((const Node *)r->pointer)->made;
+}
+
+/* Room for more entries in retired; false when memory runs out. */
+static bool retired_room(PureRetired *retired, size_t more) {
+    if (retired->count + more <= retired->capacity) return true;
+    size_t capacity = 2 * retired->capacity > retired->count + more ? 2 * retired->capacity : retired->count + more;
+    Retired *grown = realloc(retired->entries, capacity * sizeof(*grown));
+    if (!grown) return false;
+    retired->entries = grown;
+    retired->capacity = capacity;
+    return true;
+}
+
+void pure_retired_pass(PureRetired *from, uint64_t epoch, PureRetired *kept, PureRetired *gone) {
+    Retired *entries = from->entries;
+    size_t keeping = 0;
+    for (size_t i = 0; kept && i < from->count; ++i) keeping += made_at(&entries[i]) < epoch;
+    bool room = (!keeping || retired_room(kept, keeping)) &&
+                (!gone || keeping == from->count || retired_room(gone, from->count - keeping));
+    for (size_t i = 0; i < from->count; ++i) {
+        PureRetired *to = kept && made_at(&entries[i]) < epoch ? kept : gone;
+        if (room && to) ((Retired *)to->entries)[to->count++] = entries[i];
+        /* else kept for good rather than freed while read (memory ran out) */
+    }
+    free(entries);
+    *from = (PureRetired){0};
+}
+
 static Node *new_node(PureList *list, bool leaf) {
     Node *node = list->spare[leaf];
     if (node) {
@@ -570,16 +600,20 @@ void pure_list_clear(PureList *list) {
 
 size_t pure_list_count(const PureList *list) { return list->count; }
 
-PureList *pure_list_share(PureList *list, PureRetired *retired) {
+PureList *pure_list_share(PureList *list, PureRetired *retired, uint64_t epoch) {
     PureList *copy = malloc(sizeof(*copy));
     if (!copy) return NULL;
     *copy = (PureList){.layout = list->layout, .root = list->root, .count = list->count, .bytes = list->bytes,
                        .id = list->id, .version = list->version, .refs = 1, .copy = true};
     list->retired = retired;
-    ++list->shares;
+    list->shares = epoch > list->shares ? epoch : list->shares + 1;
     return copy;
 }
 
+void pure_list_mark(PureList *list, uint64_t epoch) {
+    if (list->retired && epoch > list->shares) list->shares = epoch;
+}
+
 void pure_list_hold(PureList *copy) { __atomic_add_fetch(&copy->refs, 1, __ATOMIC_RELAXED); }
 
 bool pure_list_current(const PureList *copy, const PureList *list) {
diff --git a/core/module-list.h b/core/module-list.h
index a4e760a..abe19cb 100644
--- a/core/module-list.h
+++ b/core/module-list.h
@@ -59,8 +59,18 @@ size_t pure_list_count(const PureList *);
  * freed, since the copy may still read them. The copy is freed with
  * pure_list_free, on any thread, when its last reference goes
  * (pure_list_hold takes one more). Only between events (no change waits to
- * be committed or undone). */
-PureList *pure_list_share(PureList *, PureRetired *retired);
+ * be committed or undone). epoch numbers the copy (as a snapshot of an app
+ * numbers its lists' copies, one number for all of them): a node or text
+ * the list makes after it has the epoch, and a copy of epoch e reads only
+ * what was made before e. */
+PureList *pure_list_share(PureList *, PureRetired *retired, uint64_t epoch);
+/* A copy made before and still current (the list unchanged since) is read
+ * as of epoch too: what the list lets go from now on is retired. */
+void pure_list_mark(PureList *, uint64_t epoch);
+/* Of what was retired, what a copy of epoch may still read (made before
+ * it) goes to kept (none when kept is NULL), and the rest to gone, for
+ * pure_retired_free (later, and where it pleases): from is left empty. */
+void pure_retired_pass(PureRetired *from, uint64_t epoch, PureRetired *kept, PureRetired *gone);
 void pure_list_hold(PureList *copy);
 /* Whether copy shares list's items as they are now. */
 bool pure_list_current(const PureList *copy, const PureList *list);
diff --git a/core/module.c b/core/module.c
index c65ef80..79d3c5b 100644
--- a/core/module.c
+++ b/core/module.c
@@ -2544,22 +2544,40 @@ bool pure_app_defer(PureApp *app, bool deferred) {
     return true;
 }
 
-/* ---- Snapshots. Snapshots of one app form a chain from oldest to newest,
- * each holding the next, so they go oldest first. List items the app lets go
- * while a snapshot is its newest may be read by that one and older ones, so
- * they go with it. */
+/* ---- Snapshots. An app's snapshots are numbered as they are made (their
+ * epoch), and the ones still held are linked, oldest to newest, under their
+ * register (which each of them holds, so it outlives the app). A list's node
+ * or text the app lets go while a snapshot is its newest may be read by that
+ * snapshot and by older ones made after it was: it goes with that snapshot,
+ * and when that one goes, to the newest older one still held that may read
+ * it (made after it was), or back to the system. So a snapshot held for long
+ * keeps what it shares, the states as they were then, and nothing made
+ * since. The links and retired lists move under the register's lock, a spin
+ * (none holds it long; what is freed is freed outside it). */
+typedef struct {
+    unsigned refs;  /* the snapshots not yet freed */
+    bool locked;
+    uint64_t epoch; /* snapshots made */
+} Register;
+
 struct PureSnapshot {
     unsigned refs;
     const PureModule *module;
-    uint64_t changes;
+    uint64_t changes, epoch;
     double values[PURE_MODULE_STATES];
     PureDraft *texts[PURE_MODULE_STATES];
     PureList *lists[PURE_MODULE_STATES];
     PureList *runtime[PURE_LISTS_STORED - PURE_LIST_HEARD]; /* only in a checkpoint's of one */
     PureRetired retired;
-    struct PureSnapshot *newer;
+    struct PureSnapshot *older, *newer; /* the held ones beside it */
+    Register *all;
 };
 
+static void lock_register(Register *r) {
+    while (__atomic_test_and_set(&r->locked, __ATOMIC_ACQUIRE)) { /* another thread moves links: a moment */ }
+}
+static void unlock_register(Register *r) { __atomic_clear(&r->locked, __ATOMIC_RELEASE); }
+
 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) {
@@ -2567,17 +2585,26 @@ const PureList *pure_snapshot_list(const PureSnapshot *snapshot, uint32_t state)
 }
 
 void pure_snapshot_release(PureSnapshot *snapshot) {
-    while (snapshot && !__atomic_sub_fetch(&snapshot->refs, 1, __ATOMIC_ACQ_REL)) {
-        PureSnapshot *newer = snapshot->newer;
-        for (uint32_t i = 0; i < PURE_MODULE_STATES; ++i) {
-            pure_draft_close(snapshot->texts[i]);
-            pure_list_free(snapshot->lists[i]);
-        }
-        for (unsigned r = 0; r < PURE_LISTS_STORED - PURE_LIST_HEARD; ++r) pure_list_free(snapshot->runtime[r]);
-        pure_retired_free(&snapshot->retired);
-        free(snapshot);
-        snapshot = newer;
+    if (!snapshot || __atomic_sub_fetch(&snapshot->refs, 1, __ATOMIC_ACQ_REL)) return;
+    Register *all = snapshot->all;
+    PureRetired gone = {0};
+    if (all) { /* out of the links: what the one before may read goes to it (still linked, so still there) */
+        lock_register(all);
+        PureSnapshot *older = snapshot->older;
+        if (older) older->newer = snapshot->newer;
+        if (snapshot->newer) snapshot->newer->older = older;
+        pure_retired_pass(&snapshot->retired, older ? older->epoch : 0, older ? &older->retired : NULL, &gone);
+        unlock_register(all);
+    } else
+        gone = snapshot->retired;
+    pure_retired_free(&gone);
+    for (uint32_t i = 0; i < PURE_MODULE_STATES; ++i) {
+        pure_draft_close(snapshot->texts[i]);
+        pure_list_free(snapshot->lists[i]);
     }
+    for (unsigned r = 0; r < PURE_LISTS_STORED - PURE_LIST_HEARD; ++r) pure_list_free(snapshot->runtime[r]);
+    free(snapshot);
+    if (all && !__atomic_sub_fetch(&all->refs, 1, __ATOMIC_ACQ_REL)) free(all);
 }
 
 void pure_app_stop(PureApp *app) {
@@ -2588,9 +2615,11 @@ void pure_app_stop(PureApp *app) {
         pure_list_free(app->lists[i]);
         pure_draft_close(app->list_drafts[i]);
     }
-    /* Items the newest snapshot may still read go when it goes. */
+    /* Items the newest snapshot may still read go with it. */
     if (app->snapshot) {
-        app->snapshot->retired = app->retired;
+        lock_register(app->snapshot->all);
+        pure_retired_pass(&app->retired, UINT64_MAX, &app->snapshot->retired, NULL);
+        unlock_register(app->snapshot->all);
         pure_snapshot_release(app->snapshot);
     } else
         pure_retired_free(&app->retired);
@@ -3149,10 +3178,16 @@ PureSnapshot *pure_save_snapshot(PureApp *app, uint32_t runtime) {
     if (app->open || app->reading) return NULL;
     const PureModule *m = app->module;
     PureSnapshot *made = calloc(1, sizeof(*made)), *last = app->snapshot;
-    if (!made) return NULL;
+    Register *all = last ? last->all : calloc(1, sizeof(*all));
+    if (!made || !all) {
+        free(made);
+        if (!last) free(all);
+        return NULL;
+    }
     made->refs = 1;
     made->module = m;
     made->changes = app->changes;
+    made->epoch = all->epoch + 1; /* the app's thread alone makes them */
     memcpy(made->values, app->values, sizeof(made->values));
     bool ok = true;
     for (uint32_t i = 0; i < m->states && ok; ++i) {
@@ -3161,25 +3196,32 @@ PureSnapshot *pure_save_snapshot(PureApp *app, uint32_t runtime) {
         if (last && last->lists[i] && pure_list_current(last->lists[i], app->lists[i])) {
             pure_list_hold(last->lists[i]);
             made->lists[i] = last->lists[i];
+            pure_list_mark(app->lists[i], made->epoch);
         } else
-            ok = (made->lists[i] = pure_list_share(app->lists[i], &app->retired)) != NULL;
+            ok = (made->lists[i] = pure_list_share(app->lists[i], &app->retired, made->epoch)) != NULL;
     }
     if (ok && runtime >= PURE_LIST_HEARD && runtime < PURE_LISTS_STORED) {
         PureList *list = runtime == PURE_LIST_HEARD ? app->heard : app->outbox;
-        ok = (made->runtime[runtime - PURE_LIST_HEARD] = pure_list_share(list, &app->retired)) != NULL;
+        ok = (made->runtime[runtime - PURE_LIST_HEARD] = pure_list_share(list, &app->retired, made->epoch)) != NULL;
     }
-    if (!ok) {
+    if (!ok) { /* never linked: it goes alone */
+        if (!last) free(all);
         pure_snapshot_release(made);
         return NULL;
     }
-    made->refs = last ? 3 : 2; /* the app's, the caller's and the last one's */
-    if (last) {
-        last->retired = app->retired;
+    made->refs = 2; /* the app's and the caller's */
+    made->all = all;
+    lock_register(all);
+    all->epoch = made->epoch;
+    ++all->refs;
+    made->older = last;
+    if (last) { /* what the app let go while last was its newest goes with it */
         last->newer = made;
-        pure_snapshot_release(last);
-    } else
-        pure_retired_free(&app->retired); /* shared only by a snapshot never made */
-    app->retired = (PureRetired){0};
+        pure_retired_pass(&app->retired, UINT64_MAX, &last->retired, NULL);
+    }
+    unlock_register(all);
+    if (last) pure_snapshot_release(last);
+    else pure_retired_free(&app->retired); /* shared only by a snapshot never made */
     app->snapshot = made;
     return made;
 }
diff --git a/core/test-module.c b/core/test-module.c
index 788bfc6..1446e73 100644
--- a/core/test-module.c
+++ b/core/test-module.c
@@ -5,6 +5,9 @@
 #include "store.h"
 #include <dirent.h>
 #include <fcntl.h>
+#ifdef __GLIBC__
+#include <malloc.h>
+#endif
 #include <math.h>
 #include <pthread.h>
 #include <stdio.h>
@@ -1056,6 +1059,7 @@ static void list_trees(void) {
     Model shared[SHARES] = {{0}};
     PureRetired retired[SHARES + 1] = {{0}};
     unsigned first = 0, held = 0, splits_seen = 0;
+    uint64_t epoch = 0; /* the copies', one after another */
     size_t largest = 0, events = 0, undone = 0;
     for (unsigned event = 0; event < 6000; ++event) {
         size_t mark = undo.count, journal = pure_list_changes(list).size;
@@ -1137,7 +1141,7 @@ static void list_trees(void) {
                 --held;
             }
             unsigned slot = (first + held) % SHARES;
-            CHECK((copies[slot] = pure_list_share(list, &retired[slot])));
+            CHECK((copies[slot] = pure_list_share(list, &retired[slot], ++epoch)));
             model_copy(&shared[slot], &model);
             ++held;
         }
@@ -1201,7 +1205,7 @@ static void list_trees(void) {
     CHECK(keyed);
     for (unsigned round = 0; round < 40000; ++round) {
         if (round == 20000) {
-            PureList *copy = pure_list_share(keyed, &keyed_retired);
+            PureList *copy = pure_list_share(keyed, &keyed_retired, 1);
             CHECK(copy && pure_list_keyed(keyed, 0) == PURE_OK && same_items(copy, &keys) && same_items(keyed, &keys));
             pure_list_free(copy);
         }
@@ -3675,6 +3679,51 @@ static void snapshots_read(Spec spec, void (*event)(PureApp *, uint32_t *), unsi
     remove_store(path);
 }
 
+/* A snapshot held while the app goes on for long keeps the states as they
+ * were then, and nothing the app makes and lets go after it: holding one
+ * costs the states of its time, however many events and snapshots follow
+ * (it held them all before, each snapshot holding the next). */
+static size_t snapshot_held_growth;
+
+static void snapshot_held(Spec spec, void (*event)(PureApp *, uint32_t *)) {
+    static char was[1 << 16], shown_text[1 << 16];
+    char *path = new_store();
+    PureStore *store;
+    CHECK(load(spec, &module_a) == PURE_OK && pure_store_open(path, &store) == PURE_OK);
+    PureApp app, reader = {0};
+    CHECK(pure_app_start(&app, &module_a, pure_store_data(store)) == PURE_OK);
+    uint32_t seed = 0x401du;
+    for (unsigned step = 0; step < 200; ++step) event(&app, &seed);
+    PureSnapshot *held = pure_app_snapshot(&app);
+    CHECK(held);
+    Shown before = {was, 0, sizeof(was)}, shown = {shown_text, 0, sizeof(shown_text)};
+    describe(&app, &before);
+    size_t used[3] = {0, 0, 0};
+    for (unsigned half = 0; half < 2; ++half) {
+#ifdef __GLIBC__
+        used[half] = mallinfo2().uordblks;
+#endif
+        for (unsigned step = 0; step < 20000; ++step) {
+            event(&app, &seed);
+            pure_snapshot_release(pure_app_snapshot(&app));
+        }
+    }
+#ifdef __GLIBC__
+    used[2] = mallinfo2().uordblks;
+    /* The first half may make the app's own copies of what the held one shares; the second makes none to keep. */
+    snapshot_held_growth = used[2] > used[1] ? used[2] - used[1] : 0;
+    CHECK(snapshot_held_growth < 256 * 1024);
+#endif
+    pure_app_read(&reader, held);
+    describe(&reader, &shown);
+    CHECK(!strcmp(before.text, shown.text));
+    pure_snapshot_release(held);
+    pure_app_stop(&reader);
+    pure_app_stop(&app);
+    pure_store_close(store);
+    remove_store(path);
+}
+
 static void crash_event(PureApp *app, uint32_t *seed) {
     (void)seed;
     CHECK(pure_app_input(app, 0, PURE_EDGE_DOWN) == PURE_OK);
@@ -3688,6 +3737,7 @@ static void snapshots(void) {
     snapshots_kept(keyed, &unkeyed, keyed_event, 3000);
     snapshots_kept(crashing, NULL, crash_event, 300);
     snapshots_read(keyed, keyed_event, 3000);
+    snapshot_held(keyed, keyed_event);
     snapshots_read(crashing, crash_event, 300);
     /* Without stored states (texts only in memory) there is no snapshot. */
     CHECK(load(crashing, &module_a) == PURE_OK);
@@ -3695,7 +3745,8 @@ static void snapshots(void) {
     CHECK(pure_app_start(&app, &module_a, (PureData){0}) == PURE_OK && !pure_app_snapshot(&app));
     pure_app_stop(&app);
     CHECK(snapshot_retired > 100);
-    printf("{\"snapshots\":6600,\"items_retired\":%zu,\"reads_on_threads\":%zu}\n", snapshot_retired, snapshot_reads);
+    printf("{\"snapshots\":6600,\"items_retired\":%zu,\"reads_on_threads\":%zu,\"held_growth_bytes\":%zu}\n",
+           snapshot_retired, snapshot_reads, snapshot_held_growth);
 }
 
 /* A message of a text, a number, bytes and a u32, in that order: the prefix