diff options
Diffstat (limited to 'core')
| -rw-r--r-- | core/module.c | 86 | ||||
| -rw-r--r-- | core/module.h | 29 | ||||
| -rw-r--r-- | core/test-module.c | 99 |
3 files changed, 190 insertions, 24 deletions
diff --git a/core/module.c b/core/module.c index 79d3c5b..cbf0efe 100644 --- a/core/module.c +++ b/core/module.c @@ -1179,7 +1179,7 @@ static PureResult fault(PureApp *app, PureFault kind, uint32_t pc) { app->fault = kind; app->fault_pc = pc; return kind == PURE_FAULT_LIMIT ? PURE_LIMIT : kind == PURE_FAULT_NOT_FOUND ? PURE_NOT_FOUND : - kind == PURE_FAULT_EXISTS ? PURE_CONFLICT : PURE_INVALID; + kind == PURE_FAULT_EXISTS ? PURE_CONFLICT : kind == PURE_FAULT_MEMORY ? PURE_BUDGET : PURE_INVALID; } static bool reserve(PureArena *arena, size_t more) { @@ -2645,7 +2645,7 @@ void pure_app_stop(PureApp *app) { typedef struct { PureVersion texts[PURE_MODULE_STATES]; size_t lists[PURE_MODULE_STATES], heard, outbox; /* stored lists' journal lengths */ - size_t undo; + size_t undo, sent_saved; uint64_t last_sent[PURE_MODULE_INPUTS]; bool opened; } Mark; @@ -2660,6 +2660,7 @@ static void mark_event(PureApp *app, Mark *mark) { 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_saved = app->sent_saved; memcpy(mark->last_sent, app->last_sent, sizeof(mark->last_sent)); } @@ -2671,6 +2672,7 @@ static void undo_event(PureApp *app, const Mark *mark) { 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); + app->sent_saved = mark->sent_saved; memcpy(app->last_sent, mark->last_sent, sizeof(app->last_sent)); if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */ } @@ -2696,6 +2698,24 @@ static PureResult hear(PureApp *app, const char *source, uint64_t seq) { return pure_list_insert(app->heard, at, item, &app->undo); } +/* The output's saved sends through seq leave the outbox: their receiver + * stored them, or answered them (in the answer's event). */ +static PureResult let_sends_go(PureApp *app, uint32_t output, uint64_t through) { + if (!app->outbox) return PURE_OK; + const char *name = app->module->input_names[output]; + size_t length = strlen(name); + 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; + PureResult removed = pure_list_remove(app->outbox, i, &app->undo); + if (removed != PURE_OK) return removed; + --app->sent_saved; + app->links_changed = true; + } + return PURE_OK; +} + static PureResult keep_reply(PureApp *app); /* Every list of a message's rows emptied: an event's rows are its own. */ @@ -2705,14 +2725,29 @@ static void clear_rows(PureApp *app) { if (m->state_type[i] == PURE_STATE_LIST && m->state_flags[i] & PURE_LIST_ROWS >> 8) pure_list_clear(app->lists[i]); } +void pure_app_memory(PureApp *app, size_t bytes) { app->memory = bytes; } + +size_t pure_app_state_bytes(const PureApp *app) { + const PureModule *m = app->module; + size_t bytes = (app->outbox ? pure_list_size(app->outbox) : 0) + (app->heard ? pure_list_size(app->heard) : 0); + for (uint32_t i = 0; i < m->states; ++i) { + if (app->lists[i] && !(m->state_flags[i] & PURE_LIST_ROWS >> 8)) bytes += pure_list_size(app->lists[i]); + else if (app->texts[i]) bytes += pure_draft_size(app->texts[i]); + } + return bytes; /* a message's rows are its own, gone when it ends */ +} + /* One event: every handler for it runs in module order against pending * states, all or nothing. A key, text or message event saves before it * returns unless saving is deferred; a frame leaves that to the next event - * (frames come 60 times a second). A message's document joins the burst. */ + * (frames come 60 times a second). A message's document joins the burst. + * With a memory budget, an event that leaves the states over it, and + * holding more than before, fails. */ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) { const PureModule *m = app->module; double pending[PURE_MODULE_STATES], locals[PURE_MODULE_LOCALS] = {0}; ev->locals = locals; + size_t held = app->memory ? pure_app_state_bytes(app) : 0; Mark mark; mark_event(app, &mark); memcpy(pending, app->values, sizeof(app->values)); @@ -2728,8 +2763,14 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) { if (result == PURE_OK && app->has_reply && ((app->reply.source == FROM_DOCUMENT && ev->modified) || app->reply.source == FROM_TEXT)) result = keep_reply(app); - /* A delivery's number is heard with its changes. */ + /* A delivery's number is heard with its changes, and an answer's sends go with it. */ if (result == PURE_OK && app->delivering.pending) result = hear(app, app->delivering.source, app->delivering.seq); + if (result == PURE_OK && app->delivering.pending && app->delivering.release) + result = let_sends_go(app, app->delivering.release - 1, app->delivering.seq); + if (result == PURE_OK && app->memory) { + size_t holds = pure_app_state_bytes(app); + if (holds > app->memory && holds > held) result = fault(app, PURE_FAULT_MEMORY, UINT32_MAX); + } if (result != PURE_OK) { undo_event(app, &mark); clear_rows(app); @@ -2950,23 +2991,14 @@ void pure_app_outbox_newest(const PureApp *app, PureSentVisit visit, void *conte * 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) { - const PureModule *m = app->module; - if (!app->outbox || output >= m->inputs) return; - const char *name = m->input_names[output]; - size_t length = strlen(name); + if (!app->outbox || output >= app->module->inputs) return; 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; - } + let_sends_go(app, output, through); } -PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload, - const uint32_t *placed) { +/* A delivery, or an answer to the sends of output (release: its number + 1). */ +static PureResult deliver(PureApp *app, uint32_t input, uint32_t release, const char *source, uint64_t seq, + PureSpan payload, const uint32_t *placed) { const PureModule *m = app->module; size_t length = source ? strlen(source) : 0; app->duplicate = false; @@ -2983,16 +3015,19 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui } app->delivering.source = source; app->delivering.seq = seq; + app->delivering.release = release; app->delivering.pending = true; PureResult result = message(app, input, NULL, -1, payload, placed); if (!app->delivering.pending) return result; /* Not applied. Refused by the app (a fault), the message was delivered - * all the same: its number is heard, saved like a key event's changes. */ + * all the same: its number is heard, saved like a key event's changes. + * Refused for memory it was not: its sender sends it again. */ app->delivering.pending = false; - if (app->fault == PURE_FAULT_NONE) return result; + if (app->fault == PURE_FAULT_NONE || app->fault == PURE_FAULT_MEMORY) return result; Mark mark; mark_event(app, &mark); PureResult heard = hear(app, source, seq); + if (heard == PURE_OK && release) heard = let_sends_go(app, release - 1, seq); if (heard != PURE_OK) { undo_event(app, &mark); return heard; @@ -3001,6 +3036,17 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui return saved == PURE_OK ? result : saved; } +PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload, + const uint32_t *placed) { + return deliver(app, input, 0, source, seq, payload, placed); +} + +PureResult pure_app_answer(PureApp *app, uint32_t input, uint32_t output, const char *source, uint64_t seq, + PureSpan payload, const uint32_t *placed) { + if (output >= app->module->inputs || app->module->input[output].kind != PURE_KIND_OUTPUT) return PURE_INVALID; + return deliver(app, input, output + 1, source, seq, payload, placed); +} + PureResult pure_app_reply(const PureApp *app, bool (*visit)(void *, PureSpan), void *context) { if (!app->has_reply) return PURE_OK; return visit_value(app, &app->arena, app->reply, visit, context); diff --git a/core/module.h b/core/module.h index 0b45ba2..c11c682 100644 --- a/core/module.h +++ b/core/module.h @@ -109,10 +109,12 @@ typedef enum { PURE_VIEW_NUM = 0, PURE_VIEW_TEXT = 1, PURE_VIEW_TABLE = 2 } Pure typedef enum { PURE_UNIT_CHAR = 0, PURE_UNIT_LINE = 1 } PureUnit; typedef enum { PURE_EDGE_DOWN, PURE_EDGE_UP } PureEdge; /* EXISTS: an item with that key is already in the keyed list (the event - * reports PURE_CONFLICT). */ + * reports PURE_CONFLICT). MEMORY: the event would leave the states holding + * more than the app's budget (pure_app_memory; it reports PURE_BUDGET). */ typedef enum { PURE_FAULT_NONE, PURE_FAULT_DIVIDE, PURE_FAULT_NON_FINITE, PURE_FAULT_CLAMP, - PURE_FAULT_LIMIT, PURE_FAULT_NOT_FOUND, PURE_FAULT_ENCODING, PURE_FAULT_EXISTS + PURE_FAULT_LIMIT, PURE_FAULT_NOT_FOUND, PURE_FAULT_ENCODING, PURE_FAULT_EXISTS, + PURE_FAULT_MEMORY } PureFault; /* plan, shape and range are the runtime's, set at load: a fold whose block @@ -310,6 +312,7 @@ typedef struct { bool document_pending; /* the current event's document is among them */ PureFault fault; uint32_t fault_pc; + size_t memory; /* the budget of what the states hold (0: none): pure_app_memory */ uint64_t ticks, events; /* The time events see as now, in milliseconds since 1970 (UTC). The host * sets it, so a script can replay any time; views never see it. */ @@ -346,7 +349,7 @@ typedef struct { PureDraft *stored_links; bool links_changed; /* since the last save */ bool duplicate; /* the last delivery was heard before and not applied */ - struct { const char *source; uint64_t seq; bool pending; } delivering; + struct { const char *source; uint64_t seq; uint32_t release; bool pending; } delivering; /* The arguments given for a view (pure_app_arguments): its input + 1 (0: * none), where their fields are, and a count of the times they were given. */ unsigned argued; @@ -443,13 +446,31 @@ void pure_app_trim(PureApp *, uint32_t output, uint64_t through); * before is not applied again: it succeeds with app->duplicate set. The * number counts as heard with the event's changes, and also when the app * refuses the message (a fault: it was delivered, and refused); not when a - * save fails. placed is as for pure_app_message_fields, or NULL. */ + * save fails, nor when the app has no memory for it (pure_app_memory: it + * comes again). placed is as for pure_app_message_fields, or NULL. */ PureResult pure_app_deliver(PureApp *, uint32_t input, const char *source, uint64_t seq, PureSpan payload, const uint32_t *placed); +/* An answer to the send numbered seq of output (serve/deliver.h's --reply), + * delivered as pure_app_deliver delivers from source; in the same event the + * output's saved sends through seq leave the outbox, so the answer has the + * room its send held. */ +PureResult pure_app_answer(PureApp *, uint32_t input, uint32_t output, const char *source, uint64_t seq, + PureSpan payload, const uint32_t *placed); /* The last number heard from source (0 when none): a host that gets * numbered messages out of order hands them over from the one after it. */ uint64_t pure_app_heard(const PureApp *, const char *source); +/* An app's memory budget (docs/design/013-memory-budget.md): what its + * states hold, as they are stored (their lists' items, their texts, the + * outbox and the numbers heard), at most bytes (0: no budget, at first). An + * event after which they hold more than that, and more than before it, + * fails with the fault PURE_FAULT_MEMORY (PURE_BUDGET) and changes nothing; + * one that frees or keeps them goes on. A delivery refused so is not heard: + * its sender sends it again. */ +void pure_app_memory(PureApp *, size_t bytes); +/* What the states hold now, as the budget counts it. */ +size_t pure_app_state_bytes(const PureApp *); + /* Readers on other threads. Right after a save an app can make a snapshot of * its saved states, for evaluating views anywhere while it keeps applying * events: numbers are copied; text states share their stored versions and diff --git a/core/test-module.c b/core/test-module.c index 1446e73..70cc02d 100644 --- a/core/test-module.c +++ b/core/test-module.c @@ -4209,6 +4209,104 @@ static void links(void) { CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "handler event")); } +/* The app's memory budget (docs/design/013-memory-budget.md): an event that + * would leave the states over it and holding more is refused and changes + * nothing; one that frees them goes on; sends waiting for their receivers + * count, so a full outbox refuses more until some are trimmed; a delivery + * refused so is not heard, and comes again; an answer has its send's room. */ +static void memory_budget(void) { + enum { ADD, DROP, SEND_KEY, OUT, TAKE, ANSWER }; + static const S nums[] = {{"n", 0}}; + static const T states[] = {{"rows", 3, 0}}; + static const RF fields[] = {{"value", 0, 0}}; + static const R records[] = {{"row", fields, 1}}; + static const F value[] = {{"value", PURE_FIELD_NUMBER}}; + static const I inputs[] = {KEY("add", "KeyA"), KEY("drop", "KeyD"), KEY("send", "KeyS"), + {"out", NULL, PURE_KIND_OUTPUT, false, value, 1}, + {"take", NULL, PURE_KIND_MESSAGE, false, value, 1}, + {"answer", NULL, PURE_KIND_MESSAGE, false, value, 1}}; + static const O code[] = { + {L, 0, 0}, {LA, 1, 0}, /* add: rows.append(row { n }) */ + {XF, 1, 0}, {LR, 1, 0}, /* drop: rows.remove(rows.first) */ + {C, 0, 1}, {SEND, 3, 0}, /* send: send out { value: 1 } */ + {FIELD, 0, 0}, {LA, 1, 0}, /* take: rows.append(row { take.value }) */ + {FIELD, 0, 0}, {LA, 1, 0}, {FIELD, 0, 0}, {LA, 1, 0}, {FIELD, 0, 0}, {LA, 1, 0}, /* answer: six rows */ + {FIELD, 0, 0}, {LA, 1, 0}, {FIELD, 0, 0}, {LA, 1, 0}, {FIELD, 0, 0}, {LA, 1, 0}, + }; + static const H handlers[] = {{1 + 2 * ADD, 0, 2}, {1 + 2 * DROP, 2, 2}, {1 + 2 * SEND_KEY, 4, 2}, {1 + 2 * TAKE, 6, 2}, + {1 + 2 * ANSWER, 8, 12}}; + Spec spec = {nums, 1, inputs, 6, NULL, 0, handlers, 5, code, LEN(code), NULL, 0, states, 1, NULL, 0, NULL, 0, + records, 1, NULL, 0}; + CHECK(load(spec, &module_a) == PURE_OK); + char *path = new_store(); + PureApp app; + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK); + for (int k = 0; k < 3; ++k) CHECK(pure_app_input(&app, ADD, PURE_EDGE_DOWN) == PURE_OK); + size_t three = pure_app_state_bytes(&app); + CHECK(three == 3 * 8); /* a row's number as it is stored */ + /* Room for one row more: the next is refused, and nothing of it stays. */ + pure_app_memory(&app, three + 8); + CHECK(pure_app_input(&app, ADD, PURE_EDGE_DOWN) == PURE_OK && pure_list_count(app.lists[1]) == 4); + CHECK(pure_app_input(&app, ADD, PURE_EDGE_DOWN) == PURE_BUDGET && app.fault == PURE_FAULT_MEMORY && + pure_list_count(app.lists[1]) == 4 && pure_app_state_bytes(&app) == three + 8); + /* Over the budget (made smaller), what frees goes on and what grows does not. */ + pure_app_memory(&app, 8); + CHECK(pure_app_input(&app, ADD, PURE_EDGE_DOWN) == PURE_BUDGET); + CHECK(pure_app_input(&app, DROP, PURE_EDGE_DOWN) == PURE_OK && pure_list_count(app.lists[1]) == 3); + /* Sends count while they wait: a full outbox refuses the next until some are trimmed. */ + size_t before = pure_app_state_bytes(&app); + pure_app_memory(&app, before); + CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_BUDGET && !pure_list_count(app.outbox)); + pure_app_memory(&app, 0); + CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_OK && pure_list_count(app.outbox) == 1); + size_t one = pure_app_state_bytes(&app) - before; + pure_app_memory(&app, before + 2 * one); + CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_OK); + CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_BUDGET && pure_list_count(app.outbox) == 2); + pure_app_trim(&app, OUT, 1); + CHECK(pure_app_save(&app) == PURE_OK && pure_list_count(app.outbox) == 1); + CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_OK && outbox_of(&app).items[1].seq == 3); + /* A delivery refused for memory is not heard: given again with room, it is applied once. */ + double taken = 7; + pure_app_memory(&app, pure_app_state_bytes(&app)); + CHECK(pure_app_deliver(&app, TAKE, "far.out", 1, (PureSpan){(const unsigned char *)&taken, 8}, NULL) == PURE_BUDGET && + app.fault == PURE_FAULT_MEMORY && pure_app_heard(&app, "far.out") == 0 && pure_list_count(app.lists[1]) == 3); + pure_app_finish(&app); + pure_app_memory(&app, 0); + CHECK(pure_app_deliver(&app, TAKE, "far.out", 1, (PureSpan){(const unsigned char *)&taken, 8}, NULL) == PURE_OK && + !app.duplicate && pure_app_heard(&app, "far.out") == 1 && pure_list_count(app.lists[1]) == 4); + pure_app_finish(&app); + CHECK(pure_list_get(app.lists[1], 3, 0).num == 7); + /* An answer takes its send out of the outbox in its own event: one that + * keeps more than the send held is refused whole at the budget (the send + * stays), and goes on with room; one that keeps less goes on at it. */ + size_t full = pure_app_state_bytes(&app); + CHECK(one < 6 * 8 && one > 8); + double paid = 9; + PureSpan nine = {(const unsigned char *)&paid, 8}; + pure_app_memory(&app, full); + CHECK(pure_app_answer(&app, ANSWER, OUT, "wire.out", 2, nine, NULL) == PURE_BUDGET && + app.fault == PURE_FAULT_MEMORY && pure_list_count(app.outbox) == 2 && pure_list_count(app.lists[1]) == 4 && + pure_app_heard(&app, "wire.out") == 0 && pure_app_state_bytes(&app) == full); + pure_app_finish(&app); + size_t heard = 4 + strlen("wire.out") + 8; /* its source among the numbers heard */ + pure_app_memory(&app, full + 6 * 8 + heard - one); + CHECK(pure_app_answer(&app, ANSWER, OUT, "wire.out", 2, nine, NULL) == PURE_OK && pure_list_count(app.outbox) == 1 && + outbox_of(&app).items[0].seq == 3 && pure_list_count(app.lists[1]) == 10 && + pure_app_heard(&app, "wire.out") == 2 && pure_app_state_bytes(&app) == full + 6 * 8 + heard - one); + pure_app_finish(&app); + pure_app_memory(&app, pure_app_state_bytes(&app)); + CHECK(pure_app_answer(&app, TAKE, OUT, "wire.out", 3, nine, NULL) == PURE_OK && !pure_list_count(app.outbox) && + pure_list_count(app.lists[1]) == 11); + pure_app_finish(&app); + CHECK(pure_app_answer(&app, TAKE, TAKE, "wire.out", 4, nine, NULL) == PURE_INVALID); /* not an output */ + pure_app_stop(&app); + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_list_count(app.lists[1]) == 11 && + !pure_list_count(app.outbox) && pure_app_heard(&app, "far.out") == 1 && pure_app_heard(&app, "wire.out") == 3); + pure_app_stop(&app); + remove_store(path); +} + /* An output whose fields are named otherwise (here two numbers swapped) is * another output: its waiting sends' bytes would mean something else. A * module that has it so is refused while sends wait, replacing or starting @@ -4360,6 +4458,7 @@ int main(void) { snapshots(); links(); output_meaning(); + memory_budget(); fold_mutations(); pointing(); list_mutations(); |