diff options
| -rw-r--r-- | core/module.c | 53 | ||||
| -rw-r--r-- | core/module.h | 25 | ||||
| -rw-r--r-- | core/test-module.c | 78 | ||||
| -rw-r--r-- | serve/serve.c | 39 | ||||
| -rw-r--r-- | tests/check.py | 68 |
5 files changed, 239 insertions, 24 deletions
diff --git a/core/module.c b/core/module.c index 2688395..a5bc3c5 100644 --- a/core/module.c +++ b/core/module.c @@ -165,7 +165,7 @@ static void write_i64(unsigned char *p, int64_t value) { const char *pure_fault_name(PureFault fault) { static const char *const names[PURE_FAULT_COUNT] = {"none", "divide", "non_finite", "clamp", "limit", "not_found", - "encoding", "exists", "memory", "overflow"}; + "encoding", "exists", "memory", "overflow", "work"}; return fault < PURE_FAULT_COUNT ? names[fault] : "none"; } @@ -1437,10 +1437,17 @@ typedef struct { 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 : + return kind == PURE_FAULT_LIMIT || kind == PURE_FAULT_WORK ? PURE_LIMIT : kind == PURE_FAULT_NOT_FOUND ? PURE_NOT_FOUND : kind == PURE_FAULT_EXISTS ? PURE_CONFLICT : kind == PURE_FAULT_MEMORY ? PURE_BUDGET : PURE_INVALID; } +/* Count what the unit running (an event, or a view's evaluation) is about + * to run, as written: false past the app's work budget. */ +static bool charge(PureApp *app, uint64_t instructions) { + app->work += instructions; + return !app->work_budget || app->work <= app->work_budget; +} + static bool reserve(PureArena *arena, size_t more) { if (more <= arena->capacity - arena->used) return true; size_t capacity = arena->capacity ? arena->capacity : 256; @@ -2598,6 +2605,7 @@ static PureResult run(const PureModule *m, PureValue *values, PureBlock block, E Slot high = stack[--sp], low = stack[--sp]; slice_of(list, m->state_key[arg >> 8 & 0xFF] - 1u, low, high, &from, &to); } + if (!charge(app, (uint64_t)inner.length * (to - from))) { kind = PURE_FAULT_WORK; goto failed; } /* as written */ if (!op->shape || !fold_shaped(m, values, op, list, &per, &fold)) { if (op->range) /* only the items a range of keys allows can count */ range_bounds(m, values, &m->range[op->range - 1], list, m->state_key[arg >> 8 & 0xFF] - 1u, &per, &from, &to); @@ -2628,7 +2636,13 @@ static PureResult run(const PureModule *m, PureValue *values, PureBlock block, E if (arg & PURE_EACH_SLICED) { /* the items with keys from low up to before high */ Slot high = stack[--sp], low = stack[--sp]; slice_of(app->lists[walk.list], m->state_key[walk.list] - 1u, low, high, &walk.at, &walk.end); - } else if (op->range) /* a where that bounds the key: only the items in bounds */ + } + size_t items = pure_list_count(app->lists[walk.list]); /* as written: what it starts with, the list only shrinking */ + if (!charge(app, (uint64_t)inner.length * ((walk.end < items ? walk.end : items) - walk.at))) { + kind = PURE_FAULT_WORK; + goto failed; + } + if (!(arg & PURE_EACH_SLICED) && op->range) /* a where that bounds the key: only the items in bounds */ range_bounds(m, values, &m->range[op->range - 1], app->lists[walk.list], m->state_key[walk.list] - 1u, &per, &walk.at, &walk.end); while (walk.at < walk.end && walk.at < pure_list_count(app->lists[walk.list])) { @@ -3083,6 +3097,13 @@ static void clear_rows(PureApp *app) { void pure_app_memory(PureApp *app, size_t bytes) { app->memory = bytes; } +void pure_app_work(PureApp *app, uint64_t instructions) { + app->work_budget = instructions; + for (unsigned i = 0; i < app->row_tables; ++i) app->rows[i].changes = UINT64_MAX; /* a filter refused before may pass */ +} + +PureFault pure_app_view_fault(const PureApp *app) { return app->view_fault; } + size_t pure_app_state_bytes(const PureApp *app) { size_t bytes = (app->outbox ? pure_list_size(app->outbox) : 0) + (app->heard ? pure_list_size(app->heard) : 0); for (uint32_t left = app->counted; left; left &= left - 1) { @@ -3108,9 +3129,12 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) { memcpy(pending, app->values, sizeof(app->values)); app->fault = PURE_FAULT_NONE; app->fault_pc = UINT32_MAX; + app->work = 0; PureResult result = PURE_OK; for (uint32_t i = 0; i < m->handlers && result == PURE_OK; ++i) - if (m->handler[i].event == event) result = run(m, pending, m->handler[i].code, ev, NULL); + if (m->handler[i].event == event) + result = charge(app, m->handler[i].code.length) ? run(m, pending, m->handler[i].code, ev, NULL) + : fault(app, PURE_FAULT_WORK, m->handler[i].code.start); /* An edit needs the revision it was made against. A save rebases an * edited document and releasing texts may reclaim earlier text versions: * keep such a reply by copying it first. */ @@ -3420,7 +3444,11 @@ static PureResult evaluate(PureApp *app, uint32_t view, size_t row, Slot *out) { if (!argue(app, view, &ev)) return PURE_INVALID; PureFault fault = app->fault; uint32_t fault_pc = app->fault_pc; - PureResult result = run(app->module, values, app->module->view_code[view], &ev, out); + PureBlock code = app->module->view_code[view]; + app->work = 0; + app->fault = PURE_FAULT_NONE; + PureResult result = charge(app, code.length) ? run(app->module, values, code, &ev, out) : PURE_LIMIT; + app->view_fault = result == PURE_OK ? PURE_FAULT_NONE : app->fault != PURE_FAULT_NONE ? app->fault : PURE_FAULT_WORK; app->fault = fault; app->fault_pc = fault_pc; return result; @@ -3461,9 +3489,18 @@ static struct PureRows *filtered(PureApp *app, uint32_t table) { } if (rows->changes == app->changes && (!m->view_arguments[table] || rows->given == app->arguments_given)) return rows; rows->count = 0; + /* A unit of work of its own, counted as written: the filter for every item. */ + const PureList *list = app->lists[m->view_list[table]]; + app->work = 0; + rows->fault = charge(app, (uint64_t)m->view_code[table].length * pure_list_count(list)) ? PURE_FAULT_NONE : PURE_FAULT_WORK; + uint64_t work = app->work; + if (rows->fault != PURE_FAULT_NONE) { + rows->changes = app->changes; + rows->given = app->arguments_given; + return rows; + } /* A leaf of the list at a time when the filter has lane forms; a leaf * where an item would fault goes item by item, leaving that item out. */ - const PureList *list = app->lists[m->view_list[table]]; PureRun run; if (m->view_plan[table]) { double store[PURE_MODULE_STACK][PURE_LIST_RUN], scratch[PURE_LIST_RUN]; @@ -3488,6 +3525,7 @@ static struct PureRows *filtered(PureApp *app, uint32_t table) { } else for (size_t item = 0, items = pure_list_count(list); item < items; ++item) if (!keep_evaluated(app, table, rows, item)) return NULL; + app->work = work; /* the filter's, whichever items ran alone */ rows->changes = app->changes; rows->given = app->arguments_given; return rows; @@ -3501,6 +3539,7 @@ bool pure_app_row_item(PureApp *app, uint32_t table, size_t row, size_t *item) { return row < pure_list_count(app->lists[m->view_list[table]]); } struct PureRows *rows = filtered(app, table); + app->view_fault = rows ? rows->fault : PURE_FAULT_NONE; if (!rows || row >= rows->count) return false; *item = rows->items[row]; return true; @@ -3528,6 +3567,7 @@ bool pure_app_item_row(PureApp *app, uint32_t table, size_t item, size_t *row) { /* A view outside a table (column false), or a column and a row of its table. */ static PureResult view_of(PureApp *app, uint32_t view, bool column, size_t row, uint8_t type, Slot *slot) { const PureModule *m = app->module; + app->view_fault = PURE_FAULT_NONE; if (view >= m->views) return PURE_INVALID; uint8_t kind = m->view_type[view] == PURE_VIEW_BOOL ? PURE_VIEW_NUM : m->view_type[view]; /* a bool's value is 0 or 1 */ if (kind != type || (m->view_table[view] != 0) != column) return PURE_INVALID; @@ -3559,6 +3599,7 @@ size_t pure_app_rows(PureApp *app, uint32_t table) { const PureModule *m = app->module; if (table >= m->views || m->view_type[table] != PURE_VIEW_TABLE) return 0; struct PureRows *rows = filtered(app, table); + app->view_fault = rows ? rows->fault : PURE_FAULT_NONE; return rows ? rows->count : m->view_code[table].length ? 0 : pure_list_count(app->lists[m->view_list[table]]); } diff --git a/core/module.h b/core/module.h index e106fc7..c4919c9 100644 --- a/core/module.h +++ b/core/module.h @@ -153,14 +153,15 @@ 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). MEMORY: the event would leave the states holding * more than the app's budget (pure_app_memory; it reports PURE_BUDGET). - * OVERFLOW: an i64 past its range. */ + * OVERFLOW: an i64 past its range. WORK: the event (or a view's evaluation) + * would run more than the app's work budget (pure_app_work; PURE_LIMIT). */ 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_MEMORY, PURE_FAULT_OVERFLOW, PURE_FAULT_COUNT + PURE_FAULT_MEMORY, PURE_FAULT_OVERFLOW, PURE_FAULT_WORK, PURE_FAULT_COUNT } PureFault; /* A fault's name as hosts report it: none, divide, non_finite, clamp, limit, - * not_found, encoding, exists, memory, overflow. */ + * not_found, encoding, exists, memory, overflow, work. */ const char *pure_fault_name(PureFault); /* plan, shape and range are the runtime's, set at load: a fold whose block @@ -377,6 +378,8 @@ typedef struct { PureFault fault; uint32_t fault_pc; size_t memory; /* the budget of what the states hold (0: none): pure_app_memory */ + uint64_t work_budget, work; /* pure_app_work's budget (0: none), and what the unit running has used */ + PureFault view_fault; /* why the last view evaluated had no value (pure_app_view_fault) */ char why[128]; /* why the last start or replace was refused, when it says (else empty; a failed start keeps it) */ uint32_t counted; /* the states it counts, a bit each: lists (not a message's rows) and texts */ uint64_t ticks, events; @@ -384,7 +387,7 @@ typedef struct { * sets it, so a script can replay any time; views never see it. */ double now; uint64_t changes; /* counts every change of states, never back: what a filter's rows are fresh for */ - struct PureRows { uint32_t table; uint64_t changes, given; size_t *items, count, capacity; } *rows; /* a filtered table's items */ + struct PureRows { uint32_t table; uint64_t changes, given; size_t *items, count, capacity; PureFault fault; } *rows; /* a filtered table's items */ unsigned row_tables; /* Current message or text event, kept until pure_app_finish. */ void *document; @@ -544,6 +547,20 @@ 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 *); +/* An app's work budget (docs/design/017-apps-on-one-host.md): what one + * event, or one evaluation of a view (a value, a cell, a table's filter), + * may run, in instructions counted as written, whatever the plan: a block's + * length each time it runs, and a fold's or a walk's block length for each + * item of its list (of its slice, when it takes one) it goes over, all + * counted as the fold or walk begins. Past it the event fails with the fault + * PURE_FAULT_WORK (PURE_LIMIT) and changes nothing, and a view has no value. + * 0: no budget, at first. */ +void pure_app_work(PureApp *, uint64_t instructions); +/* Why the last view evaluated on this app (or a table's rows asked for) + * had no value: PURE_FAULT_WORK past the work budget, another fault, or + * PURE_FAULT_NONE when it had one. */ +PureFault pure_app_view_fault(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 4fe5182..f627dea 100644 --- a/core/test-module.c +++ b/core/test-module.c @@ -2692,13 +2692,14 @@ static void lane_folds(void) { double x = 0, y = 0; bool ok_a = pure_app_view(&a, v, &x), ok_b = pure_app_view(&b, v, &y); faults += !ok_a; + CHECK(a.work == b.work); /* counted as written, however it ran */ if (!same_number(ok_a, x, ok_b, y)) { fprintf(stderr, "lane round %u view %u: %d %g / %d %g\n", round, v, ok_a, x, ok_b, y); abort(); } } size_t rows = pure_app_rows(&a, VIEWS); - CHECK(rows == pure_app_rows(&b, VIEWS)); + CHECK(rows == pure_app_rows(&b, VIEWS) && a.work == b.work); for (size_t row = 0; row < rows; ++row) { size_t item_a, item_b; CHECK(pure_app_row_item(&a, VIEWS, row, &item_a) && pure_app_row_item(&b, VIEWS, row, &item_b) && item_a == item_b); @@ -2706,6 +2707,7 @@ static void lane_folds(void) { double x = 0, y = 0; bool ok_a = pure_app_cell(&a, VIEWS + 1 + c, row, &x), ok_b = pure_app_cell(&b, VIEWS + 1 + c, row, &y); ++cells; + CHECK(a.work == b.work); if (!same_number(ok_a, x, ok_b, y)) { fprintf(stderr, "lane round %u column %u row %zu: %d %g / %d %g\n", round, c, row, ok_a, x, ok_b, y); abort(); @@ -2714,7 +2716,8 @@ static void lane_folds(void) { } passes += pure_app_passes(&a); PureResult went_a = pure_app_input(&a, 0, PURE_EDGE_DOWN), went_b = pure_app_input(&b, 0, PURE_EDGE_DOWN); - if (went_a != went_b || a.fault != b.fault || a.fault_pc != b.fault_pc || memcmp(a.values, b.values, 4 * sizeof(double))) { + if (went_a != went_b || a.fault != b.fault || a.fault_pc != b.fault_pc || a.work != b.work || + memcmp(a.values, b.values, 4 * sizeof(double))) { fprintf(stderr, "lane round %u handler: %d %d %u / %d %d %u\n", round, went_a, a.fault, a.fault_pc, went_b, b.fault, b.fault_pc); abort(); @@ -4334,6 +4337,76 @@ static void links(void) { * 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. */ +/* ---- Work budgets (docs/design/017) ------------------------------------ + * An event, or one evaluation of a view, runs at most the app's work + * budget, counted as written as it begins: a block's length, and a fold's or + * a walk's block length for every item of its list. Past it an event fails + * with work and changes nothing; a view, a cell or a table's rows have no + * value, and pure_app_view_fault says why. (lane_folds checks that a planned + * fold counts what the same fold run item by item counts.) */ +static void work_budget(void) { + CHECK(!strcmp(pure_fault_name(PURE_FAULT_WORK), "work")); + /* The shop: restock runs 3 instructions; clean 23, and its walk's 22 for each item stocked. */ + CHECK(load(walk_spec(walk_code, LEN(walk_code), walk_inputs, 3, walk_handlers, 3, walk_lists, NULL, 0), &module_a) == PURE_OK); + PureApp app; + CHECK(pure_app_start(&app, &module_a, (PureData){0}) == PURE_OK); + unsigned char payload[8]; + for (uint32_t item = 0; item < 5; ++item) { + put32(put32(payload, item), 2); + CHECK(pure_app_message(&app, 2, NULL, -1, (PureSpan){payload, 8}) == PURE_OK && app.work == 3); + } + pure_app_work(&app, 23 + 22 * 5 - 1); + CHECK(pure_app_input(&app, 1, PURE_EDGE_DOWN) == PURE_LIMIT && app.fault == PURE_FAULT_WORK); + for (size_t i = 0; i < 5; ++i) CHECK(pure_list_get(app.lists[W_STOCK], i, 1).num == 2); /* nothing changed */ + pure_app_work(&app, 23 + 22 * 5); + CHECK(pure_app_input(&app, 1, PURE_EDGE_DOWN) == PURE_OK && app.work == 23 + 22 * 5); + for (size_t i = 0; i < 5; ++i) CHECK(pure_list_get(app.lists[W_STOCK], i, 1).num == 3); + pure_app_work(&app, 2); + put32(put32(payload, 9), 1); + CHECK(pure_app_message(&app, 2, NULL, -1, (PureSpan){payload, 8}) == PURE_LIMIT && app.fault == PURE_FAULT_WORK); + CHECK(pure_list_count(app.lists[W_STOCK]) == 5); + pure_app_work(&app, 0); /* none */ + CHECK(pure_app_message(&app, 2, NULL, -1, (PureSpan){payload, 8}) == PURE_OK && pure_list_count(app.lists[W_STOCK]) == 6); + pure_app_stop(&app); + pure_module_release(&module_a); + + /* Views: open is 4 instructions and its fold's 3 for each task. */ + CHECK(load(fold_spec(fold_code, LEN(fold_code), fold_views, LEN(fold_views), fold_handlers, 2), &module_a) == PURE_OK); + CHECK(pure_app_start(&app, &module_a, (PureData){0}) == PURE_OK); + for (const char *word = "abc"; *word; ++word) + CHECK(pure_app_text(&app, 0, (PureSpan){(const unsigned char *)word, 1}) == PURE_OK && app.work == 7); + double value; + CHECK(number_view(&app, 0) == 3 && app.work == 4 + 3 * 3 && pure_app_view_fault(&app) == PURE_FAULT_NONE); + pure_app_work(&app, 4 + 3 * 3 - 1); + CHECK(!pure_app_view(&app, 0, &value) && pure_app_view_fault(&app) == PURE_FAULT_WORK); + CHECK(app.fault == PURE_FAULT_NONE); /* a view leaves the app's own fault alone */ + pure_app_work(&app, 4 + 3 * 3); + CHECK(number_view(&app, 0) == 3 && pure_app_view_fault(&app) == PURE_FAULT_NONE); + pure_app_work(&app, 0); + CHECK(pure_app_input(&app, 1, PURE_EDGE_DOWN) == PURE_OK && pure_app_point(&app, 1, 0) == PURE_OK && + pure_app_input(&app, 1, PURE_EDGE_DOWN) == PURE_OK && pure_app_point(&app, 1, 1) == PURE_OK && + pure_app_input(&app, 1, PURE_EDGE_DOWN) == PURE_OK); /* all done */ + CHECK(!pure_app_view(&app, 4, &value) && pure_app_view_fault(&app) == PURE_FAULT_NOT_FOUND); /* fewest open: none */ + pure_app_stop(&app); + pure_module_release(&module_a); + + /* A table's filter is a unit of its own: 3 instructions for each task; a cell's column is 2. */ + CHECK(load(filter_spec(filter_code, LEN(filter_code), filter_views, 3), &module_a) == PURE_OK); + CHECK(pure_app_start(&app, &module_a, (PureData){0}) == PURE_OK); + for (const char *word = "abcd"; *word; ++word) + CHECK(pure_app_text(&app, 0, (PureSpan){(const unsigned char *)word, 1}) == PURE_OK); + pure_app_work(&app, 3 * 4 - 1); + size_t at; + CHECK(pure_app_rows(&app, 0) == 0 && pure_app_view_fault(&app) == PURE_FAULT_WORK && !pure_app_row_item(&app, 0, 0, &at)); + CHECK(!pure_app_cell(&app, 2, 0, &value) && pure_app_view_fault(&app) == PURE_FAULT_WORK); + pure_app_work(&app, 3 * 4); /* the rows refused before are asked again */ + CHECK(pure_app_rows(&app, 0) == 4 && app.work == 3 * 4 && pure_app_view_fault(&app) == PURE_FAULT_NONE); + CHECK(pure_app_cell(&app, 2, 3, &value) && value == 3 && app.work == 2); + pure_app_stop(&app); + pure_module_release(&module_a); + puts("{\"work_budget\":\"events-views-cells-rows-rolled-back\"}"); +} + static void memory_budget(void) { enum { ADD, DROP, SEND_KEY, OUT, TAKE, ANSWER }; static const S nums[] = {{"n", 0}}; @@ -5166,6 +5239,7 @@ int main(void) { bools(); integers(); memory_budget(); + work_budget(); fold_mutations(); pointing(); list_mutations(); diff --git a/serve/serve.c b/serve/serve.c index 6d9d5d6..b283874 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -83,6 +83,8 @@ static PureSealKey *seal_key; static bool sealed_only; static struct Sender { char name[PURE_SOURCE + 1]; unsigned char key[PURE_SEAL_KEY]; } senders[PURE_MODULE_INPUTS]; static unsigned sender_count; +static size_t memory_budget; /* --memory: the app's budget of what its states hold (0: none) */ +static uint64_t work_budget; /* --work: what an event or a view's evaluation may run (0: none) */ /* A delivery's X-Pure-Seal: the suite, the sender's session (its * encapsulated key), the seal's number in it and, when the sender sealed @@ -1333,6 +1335,7 @@ static PureResult show_given(PureApp *app, const Route *route, size_t available, } else { *type = "application/json"; size_t rows = pure_app_rows(app, view), from = route->from < rows ? route->from : rows; + if (pure_app_view_fault(app) == PURE_FAULT_WORK) status = PURE_LIMIT, rows = from = 0; /* its filter's */ size_t end = from + (route->count < rows - from ? route->count : rows - from); /* Each column's key, ,"name": (the first row's first without the comma), made once. */ struct { uint32_t view; size_t length; char key[PURE_MODULE_NAME + 4]; } columns[PURE_MODULE_VIEWS]; @@ -1355,26 +1358,33 @@ static PureResult show_given(PureApp *app, const Route *route, size_t available, int64_t integer; if (!pure_app_cell_i64(app, columns[k].view, row, &integer)) put_bytes(&o, "null", 4); else put_integer(&o, integer); - continue; - } - if (kind == PURE_VIEW_NUM || kind == PURE_VIEW_BOOL) { + } else if (kind == PURE_VIEW_NUM || kind == PURE_VIEW_BOOL) { if (!pure_app_cell(app, columns[k].view, row, &value)) put_bytes(&o, "null", 4); else if (kind == PURE_VIEW_BOOL) put_bytes(&o, value ? "true" : "false", value ? 4 : 5); else put_number(&o, value); - continue; + } else { + size_t mark = o.size; + put_bytes(&o, "\"", 1); + if (pure_app_cell_text(app, columns[k].view, row, put_json_span, &o) == PURE_OK) put_bytes(&o, "\"", 1); + else if (!o.failed) { + o.size = mark; + put_bytes(&o, "null", 4); + } } - size_t mark = o.size; - put_bytes(&o, "\"", 1); - if (pure_app_cell_text(app, columns[k].view, row, put_json_span, &o) == PURE_OK) put_bytes(&o, "\"", 1); - else if (!o.failed) { - o.size = mark; - put_bytes(&o, "null", 4); + if (pure_app_view_fault(app) == PURE_FAULT_WORK) { /* a cell past the work budget fails the table */ + status = PURE_LIMIT; + end = row; + break; } } put_bytes(&o, "}", 1); } put_bytes(&o, "]}", 2); } + if (status != PURE_OK && pure_app_view_fault(app) == PURE_FAULT_WORK) { /* past the work budget: 422 work */ + answer->fault = PURE_FAULT_WORK; + status = PURE_LIMIT; + } if (status == PURE_OK && !o.bytes && !(o.bytes = malloc(1))) o.failed = true; /* an empty value has a buffer too */ if (status == PURE_OK && o.failed) status = PURE_ALLOCATION; if (status == PURE_OK && o.size > available) status = PURE_BUDGET; @@ -1925,6 +1935,7 @@ static PureResult run_query(const Job *job, PureSnapshot *snapshot, Applied *a) } PureApp reader = {0}; pure_app_read(&reader, snapshot); + pure_app_work(&reader, work_budget); reader.module = &held->module; /* the app's states, the query's views */ Route route = job->route; route.target = (uint32_t)pure_module_view(&held->module, job->effect.key); @@ -2262,7 +2273,6 @@ static void accept_all(Host *host) { } } -static size_t memory_budget; /* --memory: the app's budget of what its states hold (0: none) */ static bool start_executor(Executor *ex, const PureModule *module, uint32_t show, uint32_t edit, const char *store) { ex->module = module; @@ -2275,6 +2285,7 @@ static bool start_executor(Executor *ex, const PureModule *module, uint32_t show } pure_app_background_checkpoints(&ex->app, true); pure_app_memory(&ex->app, memory_budget); + pure_app_work(&ex->app, work_budget); return true; } @@ -2294,6 +2305,7 @@ static bool start_shard(Host *host, const PureModule *module, bool documents, ui host->module = module; host->documents = documents; host->wake = -1; + pure_app_work(&host->reader, work_budget); /* the views it shows from snapshots */ if (!host->owner && !start_executor(&host->local, module, show, edit, store)) return false; host->work = calloc(host->capacity, sizeof(*host->work)); host->table = calloc(host->capacity, sizeof(*host->table)); @@ -2683,6 +2695,7 @@ int main(int argc, char **argv) { else if (!strcmp(argv[i], "--link-idle-ms") && number(argv[i + 1], 86400000, &value) && value) link_idle_ms = value; else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value; else if (!strcmp(argv[i], "--memory") && number(argv[i + 1], SIZE_MAX, &value) && value) memory_budget = (size_t)value; + else if (!strcmp(argv[i], "--work") && number(argv[i + 1], UINT64_MAX, &value) && value) work_budget = value; else if (!strcmp(argv[i], "--window") && number(argv[i + 1], SIZE_MAX, &value) && value) window_bytes = (size_t)value; else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value; else if (!strcmp(argv[i], "--threads") && !strcmp(argv[i + 1], "auto")) threads = 0; @@ -2874,7 +2887,7 @@ usage: fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] " "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=URL ...] [--name NAME] " "[--send-ca FILE] [--reply OUTPUT=INPUT ...] [--seal OUTPUT=KEY ...] [--key FILE] [--sealed-only] " - "[--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] [--link-idle-ms 30000] [--memory BYTES] " + "[--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] [--link-idle-ms 30000] [--memory BYTES] [--work N] " "[--window 4194304] [--surface LIBRARY [-- ARGUMENTS]]\n" " --send OUTPUT=pure://host:port/INPUT: to a Pure node's --link-port, the way between nodes (UDP, no\n" " connection, docs/design/012-link-protocol.md; its key by --seal OUTPUT=KEY)\n" @@ -2884,6 +2897,8 @@ usage: " .pmod unless given; the first line out says it)\n" " --memory BYTES: what the app's states may hold (lists' items, texts, sends not yet stored): an event\n" " past it is refused, 503 memory (docs/design/013-memory-budget.md)\n" + " --work N: what an event, or one evaluation of a view, may run, in instructions counted as written: past\n" + " it the event is refused and the view has no value, 422 work (docs/design/017-apps-on-one-host.md)\n" " --window BYTES: what each --send holds of its sends beyond what its receiver has (and 65,536 sends);\n" " the outbox keeps the rest\n", stderr); return 2; diff --git a/tests/check.py b/tests/check.py index b982256..47ddf23 100644 --- a/tests/check.py +++ b/tests/check.py @@ -3659,6 +3659,73 @@ def server_meaning(aup, pure, work): return {'values': len(expected), 'refused': refused, 'exits': len(exits), 'checkpoints': checkpoints} +WORK_SOURCE = """record Item { + n = 0 +} +state items: list of Item +input add = message { n: number } +input touch = message { step: number } +on add { + items.append(Item { n: add.n }) +} +on touch { + for i in items { + i.n = i.n + touch.step + } +} +view total = sum(i.n for i in items) +view big = each i in items where i.n > 0 { + n = i.n +} +""" + + +def server_work(aup, pure, work): + """A work budget (docs/design/017) over HTTP: as the list grows, the + walking event, the folding view and the filtered table each come to run + past the budget, from then on answer 422 work, and a refused event + changes nothing; the cheap event that adds an item still runs.""" + source = work / 'work.aup' + source.write_text(WORK_SOURCE) + compile_module(aup, source, work / 'work.pmod') + host = HttpHost([pure / 'build/pure-serve', '--module', work / 'work.pmod', '--store', work / 'work-store', '--work', 400]) + first = {} + try: + def total(): + status, _, _, body = http(host.port, 'GET', '/views/total') + return status, body + + for n in range(1, 200): + status, why, _, _ = http(host.port, 'POST', '/inputs/add?n=1') + assert status == 200, (n, status, why) # adding stays cheap + before = total() + status, why, _, _ = http(host.port, 'POST', '/inputs/touch?step=1') + assert status in (200, 422) and (status == 200 or why == 'work'), (n, status, why) + if status == 422: + first.setdefault('touch', n) + assert total() == before, n # refused: nothing changed + else: + assert 'touch' not in first, n # once past the budget, always past it + status, why, _, body = http(host.port, 'GET', '/views/total') + assert status == 200 or (status, why) == (422, 'work'), (n, status, why) + if status == 422: + first.setdefault('total', n) + else: + assert 'total' not in first, n + status, why, _, body = http(host.port, 'GET', '/views/big') + assert status == 200 or (status, why) == (422, 'work'), (n, status, why) + if status == 422: + first.setdefault('big', n) + else: + assert 'big' not in first and json.loads(body)['rows'] == n, (n, body[:80]) + if len(first) == 3: + break + assert len(first) == 3 and first['touch'] < first['big'], first # a walk's block is longer than a filter's + finally: + host.stop() + return first + + def main(): global EDITOR parser = argparse.ArgumentParser() @@ -3707,6 +3774,7 @@ def main(): report['server_answers'] = server_answers(aup, pure, work) report['server_answers_memory'] = server_answers_memory(aup, pure, work) report['server_meaning'] = server_meaning(aup, pure, work) + report['server_work'] = server_work(aup, pure, work) print(json.dumps(report)) |