aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--core/module-list.c40
-rw-r--r--core/module-list.h14
-rw-r--r--core/module.c100
-rw-r--r--core/test-module.c57
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
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 15:46:53 +0000