diff options
Diffstat (limited to 'core/module-save.c')
| -rw-r--r-- | core/module-save.c | 488 |
1 files changed, 488 insertions, 0 deletions
diff --git a/core/module-save.c b/core/module-save.c new file mode 100644 index 0000000..4149316 --- /dev/null +++ b/core/module-save.c @@ -0,0 +1,488 @@ +#include "module-save.h" +#include "module-text.h" +#include <math.h> +#include <stdio.h> +#include <stdlib.h> +#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'}; +/* 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 + * last number heard}; then the sent messages to the end, each {u8 output (in + * the table), u64le number, u32le size, payload}. */ +static const unsigned char links_magic[4] = {'P', 'L', 'K', '1'}; +static const PureVersion unsaved = {UINT64_MAX, UINT64_MAX}; + +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) { + char key[64]; + snprintf(key, sizeof(key), "%s%s", prefix, name); + return app->data.state(app->data.context, key, out); +} + +/* The whole value of a draft, contiguous; free it. */ +typedef struct { unsigned char *bytes; size_t used; } Whole; + +static bool gather(void *context, PureSpan span) { + Whole *w = context; + memcpy(w->bytes + w->used, span.data, span.size); + w->used += span.size; + return true; +} + +static PureResult read_whole(const PureDraft *draft, unsigned char **out, size_t *size) { + *size = pure_draft_size(draft); + Whole w = {malloc(*size ? *size : 1), 0}; + if (!w.bytes) return PURE_ALLOCATION; + PureResult result = pure_draft_read(draft, 0, *size, gather, &w); + if (result != PURE_OK) { free(w.bytes); return result; } + *out = w.bytes; + return PURE_OK; +} + +/* One named entry of a stored form: u8 kind, u8 name length, the name (an + * identifier), then width more bytes. */ +PURE_COLD static bool named(const unsigned char *bytes, size_t size, size_t *at, size_t width, uint8_t *kind, + char name[PURE_MODULE_NAME]) { + if (size - *at < 2) return false; + size_t length = bytes[*at + 1]; + if (size - *at - 2 < length + width || !length || length >= PURE_MODULE_NAME) return false; + for (size_t i = 0; i < length; ++i) { + unsigned char c = bytes[*at + 2 + i]; + bool letter = (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || c == '_'; + if (!letter && !(i && c >= '0' && c <= '9')) return false; + } + *kind = bytes[*at]; + memcpy(name, bytes + *at + 2, length); + name[length] = 0; + *at += 2 + length; + return true; +} + +PURE_COLD PureResult pure_save_open_text(PureApp *app, const PureModule *m, uint32_t state, PureDraft **out, + PureVersion *saved) { + PureResult result = open_state(app, "text-", m->state_names[state], out); + if (result != PURE_OK) return result; + *saved = pure_draft_version(*out); + if (!pure_draft_revision(*out)) { + uint32_t constant = m->state_ref[state]; + PureSpan initial = {m->constant_data + m->constant_offset[constant], m->constant_length[constant]}; + result = pure_draft_splice(*out, 0, pure_draft_size(*out), initial); + *saved = unsaved; + } else { + unsigned char *bytes; + size_t size; + result = read_whole(*out, &bytes, &size); + if (result == PURE_OK) { + if (!pure_utf8_valid((PureSpan){bytes, size})) result = PURE_STORE_ERROR; + free(bytes); + } + } + if (result != PURE_OK) { pure_draft_close(*out); *out = NULL; } + return result; +} + +static size_t list_header(const PureModule *m, uint32_t record, unsigned char *out) { + const PureRecord *r = &m->record[record]; + size_t at = 0; + memcpy(out, list_magic, 4); + at = 4; + out[at++] = (unsigned char)r->count; + for (unsigned f = 0; f < r->count; ++f) { + const char *name = m->record_field_names[r->first + f]; + out[at++] = m->record_field_text[r->first + f]; + out[at++] = (unsigned char)strlen(name); + memcpy(out + at, name, strlen(name)); + at += strlen(name); + } + 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); + return result; +} + +PURE_COLD PureResult pure_save_open_list(PureApp *app, const PureModule *m, uint32_t state, PureList **out, PureDraft **draft, + PureVersion *saved) { + uint32_t record = m->state_ref[state]; + const PureRecord *r = &m->record[record]; + *out = NULL; + PureResult result = open_state(app, "list-", m->state_names[state], draft); + if (result != PURE_OK) return result; + unsigned char *bytes = NULL; + size_t size = 0; + if (pure_draft_revision(*draft)) result = read_whole(*draft, &bytes, &size); + /* The stored record: its fields' kinds and names. */ + 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; + if (stored) { + bool ok = size >= 5 && !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) result = PURE_STORE_ERROR; + } + 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; + } + 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) { + *out = from; + pure_list_mirror(from, *draft, at); + *saved = pure_draft_version(*draft); + return PURE_OK; + } + if (result == PURE_OK) { + /* 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); + else if (!(*out = pure_list_new(r->count, m->record_field_text + r->first))) result = PURE_ALLOCATION; + } + pure_list_free(from); + if (result == PURE_OK) result = pure_save_rewrite_list(m, state, *out, *draft); + if (result != PURE_OK) { + pure_list_free(*out); + pure_draft_close(*draft); + *out = NULL; + *draft = NULL; + return result; + } + *saved = unsaved; + return PURE_OK; +} + +static bool scalar(uint8_t type) { + return type == PURE_STATE_NUM || type == PURE_STATE_CURSOR || type == PURE_STATE_INDEX; +} + +/* A stored value still fits its state: a cursor on a character boundary of + * its text, an index on an item of its list (or 0), a finite number. */ +PURE_COLD static bool fits(const PureModule *m, PureDraft *const *texts, PureList *const *lists, uint32_t state, double value) { + if (!isfinite(value)) return false; + if (m->state_type[state] == PURE_STATE_NUM) return true; + if (value < 0 || value != floor(value)) return false; + if (m->state_type[state] == PURE_STATE_INDEX) { + size_t count = pure_list_count(lists[m->state_ref[state]]); + return value <= (count ? (double)(count - 1) : 0); + } + const PureDraft *text = texts[m->state_ref[state]]; + size_t size = pure_draft_size(text), at = (size_t)value; + if (value > (double)size) return false; + if (at == size) return true; + unsigned char byte = 0; + Whole w = {&byte, 0}; + return pure_draft_read(text, at, 1, gather, &w) == PURE_OK && (byte & 0xC0) != 0x80; +} + +static size_t encode_values(const PureModule *m, const double *values, unsigned char *out) { + size_t at = 4; + memcpy(out, values_magic, 4); + for (uint32_t i = 0; i < m->states; ++i) { + if (!scalar(m->state_type[i])) continue; + size_t length = strlen(m->state_names[i]); + uint64_t bits; + memcpy(&bits, &values[i], sizeof(bits)); + out[at++] = m->state_type[i]; + out[at++] = (unsigned char)length; + memcpy(out + at, m->state_names[i], length); + at += length; + for (unsigned b = 0; b < 8; ++b) out[at++] = (unsigned char)(bits >> (8 * b)); + } + return at; +} + +enum { VALUES_BYTES = 4 + PURE_MODULE_STATES * (2 + PURE_MODULE_NAME + 8) }; + +PURE_COLD PureResult pure_save_open_values(PureApp *app, const PureModule *m, double *values, const int *kept, + PureDraft *const *texts, PureList *const *lists) { + bool scalars = false, fresh = false; + for (uint32_t i = 0; i < m->states; ++i) + if (scalar(m->state_type[i])) { + scalars = true; + fresh = fresh || kept[i] < 0; + } + if (!scalars) return PURE_OK; + PureResult result = app->stored_values ? PURE_OK : open_state(app, "", "values", &app->stored_values); + if (result != PURE_OK || !fresh || !pure_draft_revision(app->stored_values)) return result; + unsigned char *bytes; + size_t size; + if ((result = read_whole(app->stored_values, &bytes, &size)) != PURE_OK) return result; + bool ok = size >= 4 && !memcmp(bytes, values_magic, 4); + for (size_t at = 4; ok && at < size;) { + char name[PURE_MODULE_NAME]; + uint8_t type; + if (!(ok = named(bytes, size, &at, 8, &type, name))) break; + uint64_t bits = 0; + for (unsigned b = 0; b < 8; ++b) bits |= (uint64_t)bytes[at + b] << (8 * b); + at += 8; + double value; + memcpy(&value, &bits, sizeof(value)); + int state = pure_module_state(m, name); + if (state >= 0 && kept[state] < 0 && m->state_type[state] == type && scalar(type) && + fits(m, texts, lists, (uint32_t)state, value)) + values[state] = value; + } + free(bytes); + return ok ? PURE_OK : PURE_STORE_ERROR; +} + +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; + unsigned outputs = 0; + uint8_t table[PURE_MODULE_INPUTS]; + size_t total = 4 + 1 + 2; + for (uint32_t i = 0; i < m->inputs; ++i) + if (m->input[i].kind == PURE_KIND_OUTPUT) { + table[i] = (uint8_t)outputs++; + 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 (unsigned h = 0; h < app->heard_count; ++h) total += 1 + strlen(app->heard[h].source) + 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); + at += 4; + *at++ = (unsigned char)(outputs + app->carried_count); + for (uint32_t i = 0; i < m->inputs; ++i) { + if (m->input[i].kind != PURE_KIND_OUTPUT) continue; + size_t length = strlen(m->input_names[i]); + *at++ = (unsigned char)length; + memcpy(at, m->input_names[i], length); + at += length; + *at++ = (unsigned char)m->input[i].field_count; + for (uint32_t f = 0; f < m->input[i].field_count; ++f) *at++ = m->field[m->input[i].field_start + f].type; + at = put_le(at, app->last_sent[i], 8); + } + for (unsigned c = 0; c < app->carried_count; ++c) { + size_t length = strlen(app->carried[c].name); + *at++ = (unsigned char)length; + memcpy(at, app->carried[c].name, length); + at += length; + *at++ = 0; + at = put_le(at, app->carried[c].last, 8); + } + at = put_le(at, app->heard_count, 2); + for (unsigned h = 0; h < app->heard_count; ++h) { + size_t length = strlen(app->heard[h].source); + *at++ = (unsigned char)length; + memcpy(at, app->heard[h].source, length); + at = put_le(at + length, app->heard[h].seq, 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; + } + *size = (size_t)(at - bytes); + return bytes; +} + +/* 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 and sent messages. */ +PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) { + 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; + size_t size; + if ((result = read_whole(app->stored_links, &bytes, &size)) != PURE_OK) return result; + int output[256]; + bool fits[256]; + size_t at = 5; + unsigned outputs = size >= 5 && !memcmp(bytes, links_magic, 4) ? bytes[4] : 0; + result = size >= 5 && !memcmp(bytes, links_magic, 4) ? PURE_OK : PURE_STORE_ERROR; + for (unsigned t = 0; t < outputs && result == PURE_OK; ++t) { + char name[PURE_MODULE_NAME]; + 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); + name[length] = 0; + unsigned fields = bytes[at + 1 + length]; + at += 2 + length; + if (size - at < fields + 8) { result = PURE_STORE_ERROR; break; } + int i = pure_module_input(m, name); + output[t] = i >= 0 && m->input[i].kind == PURE_KIND_OUTPUT ? i : -1; + fits[t] = output[t] >= 0 && m->input[i].field_count == fields; + for (unsigned f = 0; fits[t] && f < fields; ++f) fits[t] = m->field[m->input[i].field_start + f].type == bytes[at + f]; + uint64_t last = get_le(bytes + at + fields, 8); + at += fields + 8; + if (output[t] >= 0) app->last_sent[output[t]] = last; + else if (last) { + struct PureCarried *grown = realloc(app->carried, (app->carried_count + 1) * sizeof(*grown)); + if (!grown) { result = PURE_ALLOCATION; break; } + app->carried = grown; + memcpy(grown[app->carried_count].name, name, length + 1); + grown[app->carried_count++].last = last; + } + } + unsigned sources = result == PURE_OK && size - at >= 2 ? (unsigned)get_le(bytes + at, 2) : 0; + if (result == PURE_OK && size - at < 2) result = PURE_STORE_ERROR; + at += 2; + if (result == PURE_OK && sources && !(app->heard = calloc(sources, sizeof(*app->heard)))) result = PURE_ALLOCATION; + app->heard_capacity = sources; + for (unsigned h = 0; h < sources && result == PURE_OK; ++h) { + size_t length = at < size ? bytes[at] : 0; + if (!length || length > PURE_SOURCE || size - at < 1 + length + 8) { result = PURE_STORE_ERROR; break; } + struct PureHeard *heard = &app->heard[app->heard_count++]; + memcpy(heard->source, bytes + at + 1, length); + heard->seq = heard->saved = get_le(bytes + at + 1 + length, 8); + at += 1 + length + 8; + } + while (result == PURE_OK && at < size) { + 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}; + at += length; + } + free(bytes); + app->sent_durable = app->sent_count; + memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent)); + return result; +} + +PureResult pure_save_commit(PureApp *app, const double *values, void *const *documents, size_t document_count, + int64_t *revisions) { + if (!pure_save_enabled(app)) { + /* Without a host that saves together, a save has one document at most. */ + PureResult result = PURE_OK; + for (size_t i = 0; i < document_count && result == PURE_OK; ++i) + result = app->data.commit(app->data.context, documents[i], &revisions[i]); + return result; + } + const PureModule *m = app->module; + PureDraft *drafts[2 * PURE_MODULE_STATES + 2]; + uint32_t states[2 * PURE_MODULE_STATES + 2]; + size_t count = 0; + 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])) { + states[count] = i; + drafts[count++] = draft; + } + } + /* Numbers, cursors and indices: stored again only when their bytes change. */ + PureVersion values_before = {0, 0}; + bool values_changed = false; + if (app->stored_values) { + 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; + 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; } + states[count] = UINT32_MAX; + drafts[count++] = app->stored_values; + } + } + /* The links, written whole when they changed. */ + PureVersion links_before = {0, 0}; + bool links_written = false; + if (app->links_changed) { + PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links); + size_t length; + unsigned char *encoded = result == PURE_OK ? encode_links(app, &length) : NULL; + if (result == PURE_OK && !encoded) result = PURE_ALLOCATION; + if (result == PURE_OK) { + links_before = pure_draft_version(app->stored_links); + pure_draft_hold(app->stored_links, true); + result = pure_draft_splice(app->stored_links, 0, pure_draft_size(app->stored_links), (PureSpan){encoded, length}); + if (result != PURE_OK) pure_draft_hold(app->stored_links, false); + } + free(encoded); + if (result != PURE_OK) { + if (values_changed) { + pure_draft_restore(app->stored_values, values_before); + pure_draft_hold(app->stored_values, false); + } + return result; + } + links_written = true; + states[count] = UINT32_MAX; + drafts[count++] = app->stored_links; + } + if (!count && !document_count) return PURE_OK; + 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]); + if (values_changed) { + if (result != PURE_OK) pure_draft_restore(app->stored_values, values_before); + pure_draft_hold(app->stored_values, false); + } + if (links_written) { + if (result != PURE_OK) pure_draft_restore(app->stored_links, links_before); + pure_draft_hold(app->stored_links, false); + } + return result; +} |