diff options
Diffstat (limited to 'core')
| -rw-r--r-- | core/module-list.c | 222 | ||||
| -rw-r--r-- | core/module-list.h | 29 | ||||
| -rw-r--r-- | core/module-save.c | 149 | ||||
| -rw-r--r-- | core/module-save.h | 16 | ||||
| -rw-r--r-- | core/module.c | 35 | ||||
| -rw-r--r-- | core/module.h | 4 | ||||
| -rw-r--r-- | core/test-module.c | 85 |
7 files changed, 356 insertions, 184 deletions
diff --git a/core/module-list.c b/core/module-list.c index 456a72e..4d90d32 100644 --- a/core/module-list.c +++ b/core/module-list.c @@ -50,8 +50,11 @@ struct PureList { Layout layout; Node *root; /* NULL when the list has no item */ size_t count; - PureDraft *draft; /* the stored copy, or NULL */ - size_t header; /* draft bytes before the first item */ + /* A stored list's changes since they were last taken, as records + * (pure_list_replay); none are kept while journaled is false. */ + bool journaled; + 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 */ PureRetired *retired; /* where shared nodes and texts go when the list lets them go */ @@ -193,12 +196,6 @@ static void drop_node(PureList *list, Node *node) { static size_t cell_bytes(const Layout *l, unsigned field, PureCell cell) { return l->text_fields[field] ? 4 + cell.size : 8; } -static size_t item_bytes(const Layout *l, const Node *leaf, unsigned i) { - size_t size = l->number_bytes; - for (unsigned t = 0; t < l->texts; ++t) size += 4 + texts(l, leaf, l->text_field[t])[i].size; - return size; -} - /* A node's summary, from its items or children. */ static void sum(const PureList *list, Node *node) { const Layout *l = &list->layout; @@ -241,27 +238,6 @@ static const Node *leaf_of(const PureList *list, size_t at, size_t *within) { return node; } -/* Where an item starts in the draft: the bytes of the children before it, - * level by level. */ -static size_t item_offset(const PureList *list, size_t at) { - const Layout *l = &list->layout; - size_t offset = list->header; - const Node *node = list->root; - if (!node) return offset; - while (!node->leaf) { - unsigned i = 0; - while (i + 1 < node->count && at >= child(node, i)->items) { - at -= child(node, i)->items; - offset += child(node, i)->bytes; - ++i; - } - node = child(node, i); - } - if (!l->texts) return offset + (at < node->count ? at : node->count) * l->number_bytes; - for (unsigned i = 0; i < at && i < node->count; ++i) offset += item_bytes(l, node, i); - return offset; -} - /* The node at *place, the list's own to change: a copy (the node retired) * while a shared copy of the list may read it. NULL without memory. A copy * moves the version on even when the change it is for does not happen: the @@ -481,6 +457,7 @@ void pure_list_free(PureList *list) { if (list->copy) { if (__atomic_sub_fetch(&list->refs, 1, __ATOMIC_ACQ_REL)) return; } else { + free(list->journal); let_go_tree(list, list->root); for (unsigned kind = 0; kind < 2; ++kind) while (list->spare[kind]) { @@ -553,17 +530,66 @@ static unsigned char *encode_cell(const Layout *l, unsigned field, PureCell cell return out; } -/* Replace remove bytes of the draft, within bytes into the item at `at` (or - * after the last item), with bytes; nothing to do without a draft. */ -static PureResult mirror(PureList *list, size_t at, size_t within, size_t remove, const unsigned char *bytes, - size_t size) { - if (!list->draft) return PURE_OK; - return pure_draft_splice(list->draft, item_offset(list, at) + within, remove, (PureSpan){bytes, size}); +/* A stored cell of field at *at in bytes (a text's bytes borrowed from + * them): false when it is cut short, not finite or not UTF-8. */ +static bool decode_cell(const Layout *l, unsigned field, const unsigned char *bytes, size_t size, size_t *at, + PureCell *cell) { + size_t width = l->text_fields[field] ? 4 : 8; + if (size - *at < width) return false; + uint64_t bits = 0; + for (size_t i = 0; i < width; ++i) bits |= (uint64_t)bytes[*at + i] << (8 * i); + *at += width; + *cell = (PureCell){0}; + if (!l->text_fields[field]) { + 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; + *cell = (PureCell){0, (unsigned char *)bytes + *at, (size_t)bits}; + *at += (size_t)bits; + return true; } -void pure_list_mirror(PureList *list, PureDraft *draft, size_t header) { - list->draft = draft; - list->header = header; +/* ---- The journal: records of a stored list's changes, in the order they + * were made. Each is a u8 kind and the u32le position, then for an insert + * the item's fields in stored form, for a set the u8 field and its cell. */ + +enum { RECORD_INSERT = 1, RECORD_REMOVE = 2, RECORD_SET = 3, RECORD_HEAD = 5, JOURNAL_KEPT = 1 << 20 }; + +/* Room for size more bytes at the journal's end, or NULL without memory. */ +static unsigned char *journal_room(PureList *list, size_t size) { + if (list->journal_capacity - list->journal_size < size) { + size_t capacity = list->journal_capacity ? 2 * list->journal_capacity : 256; + while (capacity - list->journal_size < size) capacity *= 2; + unsigned char *grown = realloc(list->journal, capacity); + if (!grown) return NULL; + list->journal = grown; + list->journal_capacity = capacity; + } + return list->journal + list->journal_size; +} + +static unsigned char *record_head(unsigned char *out, unsigned kind, size_t at) { + *out++ = (unsigned char)kind; + for (unsigned i = 0; i < 4; ++i) *out++ = (unsigned char)(at >> (8 * i)); + return out; +} + +void pure_list_journal(PureList *list, bool journaled) { + list->journaled = journaled; + if (!journaled) pure_list_forget(list, 0); +} + +PureSpan pure_list_changes(const PureList *list) { return (PureSpan){list->journal, list->journal_size}; } + +void pure_list_forget(PureList *list, size_t size) { + if (size >= list->journal_size) return; + list->journal_size = size; + if (!size && list->journal_capacity > JOURNAL_KEPT) { /* a bulk load's room goes back */ + free(list->journal); + list->journal = NULL; + list->journal_capacity = 0; + } } /* Every leaf in order, to visit (false stops it). */ @@ -584,44 +610,58 @@ static bool encode_leaf(void *context, const Node *leaf) { return true; } -PURE_COLD PureResult pure_list_write(PureList *list) { - size_t size = list->root ? list->root->bytes : 0; - unsigned char *bytes = size ? malloc(size) : NULL; - if (size && !bytes) return PURE_ALLOCATION; - Encoding encoding = {&list->layout, bytes}; +size_t pure_list_size(const PureList *list) { return list->root ? list->root->bytes : 0; } + +void pure_list_encode(const PureList *list, unsigned char *out) { + Encoding encoding = {&list->layout, out}; visit_leaves(list->root, encode_leaf, &encoding); - PureResult result = mirror(list, 0, 0, 0, bytes, size); - free(bytes); - return result; +} + +/* An item's cells in stored form at *at in bytes. */ +static bool decode_item(const Layout *l, const unsigned char *bytes, size_t size, size_t *at, PureCell *cells) { + for (unsigned f = 0; f < l->fields; ++f) + if (!decode_cell(l, f, bytes, size, at, &cells[f])) return false; + return true; } PURE_COLD PureResult pure_list_read(PureList *list, const unsigned char *bytes, size_t size) { - const Layout *l = &list->layout; PureCell cells[16]; for (size_t at = 0; at < size;) { - for (unsigned f = 0; f < l->fields; ++f) { - size_t width = l->text_fields[f] ? 4 : 8; - if (size - at < width) return PURE_STORE_ERROR; - uint64_t bits = 0; - for (size_t i = 0; i < width; ++i) bits |= (uint64_t)bytes[at + i] << (8 * i); - at += width; - if (!l->text_fields[f]) { - cells[f] = (PureCell){0}; - memcpy(&cells[f].num, &bits, sizeof(bits)); - if (!isfinite(cells[f].num)) return PURE_STORE_ERROR; - continue; - } - if (bits > size - at || !pure_utf8_valid((PureSpan){bytes + at, (size_t)bits})) - return PURE_STORE_ERROR; - cells[f] = (PureCell){0, (unsigned char *)bytes + at, (size_t)bits}; - at += (size_t)bits; - } + if (!decode_item(&list->layout, bytes, size, &at, cells)) return PURE_STORE_ERROR; PureResult result = pure_list_insert(list, list->count, cells, NULL); if (result != PURE_OK) return result; } return PURE_OK; } +PURE_COLD PureResult pure_list_replay(PureList *list, const unsigned char *bytes, size_t size) { + const Layout *l = &list->layout; + bool journaled = list->journaled; + PureResult result = PURE_OK; + list->journaled = false; + for (size_t at = 0; at < size && result == PURE_OK;) { + if (size - at < RECORD_HEAD) { result = PURE_STORE_ERROR; break; } + unsigned kind = bytes[at]; + size_t position = 0; + for (unsigned i = 0; i < 4; ++i) position |= (size_t)bytes[at + 1 + i] << (8 * i); + at += RECORD_HEAD; + PureCell cells[16]; + if (kind == RECORD_INSERT) + result = position <= list->count && decode_item(l, bytes, size, &at, cells) ? + pure_list_insert(list, position, cells, NULL) : PURE_STORE_ERROR; + else if (kind == RECORD_REMOVE) + result = position < list->count ? pure_list_remove(list, position, NULL) : PURE_STORE_ERROR; + else if (kind == RECORD_SET) { + unsigned field = at < size ? bytes[at++] : l->fields; + result = field < l->fields && position < list->count && decode_cell(l, field, bytes, size, &at, &cells[0]) ? + pure_list_set(list, position, field, cells[0], NULL) : PURE_STORE_ERROR; + } else + result = PURE_STORE_ERROR; + } + list->journaled = journaled; + return result; +} + PureResult pure_list_insert(PureList *list, size_t at, const PureCell *cells, PureUndo *undo) { const Layout *l = &list->layout; if (at > list->count || list->count == PURE_LIST_ITEMS) return PURE_LIMIT; @@ -629,29 +669,23 @@ PureResult pure_list_insert(PureList *list, size_t at, const PureCell *cells, Pu unsigned owned = 0; for (; owned < l->fields; ++owned) if (!own_cell(list, owned, cells[owned], &item[owned])) break; - size_t size = 0; - for (unsigned f = 0; list->draft && f < l->fields; ++f) size += cell_bytes(l, f, cells[f]); - unsigned char *bytes = size ? malloc(size) : NULL, *end = bytes; + size_t size = RECORD_HEAD; + for (unsigned f = 0; list->journaled && f < l->fields; ++f) size += cell_bytes(l, f, cells[f]); + unsigned char *record = list->journaled ? journal_room(list, size) : NULL; Place p; - if (owned < l->fields || (size && !bytes) || !prepare(list, at, &p)) { - free(bytes); + if (owned < l->fields || (list->journaled && !record) || !prepare(list, at, &p)) { for (unsigned f = 0; f < owned; ++f) free_text(item[f].text); return PURE_ALLOCATION; } - for (unsigned f = 0; bytes && f < l->fields; ++f) end = encode_cell(l, f, item[f], end); if (undo && !log_change(undo, (Entry){.change = INSERTED, .list = list, .at = at})) { release_extras(list, &p); - free(bytes); for (unsigned f = 0; f < l->fields; ++f) free_text(item[f].text); return PURE_ALLOCATION; } - PureResult result = mirror(list, at, 0, 0, bytes, size); - free(bytes); - if (result != PURE_OK) { - if (undo) --undo->count; - release_extras(list, &p); - for (unsigned f = 0; f < l->fields; ++f) free_text(item[f].text); - return result; + if (record) { + unsigned char *end = record_head(record, RECORD_INSERT, at); + for (unsigned f = 0; f < l->fields; ++f) end = encode_cell(l, f, item[f], end); + list->journal_size += size; } place(list, &p, item); return PURE_OK; @@ -677,15 +711,16 @@ PureResult pure_list_remove(PureList *list, size_t at, PureUndo *undo) { unsigned within, depth = descend(list, at, path, &within); if (!depth) return PURE_ALLOCATION; Node *leaf = path[depth - 1].node; + unsigned char *record = list->journaled ? journal_room(list, RECORD_HEAD) : NULL; PureCell *cells = undo ? log_cells(undo, l->fields) : NULL, kept[16]; - if (undo && (!cells || !log_change(undo, (Entry){.change = REMOVED, .list = list, .at = at, .cells = undo->cell_count}))) + if ((list->journaled && !record) || + (undo && (!cells || !log_change(undo, (Entry){.change = REMOVED, .list = list, .at = at, .cells = undo->cell_count})))) return PURE_ALLOCATION; if (!cells) cells = kept; for (unsigned f = 0; f < l->fields; ++f) cells[f] = cell_at(l, leaf, within, f); - PureResult result = mirror(list, at, 0, item_bytes(l, leaf, within), NULL, 0); - if (result != PURE_OK) { - if (undo) --undo->count; - return result; + if (record) { + record_head(record, RECORD_REMOVE, at); + list->journal_size += RECORD_HEAD; } if (undo) undo->cell_count += l->fields; else let_go_cells(list, cells); @@ -702,23 +737,18 @@ PureResult pure_list_set(PureList *list, size_t at, unsigned field, PureCell val if (!depth || !own_cell(list, field, value, ©)) return PURE_ALLOCATION; Node *leaf = path[depth - 1].node; PureCell old = cell_at(l, leaf, within, field); - size_t size = list->draft ? cell_bytes(l, field, copy) : 0; - unsigned char *bytes = size ? malloc(size) : NULL; - if ((size && !bytes) || + size_t size = RECORD_HEAD + 1 + cell_bytes(l, field, copy); + unsigned char *record = list->journaled ? journal_room(list, size) : NULL; + if ((list->journaled && !record) || (undo && !log_change(undo, (Entry){.change = CHANGED, .list = list, .at = at, .field = field, .old = old}))) { - free(bytes); free_text(copy.text); return PURE_ALLOCATION; } - if (bytes) encode_cell(l, field, copy, bytes); - size_t offset = 0; - for (unsigned f = 0; f < field; ++f) offset += cell_bytes(l, f, cell_at(l, leaf, within, f)); - PureResult result = mirror(list, at, offset, cell_bytes(l, field, old), bytes, size); - free(bytes); - if (result != PURE_OK) { - if (undo) --undo->count; - free_text(copy.text); - return result; + if (record) { + unsigned char *end = record_head(record, RECORD_SET, at); + *end++ = (unsigned char)field; + encode_cell(l, field, copy, end); + list->journal_size += size; } if (l->text_fields[field]) texts(l, leaf, field)[within] = (PureSpan){copy.text, copy.size}; else numbers(l, leaf, field)[within] = copy.num; diff --git a/core/module-list.h b/core/module-list.h index d29fe95..b9d5a92 100644 --- a/core/module-list.h +++ b/core/module-list.h @@ -86,16 +86,27 @@ size_t pure_list_seek(const PureList *, unsigned field, PureCell key, bool *foun * before] = position after for every item. CONFLICT when two items have the * same key (nothing is made). */ PureResult pure_list_sorted(const PureList *, unsigned field, PureList **out, size_t *moved); -/* A list may keep a stored copy in a draft: header bytes, then each item's - * fields in order (a number as its binary64 bits, little-endian; a text as a - * u32le length and its bytes). Every change splices the draft as well, and a - * change the draft refuses (such as its size limit) does not happen. Undoing - * an event changes only the list: the caller restores the draft's version. */ -void pure_list_mirror(PureList *, PureDraft *, size_t header); -/* Store every item after the header (the draft holds just the header). */ -PureResult pure_list_write(PureList *); -/* Append the items stored in bytes; each field is checked. Before mirroring. */ +/* The stored form of items: each item's fields in order, a number as its + * binary64 bits (little-endian), a text as a u32le length and its bytes. + * pure_list_size is the bytes of all the items so stored, and + * pure_list_encode writes them to out. */ +size_t pure_list_size(const PureList *); +void pure_list_encode(const PureList *, unsigned char *out); +/* Append the items stored in bytes; each field is checked. */ PureResult pure_list_read(PureList *, const unsigned char *bytes, size_t size); +/* A stored list journals its changes (from pure_list_journal(list, true) + * on): every insert, remove and set adds a record, a u8 kind (1 insert, 2 + * remove, 3 set) and the u32le position, then an insert's item or a set's + * u8 field and cell in stored form. A change there is no room to record does + * not happen. pure_list_changes are the records since pure_list_forget last + * kept none; forgetting down to a length taken earlier drops the records of + * changes undone since (an undo does not journal). pure_list_replay makes + * the changes of records again, in order: a list read from its stored items + * and given its journal is the list that wrote them. */ +void pure_list_journal(PureList *, bool journaled); +PureSpan pure_list_changes(const PureList *); +void pure_list_forget(PureList *, size_t size); +PureResult pure_list_replay(PureList *, const unsigned char *bytes, size_t size); /* Undo the changes logged after mark (a count taken earlier), newest first. */ void pure_undo_rollback(PureUndo *, size_t mark); void pure_undo_commit(PureUndo *); diff --git a/core/module-save.c b/core/module-save.c index 4149316..0a4beaf 100644 --- a/core/module-save.c +++ b/core/module-save.c @@ -6,10 +6,19 @@ #include <string.h> /* Stored forms. values: "PVS1", then per state {u8 type, u8 name length, - * name, binary64 bits (little-endian)}. A list: "PLS1", u8 field count, per - * field {u8 0 number | 1 text, u8 name length, name}, then its items - * (module-list.h). A text state: its UTF-8 bytes. */ -static const unsigned char values_magic[4] = {'P', 'V', 'S', '1'}, list_magic[4] = {'P', 'L', 'S', '1'}; + * name, binary64 bits (little-endian)}. A list: "PLS2", u8 field count, per + * field {u8 0 number | 1 text, u8 name length, name}, u64le the bytes of its + * items, its items, then its journal's records to the end (module-list.h): + * a save appends the records of the list's changes, and rewrites the items + * with no records once the records outgrow them. "PLS1" (read, rewritten on + * start) had no length, no records: its items reached the end. A text state: + * its UTF-8 bytes. */ +static const unsigned char values_magic[4] = {'P', 'V', 'S', '1'}, list_magic[4] = {'P', 'L', 'S', '2'}, + list_magic_1[4] = {'P', 'L', 'S', '1'}; +/* Records a list keeps before its items are stored again: at least this, + * else as many bytes as its items. So a save writes what changed, and the + * list's stored bytes stay within twice its items (and this). */ +enum { JOURNAL_MIN = 1 << 16 }; /* links: "PLK1"; 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); u16le sources, each {u8 length, the name, u64le @@ -18,6 +27,17 @@ static const unsigned char values_magic[4] = {'P', 'V', 'S', '1'}, list_magic[4] static const unsigned char links_magic[4] = {'P', 'L', 'K', '1'}; static const PureVersion unsaved = {UINT64_MAX, UINT64_MAX}; +static unsigned char *put_le(unsigned char *at, uint64_t value, unsigned width) { + for (unsigned i = 0; i < width; ++i) *at++ = (unsigned char)(value >> (8 * i)); + return at; +} + +static uint64_t get_le(const unsigned char *at, unsigned width) { + uint64_t value = 0; + for (unsigned i = 0; i < width; ++i) value |= (uint64_t)at[i] << (8 * i); + return value; +} + bool pure_save_enabled(const PureApp *app) { return app->data.state && app->data.save; } static PureResult open_state(PureApp *app, const char *prefix, const char *name, PureDraft **out) { @@ -88,7 +108,7 @@ PURE_COLD PureResult pure_save_open_text(PureApp *app, const PureModule *m, uint return result; } -static size_t list_header(const PureModule *m, uint32_t record, unsigned char *out) { +static size_t list_header(const PureModule *m, uint32_t record, uint64_t items, unsigned char *out) { const PureRecord *r = &m->record[record]; size_t at = 0; memcpy(out, list_magic, 4); @@ -101,27 +121,24 @@ static size_t list_header(const PureModule *m, uint32_t record, unsigned char *o memcpy(out + at, name, strlen(name)); at += strlen(name); } + for (unsigned b = 0; b < 8; ++b) out[at++] = (unsigned char)(items >> (8 * b)); return at; } -PURE_COLD PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, PureList *list, PureDraft *draft) { - unsigned char header[5 + PURE_RECORD_FIELDS * (2 + PURE_MODULE_NAME)]; - size_t length = list_header(m, m->state_ref[state], header); - PureVersion before = pure_draft_version(draft); - pure_draft_hold(draft, true); - PureResult result = pure_draft_splice(draft, 0, pure_draft_size(draft), (PureSpan){header, length}); - pure_list_mirror(list, draft, length); - if (result == PURE_OK) result = pure_list_write(list); - if (result != PURE_OK) { - pure_draft_restore(draft, before); - pure_list_mirror(list, NULL, 0); - } - pure_draft_hold(draft, false); +PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, const PureList *list, PureDraft *draft) { + size_t items = pure_list_size(list), length; + unsigned char header[5 + PURE_RECORD_FIELDS * (2 + PURE_MODULE_NAME) + 8], *bytes = malloc(sizeof(header) + items); + if (!bytes) return PURE_ALLOCATION; + length = list_header(m, m->state_ref[state], items, header); + memcpy(bytes, header, length); + pure_list_encode(list, bytes + length); + PureResult result = pure_draft_splice(draft, 0, pure_draft_size(draft), (PureSpan){bytes, length + items}); + free(bytes); return result; } PURE_COLD PureResult pure_save_open_list(PureApp *app, const PureModule *m, uint32_t state, PureList **out, PureDraft **draft, - PureVersion *saved) { + PureVersion *saved, size_t *journaled) { uint32_t record = m->state_ref[state]; const PureRecord *r = &m->record[record]; *out = NULL; @@ -134,30 +151,36 @@ PURE_COLD PureResult pure_save_open_list(PureApp *app, const PureModule *m, uint char names[PURE_RECORD_FIELDS][PURE_MODULE_NAME] = {{0}}; uint8_t text[PURE_RECORD_FIELDS] = {0}; unsigned count = 0; - size_t at = 0; - bool stored = result == PURE_OK && bytes; + size_t at = 0, items = 0; + bool stored = result == PURE_OK && bytes, first = stored && size >= 4 && !memcmp(bytes, list_magic_1, 4); if (stored) { - bool ok = size >= 5 && !memcmp(bytes, list_magic, 4) && bytes[4] <= PURE_RECORD_FIELDS; + bool ok = size >= 5 && (first || !memcmp(bytes, list_magic, 4)) && bytes[4] <= PURE_RECORD_FIELDS; count = ok ? bytes[4] : 0; at = 5; for (unsigned f = 0; ok && f < count; ++f) { ok = named(bytes, size, &at, 0, &text[f], names[f]) && text[f] <= 1; for (unsigned k = 0; k < f; ++k) ok = ok && strcmp(names[k], names[f]); } + if (ok && !first) { + ok = size - at >= 8 && (items = (size_t)get_le(bytes + at, 8)) <= size - at - 8; + at += 8; + } else items = size - at; if (!ok) result = PURE_STORE_ERROR; } + /* The stored items, then the changes since they were stored. */ PureList *from = NULL; if (result == PURE_OK && stored) { from = pure_list_new(count, text); - result = from ? pure_list_read(from, bytes + at, size - at) : PURE_ALLOCATION; + result = from ? pure_list_read(from, bytes + at, items) : PURE_ALLOCATION; + if (result == PURE_OK) result = pure_list_replay(from, bytes + at + items, size - at - items); } + *journaled = result == PURE_OK && stored ? size - at - items : 0; free(bytes); bool same = stored && count == r->count; for (unsigned f = 0; same && f < count; ++f) same = text[f] == m->record_field_text[r->first + f] && !strcmp(names[f], m->record_field_names[r->first + f]); - if (result == PURE_OK && same) { + if (result == PURE_OK && same && !first) { *out = from; - pure_list_mirror(from, *draft, at); *saved = pure_draft_version(*draft); return PURE_OK; } @@ -165,10 +188,14 @@ PURE_COLD PureResult pure_save_open_list(PureApp *app, const PureModule *m, uint /* New, or stored with another record: its items move by field name. */ const char *field_names[PURE_RECORD_FIELDS]; for (unsigned f = 0; f < count; ++f) field_names[f] = names[f]; - if (from) result = pure_list_migrate(out, from, field_names, text, count, m, record); + if (same) { /* stored in the earlier form: the same items, stored again */ + *out = from; + from = NULL; + } else if (from) result = pure_list_migrate(out, from, field_names, text, count, m, record); else if (!(*out = pure_list_new(r->count, m->record_field_text + r->first))) result = PURE_ALLOCATION; } pure_list_free(from); + *journaled = 0; if (result == PURE_OK) result = pure_save_rewrite_list(m, state, *out, *draft); if (result != PURE_OK) { pure_list_free(*out); @@ -258,17 +285,6 @@ PURE_COLD PureResult pure_save_open_values(PureApp *app, const PureModule *m, do static bool same_version(PureVersion a, PureVersion b) { return a.ref == b.ref && a.bytes == b.bytes; } -static unsigned char *put_le(unsigned char *at, uint64_t value, unsigned width) { - for (unsigned i = 0; i < width; ++i) *at++ = (unsigned char)(value >> (8 * i)); - return at; -} - -static uint64_t get_le(const unsigned char *at, unsigned width) { - uint64_t value = 0; - for (unsigned i = 0; i < width; ++i) value |= (uint64_t)at[i] << (8 * i); - return value; -} - /* The links as stored; NULL when memory runs out. */ static unsigned char *encode_links(const PureApp *app, size_t *size) { const PureModule *m = app->module; @@ -406,6 +422,23 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) { return result; } +/* After a save of lists (the first count of states): kept, each list's + * changes are stored (its journal empties), or, failed, each draft is as it + * was; the holds the save took go either way. */ +static void settle_lists(PureApp *app, PureDraft *const *drafts, const uint32_t *states, size_t count, + const PureVersion *before, const size_t *journaled, bool saved) { + for (size_t k = 0; k < count; ++k) { + uint32_t i = states[k]; + if (app->module->state_type[i] != PURE_STATE_LIST) continue; + if (!saved) pure_draft_restore(drafts[k], before[i]); + else { + pure_list_forget(app->lists[i], 0); + app->journaled[i] = journaled[i]; + } + pure_draft_hold(drafts[k], false); + } +} + PureResult pure_save_commit(PureApp *app, const double *values, void *const *documents, size_t document_count, int64_t *revisions) { if (!pure_save_enabled(app)) { @@ -422,11 +455,38 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc for (uint32_t i = 0; i < m->states; ++i) { PureDraft *draft = m->state_type[i] == PURE_STATE_TEXT ? app->texts[i] : m->state_type[i] == PURE_STATE_LIST ? app->list_drafts[i] : NULL; - if (draft && !same_version(pure_draft_version(draft), app->saved[i])) { + if (draft && (!same_version(pure_draft_version(draft), app->saved[i]) || + (m->state_type[i] == PURE_STATE_LIST && pure_list_changes(app->lists[i]).size))) { states[count] = i; drafts[count++] = draft; } } + /* A list's changes go after its stored records, or, once the records + * would outgrow its items, the items are stored again with none. */ + PureVersion lists_before[PURE_MODULE_STATES]; + size_t journaled[PURE_MODULE_STATES]; + for (size_t k = 0; k < count; ++k) { + uint32_t i = states[k]; + if (m->state_type[i] != PURE_STATE_LIST) continue; + PureSpan changes = pure_list_changes(app->lists[i]); + size_t items = pure_list_size(app->lists[i]); + lists_before[i] = pure_draft_version(drafts[k]); + pure_draft_hold(drafts[k], true); + PureResult result = PURE_OK; + if (!changes.size) journaled[i] = app->journaled[i]; + else if (app->journaled[i] + changes.size > (items > JOURNAL_MIN ? items : JOURNAL_MIN)) { + result = pure_save_rewrite_list(m, i, app->lists[i], drafts[k]); + journaled[i] = 0; + } else { + result = pure_draft_splice(drafts[k], pure_draft_size(drafts[k]), 0, changes); + journaled[i] = app->journaled[i] + changes.size; + } + if (result != PURE_OK) { + settle_lists(app, drafts, states, k + 1, lists_before, journaled, false); + return result; + } + } + size_t list_count = count; /* Numbers, cursors and indices: stored again only when their bytes change. */ PureVersion values_before = {0, 0}; bool values_changed = false; @@ -434,14 +494,21 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc unsigned char encoded[VALUES_BYTES], *stored; size_t length = encode_values(m, values, encoded), size; PureResult result = read_whole(app->stored_values, &stored, &size); - if (result != PURE_OK) return result; + if (result != PURE_OK) { + settle_lists(app, drafts, states, list_count, lists_before, journaled, false); + return result; + } values_changed = size != length || memcmp(stored, encoded, length); free(stored); if (values_changed) { values_before = pure_draft_version(app->stored_values); pure_draft_hold(app->stored_values, true); result = pure_draft_splice(app->stored_values, 0, size, (PureSpan){encoded, length}); - if (result != PURE_OK) { pure_draft_hold(app->stored_values, false); return result; } + if (result != PURE_OK) { + pure_draft_hold(app->stored_values, false); + settle_lists(app, drafts, states, list_count, lists_before, journaled, false); + return result; + } states[count] = UINT32_MAX; drafts[count++] = app->stored_values; } @@ -466,6 +533,7 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc pure_draft_restore(app->stored_values, values_before); pure_draft_hold(app->stored_values, false); } + settle_lists(app, drafts, states, list_count, lists_before, journaled, false); return result; } links_written = true; @@ -476,6 +544,7 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc PureResult result = app->data.save(app->data.context, documents, document_count, drafts, count, revisions); for (size_t k = 0; result == PURE_OK && k < count; ++k) if (states[k] != UINT32_MAX) app->saved[states[k]] = pure_draft_version(drafts[k]); + settle_lists(app, drafts, states, list_count, lists_before, journaled, result == PURE_OK); if (values_changed) { if (result != PURE_OK) pure_draft_restore(app->stored_values, values_before); pure_draft_hold(app->stored_values, false); diff --git a/core/module-save.h b/core/module-save.h index bc7adcc..f83e294 100644 --- a/core/module-save.h +++ b/core/module-save.h @@ -4,23 +4,25 @@ /* App states a data host keeps (docs/design/008-module-format.md), internal * to the module executor. Each text state is a stored draft, each list state - * a stored draft its list mirrors, and numbers, cursors and indices are one - * stored value by name. A save publishes what changed, all or nothing. */ + * a stored draft of its items and the records of its changes since, and + * numbers, cursors and indices are one stored value by name. A save + * publishes what changed, all or nothing. */ bool pure_save_enabled(const PureApp *); /* A text state of m: its stored value, or its initial constant when none. */ PureResult pure_save_open_text(PureApp *, const PureModule *m, uint32_t state, PureDraft **out, PureVersion *saved); -/* A list state of m: its stored items, fields moved by name when the stored - * record differs from m's. */ +/* A list state of m: its stored items with its stored changes made again, + * fields moved by name when the stored record differs from m's; journaled is + * the bytes of the records stored after its items. */ PureResult pure_save_open_list(PureApp *, const PureModule *m, uint32_t state, PureList **out, PureDraft **draft, - PureVersion *saved); + PureVersion *saved, size_t *journaled); /* 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); -/* Store a list whose record changed in its draft, header and items anew. */ -PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, PureList *list, PureDraft *draft); +/* Store a list in its draft anew: header and items, no records. */ +PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, const PureList *list, PureDraft *draft); /* The stored links of a starting app (module.h, PureApp); CONFLICT when an * output with sends left is missing or has other fields. */ PureResult pure_save_open_links(PureApp *, const PureModule *m); diff --git a/core/module.c b/core/module.c index d80b19f..8d889f7 100644 --- a/core/module.c +++ b/core/module.c @@ -2425,8 +2425,9 @@ void pure_app_finish(PureApp *app) { app->arena.used = 0; } -/* The start of a burst: what a failed save returns to. Text and stored list - * drafts stay held until the save, so every version since stays readable. */ +/* The start of a burst: what a failed save returns to. Text drafts stay held + * until the save, so every version since stays readable; a stored list's + * journal is as long as its changes since the last save. */ static void open_burst(PureApp *app) { if (app->open) return; const PureModule *m = app->module; @@ -2438,10 +2439,7 @@ static void open_burst(PureApp *app) { pure_draft_hold(app->texts[i], true); app->durable_texts[i] = pure_draft_version(app->texts[i]); } - if (app->list_drafts[i]) { - pure_draft_hold(app->list_drafts[i], true); - app->durable_lists[i] = pure_draft_version(app->list_drafts[i]); - } + if (app->list_drafts[i]) app->durable_lists[i] = pure_list_changes(app->lists[i]).size; } app->open = true; } @@ -2461,10 +2459,7 @@ static void close_burst(PureApp *app, bool keep) { if (!keep) pure_draft_restore(app->texts[i], app->durable_texts[i]); pure_draft_hold(app->texts[i], false); } - if (app->list_drafts[i]) { - if (!keep) pure_draft_restore(app->list_drafts[i], app->durable_lists[i]); - pure_draft_hold(app->list_drafts[i], false); - } + if (app->list_drafts[i] && !keep) pure_list_forget(app->lists[i], app->durable_lists[i]); } if (keep) pure_undo_commit(&app->undo); else { @@ -2588,7 +2583,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], lists[PURE_MODULE_STATES]; + PureVersion texts[PURE_MODULE_STATES]; + size_t lists[PURE_MODULE_STATES]; /* stored lists' journal lengths */ size_t undo, sent; uint64_t last_sent[PURE_MODULE_INPUTS]; bool opened; @@ -2599,7 +2595,7 @@ static void mark_event(PureApp *app, Mark *mark) { open_burst(app); for (uint32_t i = 0; i < app->module->states; ++i) { if (app->texts[i]) mark->texts[i] = pure_draft_version(app->texts[i]); - if (app->list_drafts[i]) mark->lists[i] = pure_draft_version(app->list_drafts[i]); + if (app->list_drafts[i]) mark->lists[i] = pure_list_changes(app->lists[i]).size; } mark->undo = app->undo.count; mark->sent = app->sent_count; @@ -2609,7 +2605,7 @@ static void mark_event(PureApp *app, Mark *mark) { static void undo_event(PureApp *app, const Mark *mark) { for (uint32_t i = 0; i < app->module->states; ++i) { if (app->texts[i]) pure_draft_restore(app->texts[i], mark->texts[i]); - if (app->list_drafts[i]) pure_draft_restore(app->list_drafts[i], mark->lists[i]); + if (app->list_drafts[i]) pure_list_forget(app->lists[i], mark->lists[i]); } pure_undo_rollback(&app->undo, mark->undo); drop_sent(app, mark->sent); @@ -3238,13 +3234,14 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) PureDraft *texts[PURE_MODULE_STATES] = {0}, *list_drafts[PURE_MODULE_STATES] = {0}; PureList *lists[PURE_MODULE_STATES] = {0}; PureVersion saved[PURE_MODULE_STATES] = {{0, 0}}, before[PURE_MODULE_STATES] = {{0, 0}}; + size_t journaled[PURE_MODULE_STATES] = {0}; int kept[PURE_MODULE_STATES]; bool rewrite[PURE_MODULE_STATES] = {false}; PureResult result = PURE_OK; 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; - if (kept[i] >= 0) saved[i] = app->saved[kept[i]]; + if (kept[i] >= 0) saved[i] = app->saved[kept[i]], journaled[i] = app->journaled[kept[i]]; if (module->state_type[i] == PURE_STATE_TEXT) { if (kept[i] >= 0) texts[i] = app->texts[kept[i]]; else if (result == PURE_OK) @@ -3258,7 +3255,7 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) list_drafts[i] = app->list_drafts[kept[i]]; rewrite[i] = result == PURE_OK && list_drafts[i] && lists[i] != app->lists[kept[i]]; } else if (stored && !(module->state_flags[i] & PURE_LIST_ROWS >> 8)) { /* rows are never stored */ - result = pure_save_open_list(app, module, i, &lists[i], &list_drafts[i], &saved[i]); + result = pure_save_open_list(app, module, i, &lists[i], &list_drafts[i], &saved[i], &journaled[i]); } else if (!(lists[i] = new_list(module, module->state_ref[i]))) result = PURE_ALLOCATION; } else values[i] = kept[i] >= 0 ? app->values[kept[i]] : module->initial[i]; } @@ -3306,7 +3303,10 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) pure_draft_hold(list_drafts[i], true); } for (uint32_t i = 0; i < module->states && result == PURE_OK; ++i) - if (rewrite[i]) result = pure_save_rewrite_list(module, i, lists[i], list_drafts[i]); + if (rewrite[i]) { + result = pure_save_rewrite_list(module, i, lists[i], list_drafts[i]); + journaled[i] = 0; + } for (uint32_t i = 0; i < module->states; ++i) if (rewrite[i]) { if (result != PURE_OK) pure_draft_restore(list_drafts[i], before[i]); @@ -3340,7 +3340,10 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) memcpy(app->texts, texts, sizeof(texts)); memcpy(app->lists, lists, sizeof(lists)); memcpy(app->list_drafts, list_drafts, sizeof(list_drafts)); + memcpy(app->journaled, journaled, sizeof(journaled)); memcpy(app->saved, saved, sizeof(saved)); + for (uint32_t i = 0; i < module->states; ++i) + if (list_drafts[i]) pure_list_journal(lists[i], true); /* a stored list's changes are recorded from now on */ if (old) remap_outputs(app, old, module, true); drop_passes(app); /* kept by instruction of the module they were made for */ app->module = module; diff --git a/core/module.h b/core/module.h index ed6c062..c8a5f3d 100644 --- a/core/module.h +++ b/core/module.h @@ -265,13 +265,15 @@ typedef struct { PureList *lists[PURE_MODULE_STATES]; PureUndo undo; /* list changes of the current event */ PureDraft *list_drafts[PURE_MODULE_STATES]; /* stored lists, when kept */ + size_t journaled[PURE_MODULE_STATES]; /* a stored list's record bytes after its items */ PureDraft *stored_values; /* numbers, cursors and indices, when kept */ PureVersion saved[PURE_MODULE_STATES]; /* text and list versions at the last save */ /* Events since the last save: what a failed save returns to, and the * documents messages opened (held until the save). */ bool deferred, open; double durable_values[PURE_MODULE_STATES]; - PureVersion durable_texts[PURE_MODULE_STATES], durable_lists[PURE_MODULE_STATES]; + PureVersion durable_texts[PURE_MODULE_STATES]; + size_t durable_lists[PURE_MODULE_STATES]; /* each stored list's journal length */ uint64_t durable_ticks, durable_events; struct PurePending { void *document; char key[65]; bool edited; } *pending; /* grows to PURE_APP_PENDING */ unsigned pending_count, pending_capacity; diff --git a/core/test-module.c b/core/test-module.c index 11e7362..777f7dc 100644 --- a/core/test-module.c +++ b/core/test-module.c @@ -994,9 +994,10 @@ static bool collect_bytes(void *context, PureSpan span) { } /* The stored form of the model: a number's binary64 bits, a text's u32le length and bytes. */ -static bool stored_as(PureDraft *draft, const Model *m) { - Grown got = {0}; - CHECK(pure_draft_read(draft, 0, pure_draft_size(draft), collect_bytes, &got) == PURE_OK); +static bool stored_as(const PureList *list, const Model *m) { + Grown got = {pure_list_size(list) ? malloc(pure_list_size(list)) : NULL, pure_list_size(list), 0}; + CHECK(got.bytes || !got.size); + pure_list_encode(list, got.bytes); size_t at = 0; bool same = true; for (size_t i = 0; i < m->count && same; ++i) { @@ -1014,13 +1015,28 @@ static bool stored_as(PureDraft *draft, const Model *m) { return same; } +/* A list read from stored items and given the records stored after them. */ +static PureList *stored_list(const Grown *stored, size_t items) { + static const uint8_t text_fields[2] = {0, 1}; + PureList *list = pure_list_new(2, text_fields); + CHECK(list && pure_list_read(list, stored->bytes, items) == PURE_OK); + if (pure_list_replay(list, stored->bytes + items, stored->size - items) != PURE_OK) { + pure_list_free(list); + return NULL; + } + return list; +} + static void list_trees(void) { static const uint8_t text_fields[2] = {0, 1}; uint32_t seed = 0x5eed1234u; PureList *list = pure_list_new(2, text_fields); - PureDraft *draft = pure_draft_new(); - CHECK(list && draft); - pure_list_mirror(list, draft, 0); + CHECK(list); + pure_list_journal(list, true); + /* What a store would hold: the items as last stored whole, then the + * records of each save since. */ + Grown stored = {0}; + size_t stored_items = 0, saves = 0, rewrites = 0; PureUndo undo = {0}; Model model = {0}, before = {0}; /* Up to four shares at a time, each with what it shared and the items @@ -1032,10 +1048,8 @@ static void list_trees(void) { unsigned first = 0, held = 0, splits_seen = 0; size_t largest = 0, events = 0, undone = 0; for (unsigned event = 0; event < 6000; ++event) { - size_t mark = undo.count; + size_t mark = undo.count, journal = pure_list_changes(list).size; model_copy(&before, &model); - pure_draft_hold(draft, true); - PureVersion version = pure_draft_version(draft); /* Mostly growing to a few thousand, then shrinking back down, so the * tree gains and loses levels. */ unsigned grow = event < 2500 ? 70 : event < 4500 ? 30 : 60; @@ -1067,19 +1081,41 @@ static void list_trees(void) { } } } - if (next_random(&seed) % 5 == 0) { /* undone: the list and its stored copy as before */ + if (next_random(&seed) % 5 == 0) { /* undone: the list and its journal as before */ pure_undo_rollback(&undo, mark); - pure_draft_restore(draft, version); + pure_list_forget(list, journal); model_copy(&model, &before); ++undone; } else pure_undo_commit(&undo); - pure_draft_hold(draft, false); ++events; largest = model.count > largest ? model.count : largest; splits_seen += model.count > 32; if (event % 7 == 0) CHECK(same_items(list, &model)); - if (event % 50 == 0) CHECK(stored_as(draft, &model)); + /* A save every few events: its records after the stored ones, or now + * and then the items stored whole with none after them. */ + if (event % 3 == 0) { + PureSpan changes = pure_list_changes(list); + if (event % 240 == 0) { + stored.size = 0; + Grown whole = {malloc(pure_list_size(list) + 1), 0, 0}; + CHECK(whole.bytes); + pure_list_encode(list, whole.bytes); + collect_bytes(&stored, (PureSpan){whole.bytes, pure_list_size(list)}); + free(whole.bytes); + stored_items = pure_list_size(list); + ++rewrites; + } else + collect_bytes(&stored, changes); + pure_list_forget(list, 0); + ++saves; + } + if (event % 51 == 0) { /* just saved: what a restart finds is the list as it is */ + PureList *restarted = stored_list(&stored, stored_items); + CHECK(restarted && same_items(restarted, &model)); + pure_list_free(restarted); + CHECK(stored_as(list, &model)); + } /* A share now and then, as a snapshot takes one; the oldest goes with * the items and nodes retired while it was the newest. */ if (event % 40 == 20) { @@ -1098,14 +1134,33 @@ static void list_trees(void) { if (event % 25 == 0) for (unsigned i = 0; i < held; ++i) CHECK(same_items(copies[(first + i) % SHARES], &shared[(first + i) % SHARES])); } - CHECK(same_items(list, &model) && stored_as(draft, &model) && largest > 2000 && undone > 500); + CHECK(same_items(list, &model) && stored_as(list, &model) && largest > 2000 && undone > 500 && rewrites > 20); + /* Damaged records are refused: cut short, past the end, of no kind or no field. */ + collect_bytes(&stored, pure_list_changes(list)); + pure_list_forget(list, 0); + PureList *whole = stored_list(&stored, stored_items); + CHECK(whole && same_items(whole, &model)); + pure_list_free(whole); + size_t length = stored.size; + static const unsigned char damaged[][8] = { + {1, 0xff, 0xff, 0xff, 0}, {2, 0xff, 0xff, 0xff, 0}, {9, 0, 0, 0, 0}, {3, 0, 0, 0, 0, 7}, {3, 0, 0, 0, 0, 0, 1}}; + static const size_t damaged_size[] = {5, 5, 5, 6, 7}; + for (unsigned d = 0; d < 5; ++d) { + stored.size = length; + collect_bytes(&stored, (PureSpan){damaged[d], damaged_size[d]}); + CHECK(!stored_list(&stored, stored_items)); + } + CHECK(length > stored_items); + stored.size = length - 1; /* the last record cut short */ + CHECK(!stored_list(&stored, stored_items)); + free(stored.bytes); + printf("{\"journal_saves\":%zu,\"journal_rewrites\":%zu}\n", saves, rewrites); for (; held; --held, first = (first + 1) % SHARES) { CHECK(same_items(copies[first], &shared[first])); pure_list_free(copies[first]); pure_retired_free(&retired[first]); } pure_list_free(list); - pure_draft_close(draft); pure_undo_free(&undo); /* A keyed list: items kept in key order, found by key. */ |