diff options
| -rw-r--r-- | core/module-list.c | 8 | ||||
| -rw-r--r-- | core/module-list.h | 2 | ||||
| -rw-r--r-- | core/module-save.c | 145 | ||||
| -rw-r--r-- | core/module-save.h | 32 | ||||
| -rw-r--r-- | core/module.c | 151 | ||||
| -rw-r--r-- | core/module.h | 28 | ||||
| -rw-r--r-- | core/test-module.c | 64 |
7 files changed, 277 insertions, 153 deletions
diff --git a/core/module-list.c b/core/module-list.c index 522523f..f32fd05 100644 --- a/core/module-list.c +++ b/core/module-list.c @@ -497,7 +497,7 @@ PureList *pure_list_new(unsigned fields, const uint8_t *text_fields) { l->fields = fields; size_t at = sizeof(Node); for (unsigned f = 0; f < fields; ++f) { - l->text_fields[f] = text_fields[f] != 0; + l->text_fields[f] = text_fields[f]; /* 1 text (UTF-8), 2 bytes */ l->column[f] = (uint32_t)at; at += FANOUT * (l->text_fields[f] ? sizeof(PureSpan) : sizeof(double)); if (l->text_fields[f]) l->text_field[l->texts++] = (uint8_t)f; @@ -635,7 +635,8 @@ static bool decode_cell(const Layout *l, unsigned field, const unsigned char *by memcpy(&cell->num, &bits, sizeof(bits)); return isfinite(cell->num); } - if (bits > size - *at || !pure_utf8_valid((PureSpan){bytes + *at, (size_t)bits})) return false; + if (bits > size - *at || (l->text_fields[field] == 1 && !pure_utf8_valid((PureSpan){bytes + *at, (size_t)bits}))) + return false; *cell = (PureCell){0, (unsigned char *)bytes + *at, (size_t)bits}; *at += (size_t)bits; return true; @@ -697,7 +698,8 @@ static bool get_compact(const Layout *l, unsigned field, const unsigned char *by *cell = (PureCell){0}; if (!get_varint(bytes, size, at, &value)) return false; if (l->text_fields[field]) { - if (value > size - *at || !pure_utf8_valid((PureSpan){bytes + *at, (size_t)value})) return false; + if (value > size - *at || (l->text_fields[field] == 1 && !pure_utf8_valid((PureSpan){bytes + *at, (size_t)value}))) + return false; *cell = (PureCell){0, (unsigned char *)bytes + *at, (size_t)value}; *at += (size_t)value; return true; diff --git a/core/module-list.h b/core/module-list.h index faacf4b..a4e760a 100644 --- a/core/module-list.h +++ b/core/module-list.h @@ -46,6 +46,8 @@ typedef struct { /* A list holds up to PURE_LIST_ITEMS items, a run up to PURE_LIST_RUN. */ enum { PURE_LIST_ITEMS = 1 << 24, PURE_LIST_RUN = 32 }; +/* text_fields: a field's kind, 0 a number, 1 a text (UTF-8, checked when + * read), 2 bytes (any: a runtime list's). */ PureList *pure_list_new(unsigned fields, const uint8_t *text_fields); void pure_list_free(PureList *); /* Let every item go (retired where a shared copy may read it). */ diff --git a/core/module-save.c b/core/module-save.c index 968e65b..dbf45a9 100644 --- a/core/module-save.c +++ b/core/module-save.c @@ -42,21 +42,27 @@ unsigned long pure_save_test_checkpoints; enum { JOURNAL_MIN = 1 << 20 }; #define LOG_KEPT (UINT64_C(64) << 20) #endif -/* links: "PLK2"; u8 outputs, each {u8 name length, name, u8 field count, the +/* links: "PLK3"; 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); 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. + * dropped, without fields). Before the outbox list, "PLK2" kept the sent + * messages after them, to the end, each {u8 output (in the table), u64le + * number, u32le size, payload}; before the heard list, "PLK1" also kept the + * numbers heard between the two: u16le sources, each {u8 length, the name, + * u64le last number heard}. A start moves both to their lists. * - * 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}; + * The runtime's lists (module.h, PureApp) are stored as list states are, by + * names no state has (a state's is an identifier): the numbers heard + * ("0heard": source text, seq number) and the outbox ("0sent": output text, + * seq number, payload bytes). What only grows with the sources a node hears + * from, or with the sends its receivers have not stored yet, so costs a + * save only the records of its changes. */ +static const unsigned char links_magic[4] = {'P', 'L', 'K', '3'}, links_magic_2[4] = {'P', 'L', 'K', '2'}, + links_magic_1[4] = {'P', 'L', 'K', '1'}; +typedef struct { const char *name; unsigned fields; uint8_t kinds[3]; char names[3][PURE_MODULE_NAME]; } RuntimeList; +static const RuntimeList runtime_lists[PURE_LISTS_STORED - PURE_LIST_HEARD] = { + {"0heard", 2, {1, 0}, {"source", "seq"}}, + {"0sent", 3, {1, 0, 2}, {"output", "seq", "payload"}}, +}; static const PureVersion unsaved = {UINT64_MAX, UINT64_MAX}; static unsigned char *put_le(unsigned char *at, uint64_t value, unsigned width) { @@ -164,16 +170,25 @@ static size_t image_header(const PureModule *m, uint32_t record, uint64_t seq, u 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); } +PureList *pure_runtime_list_new(uint32_t place) { + const RuntimeList *r = &runtime_lists[place - PURE_LIST_HEARD]; + return pure_list_new(r->fields, r->kinds); +} + +static size_t runtime_header(uint32_t place, uint64_t seq, unsigned char *out) { + const RuntimeList *r = &runtime_lists[place - PURE_LIST_HEARD]; + return header_of(r->fields, r->kinds, r->names, seq, out); +} -/* A stored list by its place: a list state, or the heard list (PURE_LIST_HEARD). */ +/* A stored list by its place: a list state, or one of the runtime's. */ static PureList *stored_list(const PureApp *app, uint32_t i) { if (i == PURE_LIST_HEARD) return app->heard; + if (i == PURE_LIST_OUTBOX) return app->outbox; 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]; + return i >= PURE_LIST_HEARD ? runtime_lists[i - PURE_LIST_HEARD].name : app->module->state_names[i]; } static const char *directory_of(const PureApp *app) { @@ -282,7 +297,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_LIST_HEARD; ++i) + for (uint32_t i = 0; i < PURE_LISTS_STORED; ++i) if (app->logged_since[i] && app->logged_since[i] < kept) kept = app->logged_since[i]; pure_log_release(app->log, kept - 1); } @@ -433,24 +448,25 @@ 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) { +PURE_COLD PureResult pure_save_open_runtime(PureApp *app, uint32_t place, PureList **out, PureSaved *saved) { + const char *name = runtime_lists[place - PURE_LIST_HEARD].name; *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); + size_t size = 0, length = runtime_header(place, 0, header) - 8; + PureResult result = pure_image_read(directory_of(app), 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; + result = (*out = pure_runtime_list_new(place)) ? 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; + struct Named *logged = app->start_log ? named_list(app->start_log, name, strlen(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; @@ -470,10 +486,11 @@ PURE_COLD PureResult pure_save_open_heard(PureApp *app, PureList **out, PureSave return PURE_OK; } -PureResult pure_save_heard_image(PureApp *app) { +PureResult pure_save_runtime_image(PureApp *app, uint32_t place) { 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); + size_t header_size = runtime_header(place, app->seq, header); + return write_image(directory_of(app), runtime_lists[place - PURE_LIST_HEARD].name, header, header_size, + stored_list(app, place)); } /* ---- Checkpoints */ @@ -496,7 +513,7 @@ PureCheckpoint *pure_app_checkpoint(PureApp *app) { * past LOG_KEPT, the one that holds it back most: the heard list is one. */ int due = -1, oldest = -1; uint64_t items = 0; - for (uint32_t i = 0; i <= PURE_LIST_HEARD && due < 0; ++i) { + for (uint32_t i = 0; i < PURE_LISTS_STORED && due < 0; ++i) { const PureList *list = stored_list(app, i); size_t size = list ? pure_list_size(list) : 0; items += size; @@ -510,7 +527,7 @@ PureCheckpoint *pure_app_checkpoint(PureApp *app) { 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_save_snapshot(app, pick == PURE_LIST_HEARD))) { + if (!c || !c->directory || !(c->snapshot = pure_save_snapshot(app, pick >= PURE_LIST_HEARD ? (uint32_t)pick : UINT32_MAX))) { if (c) free(c->directory); free(c); return NULL; @@ -518,7 +535,7 @@ PureCheckpoint *pure_app_checkpoint(PureApp *app) { c->state = (uint32_t)pick; snprintf(c->name, sizeof(c->name), "%s", stored_name(app, (uint32_t)pick)); c->seq = app->seq; - c->header_size = pick == PURE_LIST_HEARD ? header_of(2, heard_text, heard_names, c->seq, c->header) : + c->header_size = pick >= PURE_LIST_HEARD ? runtime_header((uint32_t)pick, 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]; @@ -639,8 +656,6 @@ static unsigned char *encode_links(const PureApp *app, size_t *size) { 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 (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; if (!bytes) return NULL; memcpy(at, links_magic, 4); @@ -664,15 +679,7 @@ static unsigned char *encode_links(const PureApp *app, size_t *size) { *at++ = 0; at = put_le(at, app->carried[c].last, 8); } - for (size_t s = 0; s < app->sent_count; ++s) { - const struct PureSent *sent = &app->sent[s]; - if (sent->trimmed) continue; - *at++ = table[sent->output]; - at = put_le(at, sent->seq, 8); - at = put_le(at, sent->size, 4); - if (sent->size) memcpy(at, sent->payload, sent->size); - at += sent->size; - } + (void)table; *size = (size_t)(at - bytes); return bytes; } @@ -687,12 +694,21 @@ PURE_COLD static PureResult heard_before(PureApp *app, const unsigned char *sour return pure_list_set(app->heard, at, 1, (PureCell){.num = (double)seq}, NULL); } +/* A send, as links kept it before the outbox list: at the outbox's end. */ +PURE_COLD static PureResult sent_before(PureApp *app, const char *output, uint64_t seq, const unsigned char *payload, + size_t size) { + PureCell item[3] = {{.text = (unsigned char *)output, .size = strlen(output)}, {.num = (double)seq}, + {.text = (unsigned char *)payload, .size = size}}; + return pure_list_insert(app->outbox, pure_list_count(app->outbox), item, 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 kept before the heard list, and sent messages. */ -PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, bool *moved) { - *moved = false; + * sends left in the outbox: CONFLICT; one without sends keeps its number + * for later); the sources and sends links kept before the runtime's lists + * go to them (heard_moved, sent_moved: they need images). */ +PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, bool *heard_moved, bool *sent_moved) { + *heard_moved = *sent_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; @@ -700,12 +716,14 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, boo if ((result = read_whole(app->stored_links, &bytes, &size)) != PURE_OK) return result; int output[256]; bool fits[256]; + char names[256][PURE_MODULE_NAME]; size_t at = 5; - bool first = size >= 5 && !memcmp(bytes, links_magic_1, 4), known = first || (size >= 5 && !memcmp(bytes, links_magic, 4)); + bool first = size >= 5 && !memcmp(bytes, links_magic_1, 4), second = size >= 5 && !memcmp(bytes, links_magic_2, 4), + known = first || second || (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]; + char *name = names[t]; size_t length = at < size ? bytes[at] : 0; if (!length || length >= PURE_MODULE_NAME || size - at < 2 + length) { result = PURE_STORE_ERROR; break; } memcpy(name, bytes + at + 1, length); @@ -738,32 +756,31 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, boo 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; + *heard_moved = true; app->links_changed = true; /* stored again without them */ } - while (result == PURE_OK && at < size) { + while (result == PURE_OK && (first || second) && at < size) { /* sends kept before the outbox list */ if (size - at < 13 || bytes[at] >= outputs) { result = PURE_STORE_ERROR; break; } unsigned t = bytes[at]; uint64_t seq = get_le(bytes + at + 1, 8); size_t length = (size_t)get_le(bytes + at + 9, 4); at += 13; - if (size - at < length) { result = PURE_STORE_ERROR; break; } - if (!fits[t]) { result = PURE_CONFLICT; break; } - if (app->sent_count == app->sent_capacity) { - size_t capacity = app->sent_capacity ? 2 * app->sent_capacity : 16; - struct PureSent *grown = realloc(app->sent, capacity * sizeof(*grown)); - if (!grown) { result = PURE_ALLOCATION; break; } - app->sent = grown; - app->sent_capacity = capacity; - } - unsigned char *payload = malloc(length ? length : 1); - if (!payload) { result = PURE_ALLOCATION; break; } - if (length) memcpy(payload, bytes + at, length); - app->sent[app->sent_count++] = (struct PureSent){(uint32_t)output[t], seq, payload, length, false}; + if (size - at < length || seq > PURE_SEQ_MOST) { result = PURE_STORE_ERROR; break; } + result = sent_before(app, names[t], seq, bytes + at, length); at += length; + *sent_moved = true; + app->links_changed = true; + } + if (result == PURE_OK && !first && !second && at != size) result = PURE_STORE_ERROR; + /* Every send left goes out on an output of this module with its fields. */ + for (size_t k = 0, count = pure_list_count(app->outbox); result == PURE_OK && k < count; ++k) { + PureCell name = pure_list_get(app->outbox, k, 0); + unsigned t = 0; + while (t < outputs && (strlen(names[t]) != name.size || memcmp(names[t], name.text, name.size))) ++t; + if (t == outputs || !fits[t]) result = PURE_CONFLICT; } free(bytes); - app->sent_durable = app->sent_count; + app->sent_saved = pure_list_count(app->outbox); memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent)); return result; } @@ -772,12 +789,12 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, boo * its name and its records. NULL with size 0 when no list changed. */ static unsigned char *list_changes(const PureApp *app, size_t *size) { *size = 0; - for (uint32_t i = 0; i <= PURE_LIST_HEARD; ++i) { + for (uint32_t i = 0; i < PURE_LISTS_STORED; ++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 <= PURE_LIST_HEARD; ++i) { + for (uint32_t i = 0; payload && i < PURE_LISTS_STORED; ++i) { const PureList *list = stored_list(app, i); PureSpan changes = list ? pure_list_changes(list) : (PureSpan){0}; if (!changes.size) continue; @@ -888,7 +905,7 @@ 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 <= PURE_LIST_HEARD; ++i) { + for (uint32_t i = 0; i < PURE_LISTS_STORED; ++i) { PureList *list = stored_list(app, i); PureSpan changes = list ? pure_list_changes(list) : (PureSpan){0}; if (!changes.size) continue; diff --git a/core/module-save.h b/core/module-save.h index 32e68c7..1d6440d 100644 --- a/core/module-save.h +++ b/core/module-save.h @@ -24,27 +24,29 @@ 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 state's, or the heard list's - * (PURE_LIST_HEARD) in a snapshot made with it (pure_save_snapshot). */ +/* A snapshot's list (module.c): a state's, or a runtime list's (place + * PURE_LIST_HEARD on) in a snapshot made with it (pure_save_snapshot; + * UINT32_MAX: none). */ 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); +PureSnapshot *pure_save_snapshot(PureApp *, uint32_t runtime); +/* A runtime list of a starting app (module.h, PureApp; place is + * PURE_LIST_HEARD or PURE_LIST_OUTBOX): from its image with its logged + * changes made again, or new (saved.rewrite: it needs an image, which + * pure_save_runtime_image writes). */ +PureResult pure_save_open_runtime(PureApp *, uint32_t place, PureList **out, PureSaved *saved); +PureResult pure_save_runtime_image(PureApp *, uint32_t place); +/* An empty runtime list. */ +PureList *pure_runtime_list_new(uint32_t place); /* 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. 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); +/* The stored links of a starting app (module.h, PureApp), after its runtime + * lists; CONFLICT when an output with sends left in the outbox is missing or + * has other fields. Links stored before the runtime lists kept the numbers + * heard and the sends: they go to them (moved: they need images). */ +PureResult pure_save_open_links(PureApp *, const PureModule *m, bool *heard_moved, bool *sent_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 3a694a1..cb4fe1a 100644 --- a/core/module.c +++ b/core/module.c @@ -1294,13 +1294,6 @@ static bool copy_out(void *context, PureSpan span) { static PureResult send(PureApp *app, uint32_t output, const Slot *fields, const PureArena *arena, size_t size) { const PureModule *m = app->module; const PureInputShape *shape = &m->input[output]; - if (app->sent_count == app->sent_capacity) { - size_t capacity = app->sent_capacity ? 2 * app->sent_capacity : 16; - struct PureSent *grown = realloc(app->sent, capacity * sizeof(*grown)); - if (!grown) return PURE_ALLOCATION; - app->sent = grown; - app->sent_capacity = capacity; - } unsigned char *payload = malloc(size ? size : 1); if (!payload) return PURE_ALLOCATION; Copy copy = {payload + shape->prefix}; @@ -1318,7 +1311,14 @@ static PureResult send(PureApp *app, uint32_t output, const Slot *fields, const if (result != PURE_OK) { free(payload); return result; } } } - app->sent[app->sent_count++] = (struct PureSent){output, ++app->last_sent[output], payload, size, false}; + /* The outbox's next item: a change of the list, undone with the event. */ + const char *name = m->input_names[output]; + PureCell item[3] = {{.text = (unsigned char *)name, .size = strlen(name)}, {.num = (double)(app->last_sent[output] + 1)}, + {.text = payload, .size = size}}; + PureResult result = pure_list_insert(app->outbox, pure_list_count(app->outbox), item, &app->undo); + free(payload); + if (result != PURE_OK) return result; + ++app->last_sent[output]; app->links_changed = true; return PURE_OK; } @@ -2443,6 +2443,8 @@ 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; + if (app->outbox) app->durable_outbox = pure_list_changes(app->outbox).size; + app->durable_sent_saved = app->sent_saved; app->open = true; } @@ -2464,6 +2466,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 (app->outbox && !keep) pure_list_forget(app->outbox, app->durable_outbox); if (keep) pure_undo_commit(&app->undo); else { pure_undo_rollback(&app->undo, 0); @@ -2475,25 +2478,16 @@ static void close_burst(PureApp *app, bool keep) { app->open = false; } -static void drop_sent(PureApp *app, size_t count) { - while (app->sent_count > count) free(app->sent[--app->sent_count].payload); -} - -/* 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 (the numbers - * heard go back with the burst's list changes). */ +/* After a save: what it stored is the links' saved state, every send in the + * outbox saved; or, when it failed, the links return to the last saved + * state (the sends, trims and numbers heard go back with the burst's list + * changes). */ static void settle_links(PureApp *app, bool saved) { if (saved) { - size_t kept = 0; - for (size_t i = 0; i < app->sent_count; ++i) { - if (app->sent[i].trimmed) free(app->sent[i].payload); - else app->sent[kept++] = app->sent[i]; - } - app->sent_count = app->sent_durable = kept; + app->sent_saved = app->outbox ? pure_list_count(app->outbox) : 0; memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent)); } else { - drop_sent(app, app->sent_durable); - for (size_t i = 0; i < app->sent_count; ++i) app->sent[i].trimmed = false; + app->sent_saved = app->durable_sent_saved; memcpy(app->last_sent, app->saved_sent, sizeof(app->last_sent)); } app->links_changed = false; @@ -2531,7 +2525,7 @@ 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 */ + PureList *runtime[PURE_LISTS_STORED - PURE_LIST_HEARD]; /* only in a checkpoint's of one */ PureRetired retired; struct PureSnapshot *newer; }; @@ -2539,7 +2533,7 @@ struct PureSnapshot { 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 state == PURE_LIST_HEARD ? snapshot->heard : snapshot->lists[state]; + return state >= PURE_LIST_HEARD ? snapshot->runtime[state - PURE_LIST_HEARD] : snapshot->lists[state]; } void pure_snapshot_release(PureSnapshot *snapshot) { @@ -2549,7 +2543,7 @@ void pure_snapshot_release(PureSnapshot *snapshot) { pure_draft_close(snapshot->texts[i]); pure_list_free(snapshot->lists[i]); } - pure_list_free(snapshot->heard); + 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; @@ -2571,8 +2565,7 @@ void pure_app_stop(PureApp *app) { } else pure_retired_free(&app->retired); pure_draft_close(app->stored_values); - drop_sent(app, 0); - free(app->sent); + pure_list_free(app->outbox); pure_list_free(app->heard); free(app->carried); pure_draft_close(app->stored_links); @@ -2592,8 +2585,8 @@ 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], heard; /* stored lists' journal lengths */ - size_t undo, sent; + size_t lists[PURE_MODULE_STATES], heard, outbox; /* stored lists' journal lengths */ + size_t undo; uint64_t last_sent[PURE_MODULE_INPUTS]; bool opened; } Mark; @@ -2606,8 +2599,8 @@ static void mark_event(PureApp *app, Mark *mark) { 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->outbox = app->outbox ? pure_list_changes(app->outbox).size : 0; mark->undo = app->undo.count; - mark->sent = app->sent_count; memcpy(mark->last_sent, app->last_sent, sizeof(mark->last_sent)); } @@ -2617,8 +2610,8 @@ static void undo_event(PureApp *app, const Mark *mark) { if (app->lists[i]) pure_list_forget(app->lists[i], mark->lists[i]); } if (app->heard) pure_list_forget(app->heard, mark->heard); + if (app->outbox) pure_list_forget(app->outbox, mark->outbox); 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 */ } @@ -2864,26 +2857,53 @@ PureResult pure_app_message_fields(PureApp *app, uint32_t input, const char *key return message(app, input, key, base, payload, placed); } +/* The outbox's item at: its output (the module's, by name), number and payload. */ +static bool sent_at(const PureApp *app, size_t at, uint32_t *output, uint64_t *seq, PureSpan *payload) { + PureCell name = pure_list_get(app->outbox, at, 0), bytes = pure_list_get(app->outbox, at, 2); + const PureModule *m = app->module; + uint32_t i = 0; + while (i < m->inputs && (strlen(m->input_names[i]) != name.size || memcmp(m->input_names[i], name.text, name.size))) ++i; + *output = i; + *seq = (uint64_t)pure_list_get(app->outbox, at, 1).num; + *payload = (PureSpan){bytes.text, bytes.size}; + return i < m->inputs; +} + void pure_app_outbox(const PureApp *app, PureSentVisit visit, void *context) { - for (size_t i = 0; i < app->sent_durable; ++i) { - const struct PureSent *sent = &app->sent[i]; - if (!sent->trimmed && !visit(context, sent->output, sent->seq, (PureSpan){sent->payload, sent->size})) return; + for (size_t i = 0; i < app->sent_saved; ++i) { + uint32_t output; + uint64_t seq; + PureSpan payload; + if (sent_at(app, i, &output, &seq, &payload) && !visit(context, output, seq, payload)) return; } } void pure_app_outbox_newest(const PureApp *app, PureSentVisit visit, void *context) { - for (size_t i = app->sent_durable; i-- > 0;) { - const struct PureSent *sent = &app->sent[i]; - if (!sent->trimmed && !visit(context, sent->output, sent->seq, (PureSpan){sent->payload, sent->size})) return; + for (size_t i = app->sent_saved; i-- > 0;) { + uint32_t output; + uint64_t seq; + PureSpan payload; + if (sent_at(app, i, &output, &seq, &payload) && !visit(context, output, seq, payload)) return; } } +/* An output's sends up to through go at once (for the host), and from the + * store with the next save; a failed save brings them back. Its sends are + * in order among the others: the look stops at its first one kept. */ void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) { - for (size_t i = 0; i < app->sent_durable; ++i) - if (app->sent[i].output == output && app->sent[i].seq <= through && !app->sent[i].trimmed) { - app->sent[i].trimmed = true; - app->links_changed = true; - } + const PureModule *m = app->module; + if (!app->outbox || output >= m->inputs) return; + const char *name = m->input_names[output]; + size_t length = strlen(name); + open_burst(app); + for (size_t i = 0; i < app->sent_saved;) { + PureCell of = pure_list_get(app->outbox, i, 0); + if (of.size != length || memcmp(of.text, name, length)) { ++i; continue; } + if ((uint64_t)pure_list_get(app->outbox, i, 1).num > through) break; + if (pure_list_remove(app->outbox, i, &app->undo) != PURE_OK) break; + --app->sent_saved; + app->links_changed = true; + } } PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload, @@ -3093,9 +3113,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) { return pure_save_snapshot(app, false); } +PureSnapshot *pure_app_snapshot(PureApp *app) { return pure_save_snapshot(app, UINT32_MAX); } -PureSnapshot *pure_save_snapshot(PureApp *app, bool heard) { +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; @@ -3114,7 +3134,10 @@ PureSnapshot *pure_save_snapshot(PureApp *app, bool heard) { } 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 && 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; + } if (!ok) { pure_snapshot_release(made); return NULL; @@ -3207,8 +3230,11 @@ PURE_COLD static PureResult remap_outputs(PureApp *app, const PureModule *old, c for (uint32_t f = 0; same && f < old->input[i].field_count; ++f) same = module->field[module->input[to].field_start + f].type == old->field[old->input[i].field_start + f].type; map[i] = same ? to : -1; - for (size_t s = 0; !same && s < app->sent_count; ++s) - if (app->sent[s].output == i) return PURE_CONFLICT; + size_t length = strlen(old->input_names[i]); + for (size_t s = 0, count = app->outbox ? pure_list_count(app->outbox) : 0; !same && s < count; ++s) { + PureCell name = pure_list_get(app->outbox, s, 0); + if (name.size == length && !memcmp(name.text, old->input_names[i], length)) return PURE_CONFLICT; + } } if (!apply) return PURE_OK; uint64_t last[PURE_MODULE_INPUTS] = {0}, saved[PURE_MODULE_INPUTS] = {0}; @@ -3233,7 +3259,6 @@ PURE_COLD static PureResult remap_outputs(PureApp *app, const PureModule *old, c app->carried[c] = app->carried[--app->carried_count]; app->links_changed = true; } - for (size_t s = 0; s < app->sent_count; ++s) app->sent[s].output = (uint32_t)map[app->sent[s].output]; memcpy(app->last_sent, last, sizeof(last)); memcpy(app->saved_sent, saved, sizeof(saved)); return PURE_OK; @@ -3253,11 +3278,14 @@ 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}; + /* The numbers heard and the outbox belong to the app, whatever its module. */ + PureSaved heard = {0}, sent = {0}; + if (!old && result == PURE_OK) + result = lists_stored ? pure_save_open_runtime(app, PURE_LIST_HEARD, &app->heard, &heard) : + (app->heard = pure_runtime_list_new(PURE_LIST_HEARD)) ? PURE_OK : PURE_ALLOCATION; 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; + result = lists_stored ? pure_save_open_runtime(app, PURE_LIST_OUTBOX, &app->outbox, &sent) : + (app->outbox = pure_runtime_list_new(PURE_LIST_OUTBOX)) ? 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; @@ -3312,9 +3340,13 @@ 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. */ - 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; + bool heard_moved = false, sent_moved = false; + if (result == PURE_OK && !old && !stored) app->sent_saved = pure_list_count(app->outbox); + if (result == PURE_OK) + result = old ? remap_outputs(app, old, module, false) : + stored ? pure_save_open_links(app, module, &heard_moved, &sent_moved) : PURE_OK; heard.rewrite = heard.rewrite || (heard_moved && lists_stored); + sent.rewrite = sent.rewrite || (sent_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])) @@ -3336,8 +3368,10 @@ 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) + if (result == PURE_OK && heard.rewrite && (result = pure_save_runtime_image(app, PURE_LIST_HEARD)) == PURE_OK) heard = (PureSaved){.image = app->seq}; + if (result == PURE_OK && sent.rewrite && (result = pure_save_runtime_image(app, PURE_LIST_OUTBOX)) == PURE_OK) + sent = (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) { @@ -3377,7 +3411,10 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) 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); + app->image_seq[PURE_LIST_OUTBOX] = sent.image; + app->logged_since[PURE_LIST_OUTBOX] = sent.since; + app->logged[PURE_LIST_OUTBOX] = sent.logged; + if (lists_stored) pure_list_journal(app->heard, true), pure_list_journal(app->outbox, true); } pure_save_release(app); for (uint32_t i = 0; i < module->states; ++i) { diff --git a/core/module.h b/core/module.h index 8176bea..3431c87 100644 --- a/core/module.h +++ b/core/module.h @@ -263,8 +263,9 @@ 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 }; +/* The runtime's lists' places among the stored lists' records (PureApp): + * the numbers heard, and the outbox. */ +enum { PURE_LIST_HEARD = PURE_MODULE_STATES, PURE_LIST_OUTBOX, PURE_LISTS_STORED }; typedef struct { const PureModule *module; PureData data; @@ -273,13 +274,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), by state, and the heard list's last - * (PURE_LIST_HEARD): the save each one's image holds, the first save + /* Stored lists (module-save.c), by state, then the runtime's (the heard + * list and the outbox): 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]; + uint64_t image_seq[PURE_LISTS_STORED], logged_since[PURE_LISTS_STORED]; + size_t logged[PURE_LISTS_STORED]; struct PureLog *log; PureDraft *stored_seq; uint64_t seq; @@ -320,12 +321,15 @@ typedef struct { unsigned char *copied; int64_t revision; PureArena arena, view_arena; - /* 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), 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; + /* Links (runtime-owned, saved with the states): the outbox, messages sent + * on outputs and not trimmed, in the order sent, a list of {output (its + * name), seq, payload (bytes)} stored as a list state is (module-save.c) + * and changed with the events that send (the first sent_saved are saved: + * the host sees only those; a trim lets go at once, and a failed save + * takes it back); each output's last number (and as saved), and those of + * outputs a module dropped. */ + PureList *outbox; + size_t sent_saved, durable_sent_saved, durable_outbox; 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; diff --git a/core/test-module.c b/core/test-module.c index 9a1f8a4..d03a27b 100644 --- a/core/test-module.c +++ b/core/test-module.c @@ -3787,6 +3787,15 @@ static void message_fields(void) { enum { SEND = PURE_OP_SEND }; typedef struct { uint32_t output; uint64_t seq; unsigned char payload[64]; size_t size; } Sent; typedef struct { Sent items[16]; size_t count; } Outbox; +typedef struct { size_t count; uint64_t first, last; } Tally; + +static bool tally_sent(void *context, uint32_t output, uint64_t seq, PureSpan payload) { + (void)output, (void)payload; + Tally *t = context; + if (!t->count++) t->first = seq; + t->last = seq; + return true; +} static bool collect_sent(void *context, uint32_t output, uint64_t seq, PureSpan payload) { Outbox *o = context; @@ -3875,7 +3884,7 @@ static void links(void) { CHECK(pure_app_message(&a, 2, NULL, -1, sent) == PURE_INVALID); /* Trimmed, a send is gone at once for the host and from the store at the next save. */ pure_app_trim(&a, 2, 1); - CHECK(outbox_of(&a).count == 0 && pure_app_save(&a) == PURE_OK && a.sent_count == 0); + CHECK(outbox_of(&a).count == 0 && pure_app_save(&a) == PURE_OK && pure_list_count(a.outbox) == 0); CHECK(buy(&a, 0, "cap", 1) == PURE_OK && outbox_of(&a).items[0].seq == 2); pure_app_stop(&a); pure_app_stop(&b); @@ -3951,8 +3960,59 @@ static void links(void) { 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)); + CHECK(pure_state_open(path_b, "links", &stored_links) == PURE_OK && !memcmp(read_state(stored_links), "PLK3", 4)); + pure_draft_close(stored_links); + /* Links stored before the outbox list kept the sends: a start moves them + * to it; numbering goes on from the output's last number. */ + char path_c[64]; + PureApp c; + snprintf(path_c, sizeof(path_c), "%s", new_store()); + unsigned char sends[128] = "PLK2"; + at = 5; + sends[4] = 1; /* one output: ordered {item, count, total} */ + sends[at++] = 7; + memcpy(sends + at, "ordered", 7); + at += 7; + sends[at++] = 3, sends[at++] = PURE_FIELD_TEXT, sends[at++] = PURE_FIELD_U32, sends[at++] = PURE_FIELD_NUMBER; + for (unsigned i = 0; i < 8; ++i) sends[at++] = i ? 0 : 2; /* its last number */ + for (unsigned k = 1; k <= 2; ++k) { + sends[at++] = 0; + for (unsigned i = 0; i < 8; ++i) sends[at++] = i ? 0 : (unsigned char)k; + for (unsigned i = 0; i < 4; ++i) sends[at++] = i ? 0 : 4; + memcpy(sends + at, "abcd", 4); + at += 4; + } + CHECK(pure_state_open(path_c, "links", &stored_links) == PURE_OK); + CHECK(pure_draft_splice(stored_links, 0, pure_draft_size(stored_links), (PureSpan){sends, at}) == PURE_OK); + CHECK(pure_draft_commit(stored_links, &revision) == PURE_OK); + pure_draft_close(stored_links); + CHECK(pure_app_start(&c, &module_a, pure_tree_data(path_c)) == PURE_OK); + o = outbox_of(&c); + CHECK(o.count == 2 && o.items[0].output == 2 && o.items[0].seq == 1 && o.items[1].seq == 2 && o.items[1].size == 4 && + !memcmp(o.items[1].payload, "abcd", 4)); + CHECK(buy(&c, 0, "next", 1) == PURE_OK && outbox_of(&c).items[2].seq == 3); + pure_app_stop(&c); + CHECK(pure_state_open(path_c, "links", &stored_links) == PURE_OK && !memcmp(read_state(stored_links), "PLK3", 4)); + pure_draft_close(stored_links); + /* A large outbox costs a save only its changes: the links stay small, the + * sends and trims come back after a start. */ + CHECK(pure_app_start(&c, &module_a, pure_tree_data(path_c)) == PURE_OK && pure_list_count(c.outbox) == 3); + CHECK(pure_app_defer(&c, true)); + for (unsigned k = 0; k < 20000; ++k) { + CHECK(buy(&c, 0, "many", 1) == PURE_OK); + if (k % 1000 == 999) CHECK(pure_app_save(&c) == PURE_OK); + } + CHECK(pure_state_open(path_c, "links", &stored_links) == PURE_OK && pure_draft_size(stored_links) < 256); pure_draft_close(stored_links); + pure_app_trim(&c, 2, 15000); + CHECK(pure_list_count(c.outbox) == 20003 - 15000 && pure_app_save(&c) == PURE_OK); + pure_app_stop(&c); + CHECK(pure_app_start(&c, &module_a, pure_tree_data(path_c)) == PURE_OK); + Tally tally = {0}; + pure_app_outbox(&c, tally_sent, &tally); + CHECK(tally.count == 5003 && tally.first == 15001 && tally.last == 20003); + pure_app_stop(&c); + remove_store(path_c); /* 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 |