Let a list make its next nodes of those no snapshot reads any more

Each node a list copies once a snapshot shares it went, once the last
snapshot that could read it was released, back to the system: a leaf of
TPC-C's (1.3 KB) past glibc's thread cache, often freed by a connection
thread into the owner's arena, while the owner allocated the next. That
was 70% of what snapshots cost the owner. Now what no snapshot reads
comes back to the app under the register's lock, and each snapshot the
owner makes hands it to the lists of its size as spares, as many as each
made lately; retired entries carry what the pass and the reuse ask of a
node, so neither reads it again. TPC-C in RAM, W=1, 64 terminals: 10-14%
more a second at 7% less CPU a transaction, resident 1.97x the state
bytes where it was 2.18x (docs/research/107).

4 files changed, 150 insertions, 47 deletions

diff --git a/core/module-list.c b/core/module-list.c
index 861798e..16511de 100644
--- a/core/module-list.c
+++ b/core/module-list.c
@@ -69,8 +69,9 @@ struct PureList {
     uint64_t id, version; /* which list, and a number every change moves on */
     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];
+    Node **spare[2];      /* inner nodes and leaves let go that no copy reads, kept for the next ones */
+    unsigned spares[2], spare_room[2];
+    unsigned needed[2];   /* inner nodes and leaves it made lately: halved each time the retired come back */
     unsigned refs;        /* a shared copy's references */
     bool copy;            /* a shared copy: it owns none of its nodes or texts */
 };
@@ -88,7 +89,14 @@ typedef struct {
         size_t cells; /* REMOVED: where the item's cells start in the log's cells */
     };
 } Entry;
-typedef struct { void *pointer; bool text; } Retired;
+/* A node or text let go while a copy may read it, and what the copies and
+ * the lists ask of it (so neither reads it again). */
+typedef struct {
+    void *pointer;
+    uint64_t made;  /* the share count it was made at */
+    uint32_t size;  /* a node's */
+    bool text, leaf;
+} Retired;
 
 static uint64_t lists_made;
 
@@ -160,7 +168,8 @@ static bool shared_text(const PureList *list, const unsigned char *text) { retur
 static bool shared_node(const PureList *list, const Node *node) { return node->made < list->shares; }
 static void free_text(const unsigned char *text) { if (text) free((uint64_t *)text - 1); }
 
-/* Kept until the copies that may read it are gone (pure_retired_free). */
+/* Kept until the copies that may read it are gone (pure_retired_free), or
+ * then taken back (pure_retired_reuse). */
 static void retire(PureList *list, void *pointer, bool text) {
     PureRetired *retired = list->retired;
     if (retired->count == retired->capacity) {
@@ -170,7 +179,10 @@ static void retire(PureList *list, void *pointer, bool text) {
         retired->entries = grown;
         retired->capacity = capacity;
     }
-    ((Retired *)retired->entries)[retired->count++] = (Retired){pointer, text};
+    const Node *node = text ? NULL : pointer;
+    ((Retired *)retired->entries)[retired->count++] =
+        text ? (Retired){pointer, ((const uint64_t *)pointer)[-1], 0, true, false}
+             : (Retired){pointer, node->made, node->leaf ? (uint32_t)list->layout.leaf_size : INNER_SIZE, false, node->leaf};
 }
 
 /* The list no longer has text: free it, or retire it while a copy may read it. */
@@ -195,10 +207,6 @@ 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;
@@ -213,24 +221,58 @@ static bool retired_room(PureRetired *retired, size_t more) {
 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;
+    for (size_t i = 0; kept && i < from->count; ++i) keeping += entries[i].made < 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;
+        PureRetired *to = kept && entries[i].made < 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};
+    from->count = 0; /* its room stays, for what is retired next */
+}
+
+/* Keep node for the list's next of its kind; false when it keeps enough
+ * already (or has no room for more). */
+static bool keep_spare(PureList *list, Node *node, bool leaf, unsigned enough) {
+    if (list->spares[leaf] >= enough) return false;
+    if (list->spares[leaf] == list->spare_room[leaf]) {
+        unsigned room = list->spare_room[leaf] ? 2 * list->spare_room[leaf] : SPARE_NODES;
+        Node **grown = realloc(list->spare[leaf], room * sizeof(*grown));
+        if (!grown) return false;
+        list->spare[leaf] = grown;
+        list->spare_room[leaf] = room;
+    }
+    list->spare[leaf][list->spares[leaf]++] = node;
+    return true;
+}
+
+void pure_retired_reuse(PureRetired *returned, PureList *const *lists, unsigned count) {
+    Retired *entries = returned->entries;
+    for (size_t i = 0; i < returned->count; ++i) {
+        const Retired *r = &entries[i];
+        if (r->text) {
+            free_text(r->pointer);
+            continue;
+        }
+        bool kept = false; /* by a list of its size, as many as it made lately (at least a few) */
+        for (unsigned l = 0; l < count && !kept; ++l) {
+            PureList *list = lists[l];
+            kept = list && (!r->leaf || list->layout.leaf_size == r->size) &&
+                   keep_spare(list, r->pointer, r->leaf, list->needed[r->leaf] > SPARE_NODES ? list->needed[r->leaf] : SPARE_NODES);
+        }
+        if (!kept) free(r->pointer);
+    }
+    for (unsigned l = 0; l < count; ++l)
+        if (lists[l]) lists[l]->needed[0] /= 2, lists[l]->needed[1] /= 2;
+    returned->count = 0;
 }
 
 static Node *new_node(PureList *list, bool leaf) {
-    Node *node = list->spare[leaf];
-    if (node) {
-        list->spare[leaf] = (Node *)node->first;
-        --list->spares[leaf];
-    } else if (!(node = malloc(leaf ? list->layout.leaf_size : INNER_SIZE)))
+    Node *node;
+    ++list->needed[leaf];
+    if (list->spares[leaf]) node = list->spare[leaf][--list->spares[leaf]];
+    else if (!(node = malloc(leaf ? list->layout.leaf_size : INNER_SIZE)))
         return NULL;
     node->made = list->shares;
     node->count = 0;
@@ -245,12 +287,7 @@ static Node *new_node(PureList *list, bool leaf) {
  * retired while a copy may read it, else kept for the next or freed. */
 static void drop_node(PureList *list, Node *node) {
     if (shared_node(list, node)) retire(list, node, false);
-    else if (list->spares[node->leaf] < SPARE_NODES) {
-        node->first = list->spare[node->leaf];
-        list->spare[node->leaf] = node;
-        ++list->spares[node->leaf];
-    } else
-        free(node);
+    else if (!keep_spare(list, node, node->leaf, SPARE_NODES)) free(node);
 }
 
 /* ---- The stored form: each field in order, a number as its binary64 bits
@@ -580,12 +617,10 @@ void pure_list_free(PureList *list) {
     } else {
         free(list->journal);
         let_go_tree(list, list->root);
-        for (unsigned kind = 0; kind < 2; ++kind)
-            while (list->spare[kind]) {
-                Node *next = (Node *)list->spare[kind]->first;
-                free(list->spare[kind]);
-                list->spare[kind] = next;
-            }
+        for (unsigned kind = 0; kind < 2; ++kind) {
+            while (list->spares[kind]) free(list->spare[kind][--list->spares[kind]]);
+            free(list->spare[kind]);
+        }
     }
     free(list);
 }
diff --git a/core/module-list.h b/core/module-list.h
index abe19cb..d97aa77 100644
--- a/core/module-list.h
+++ b/core/module-list.h
@@ -69,8 +69,15 @@ PureList *pure_list_share(PureList *, PureRetired *retired, uint64_t epoch);
 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. */
+ * pure_retired_reuse or pure_retired_free (later, and where it pleases):
+ * from is left empty, with its room (pure_retired_free lets that go). */
 void pure_retired_pass(PureRetired *from, uint64_t epoch, PureRetired *kept, PureRetired *gone);
+/* What was retired and no copy reads any more, on the lists' thread: each
+ * node goes back to a list of its size, to be made again, as many as the
+ * list made lately (at least a few: what it made counts half again at each
+ * return); the rest and the texts are freed. returned is left empty, with
+ * its room. */
+void pure_retired_reuse(PureRetired *returned, PureList *const *lists, unsigned count);
 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 ea6809e..e33f43f 100644
--- a/core/module.c
+++ b/core/module.c
@@ -2702,14 +2702,17 @@ bool pure_app_defer(PureApp *app, bool deferred) {
  * 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). */
+ * it (made after it was), or back to the app, whose lists make their next
+ * nodes of it (what they do not need goes 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 */
+    unsigned refs;        /* the snapshots not yet freed */
     bool locked;
-    uint64_t epoch; /* snapshots made */
+    uint64_t epoch;       /* snapshots made */
+    PureRetired returned; /* what no snapshot reads any more, for the app's lists (pure_retired_reuse) */
 } Register;
 
 struct PureSnapshot {
@@ -2739,24 +2742,25 @@ const PureList *pure_snapshot_list(const PureSnapshot *snapshot, uint32_t state)
 void pure_snapshot_release(PureSnapshot *snapshot) {
     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);
+        pure_retired_pass(&snapshot->retired, older ? older->epoch : 0, older ? &older->retired : NULL, &all->returned);
         unlock_register(all);
-    } else
-        gone = snapshot->retired;
-    pure_retired_free(&gone);
+    }
+    pure_retired_free(&snapshot->retired); /* what it had when never linked, else its room */
     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);
+    if (all && !__atomic_sub_fetch(&all->refs, 1, __ATOMIC_ACQ_REL)) {
+        pure_retired_free(&all->returned); /* the app is gone: no list takes it back */
+        free(all);
+    }
 }
 
 void pure_app_stop(PureApp *app) {
@@ -2775,8 +2779,8 @@ void pure_app_stop(PureApp *app) {
         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);
+    }
+    pure_retired_free(&app->retired);
     pure_draft_close(app->stored_values);
     free(app->carried);
     pure_draft_close(app->stored_links);
@@ -3363,6 +3367,31 @@ PureResult pure_app_cell_text(PureApp *app, uint32_t column, size_t row, bool (*
 
 PureSnapshot *pure_app_snapshot(PureApp *app) { return pure_save_snapshot(app, UINT32_MAX); }
 
+/* What no snapshot reads any more back to the app's lists, to make their
+ * next nodes of (a list copies each node it changes once a snapshot shares
+ * it: made of these, it asks the system for none). The register keeps the
+ * room for the next. */
+static void reuse_returned(PureApp *app, Register *all) {
+    lock_register(all);
+    PureRetired returned = all->returned;
+    all->returned = (PureRetired){0};
+    unlock_register(all);
+    if (returned.count) {
+        PureList *lists[PURE_MODULE_STATES + 2];
+        unsigned count = 0;
+        for (uint32_t i = 0; i < app->module->states; ++i)
+            if (app->lists[i]) lists[count++] = app->lists[i];
+        lists[count++] = app->outbox;
+        lists[count++] = app->heard;
+        pure_retired_reuse(&returned, lists, count);
+    }
+    lock_register(all);
+    bool kept = !all->returned.entries; /* none came back meanwhile: the room stays for the next */
+    if (kept) all->returned = returned;
+    unlock_register(all);
+    if (!kept) pure_retired_free(&returned);
+}
+
 PureSnapshot *pure_save_snapshot(PureApp *app, uint32_t runtime) {
     if (app->open || app->reading) return NULL;
     const PureModule *m = app->module;
@@ -3412,6 +3441,7 @@ PureSnapshot *pure_save_snapshot(PureApp *app, uint32_t runtime) {
     if (last) pure_snapshot_release(last);
     else pure_retired_free(&app->retired); /* shared only by a snapshot never made */
     app->snapshot = made;
+    reuse_returned(app, all);
     return made;
 }
 
diff --git a/core/test-module.c b/core/test-module.c
index 0a11f1c..fa388ce 100644
--- a/core/test-module.c
+++ b/core/test-module.c
@@ -1041,6 +1041,35 @@ static PureList *stored_list(const Grown *stored, size_t items) {
     return list;
 }
 
+/* A node no copy reads any more comes back to its list (pure_retired_reuse):
+ * the next node the list makes is that one, and a copy still held reads
+ * what it shared. */
+static void retired_come_back(void) {
+    static const uint8_t numbers[1] = {0};
+    PureList *list = pure_list_new(1, numbers);
+    CHECK(list);
+    for (unsigned i = 0; i < 8; ++i) CHECK(pure_list_insert(list, i, &(PureCell){.num = i}, NULL) == PURE_OK);
+    PureRetired retired[2] = {{0}};
+    PureRun run;
+    PureList *older = pure_list_share(list, &retired[0], 1);
+    CHECK(older && pure_list_run(list, 0, &run));
+    const void *first = run.path[run.depth]; /* the one leaf, which the copy reads */
+    CHECK(pure_list_set(list, 0, 0, (PureCell){.num = 100}, NULL) == PURE_OK); /* the list copies it */
+    CHECK(pure_list_run(list, 0, &run) && run.path[run.depth] != first);
+    PureList *newer = pure_list_share(list, &retired[1], 2);
+    CHECK(newer);
+    pure_list_free(older); /* no copy reads the first leaf now: it comes back */
+    pure_retired_reuse(&retired[0], &list, 1);
+    CHECK(pure_list_set(list, 1, 0, (PureCell){.num = 101}, NULL) == PURE_OK); /* a copy of the second, made of it */
+    CHECK(pure_list_run(list, 0, &run) && run.path[run.depth] == first);
+    CHECK(pure_list_get(list, 0, 0).num == 100 && pure_list_get(list, 1, 0).num == 101);
+    CHECK(pure_list_get(newer, 0, 0).num == 100 && pure_list_get(newer, 1, 0).num == 1);
+    pure_list_free(newer);
+    pure_list_free(list);
+    pure_retired_free(&retired[0]);
+    pure_retired_free(&retired[1]);
+}
+
 static void list_trees(void) {
     static const uint8_t text_fields[2] = {0, 1};
     uint32_t seed = 0x5eed1234u;
@@ -1054,7 +1083,8 @@ static void list_trees(void) {
     PureUndo undo = {0};
     Model model = {0}, before = {0};
     /* Up to four shares at a time, each with what it shared and the items
-     * and nodes retired while it was the newest. */
+     * and nodes retired while it was the newest: when the oldest goes, no
+     * copy reads those, and the list makes its next nodes of them. */
     enum { SHARES = 4 };
     PureList *copies[SHARES];
     Model shared[SHARES] = {{0}};
@@ -1137,7 +1167,7 @@ static void list_trees(void) {
             if (held == SHARES) {
                 CHECK(same_items(copies[first], &shared[first]));
                 pure_list_free(copies[first]);
-                pure_retired_free(&retired[first]);
+                pure_retired_reuse(&retired[first], &list, 1);
                 first = (first + 1) % SHARES;
                 --held;
             }
@@ -4520,6 +4550,7 @@ int main(void) {
     marks();
     text_mutations();
     lists();
+    retired_come_back();
     list_trees();
     folds();
     filters();