diff options
| -rw-r--r-- | core/module.c | 86 | ||||
| -rw-r--r-- | core/module.h | 29 | ||||
| -rw-r--r-- | core/test-module.c | 99 | ||||
| -rw-r--r-- | serve/deliver.c | 25 | ||||
| -rw-r--r-- | serve/deliver.h | 12 | ||||
| -rw-r--r-- | serve/serve.c | 72 | ||||
| -rw-r--r-- | tests/check.py | 248 |
7 files changed, 512 insertions, 59 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(); diff --git a/serve/deliver.c b/serve/deliver.c index b4d9e57..68c5f14 100644 --- a/serve/deliver.c +++ b/serve/deliver.c @@ -47,7 +47,6 @@ struct PureWire { int answer_input; /* the input answers become events of, or -1 */ Answer *answers; /* made, for the owner to take in order */ size_t answer_head, answer_count, answer_capacity; - Answer taken; /* the one the owner took last, kept until it takes the next */ uint32_t placed[2 * PURE_MODULE_FIELDS]; /* Sealed sends: to peer's key, authenticated by the node's own key when * it has one (its public half is from), in one session made at the first @@ -240,23 +239,26 @@ bool pure_wire_reply(PureWire *w, uint32_t input, char *error, size_t size) { } bool pure_wire_answer(PureWire *w, PureWireAnswer *out) { - free(w->taken.bytes); - w->taken = (Answer){0}; pthread_mutex_lock(&w->lock); bool some = w->answer_count > 0; - if (some) { - w->taken = w->answers[w->answer_head++]; - --w->answer_count; - w->delivered = w->taken.seq; /* trimmed after the owner applies it */ - } + Answer next = some ? w->answers[w->answer_head] : (Answer){0}; /* its bytes stay put while the array moves */ pthread_mutex_unlock(&w->lock); if (!some) return false; const PureInputShape *in = &w->module->input[w->answer_input]; - for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(w->taken.bytes + 4 * i); - *out = (PureWireAnswer){(uint32_t)w->answer_input, w->taken.seq, - {w->taken.bytes + w->taken.table, w->taken.size - w->taken.table}, w->placed, w->source}; + for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(next.bytes + 4 * i); + *out = (PureWireAnswer){(uint32_t)w->answer_input, next.seq, {next.bytes + next.table, next.size - next.table}, + w->placed, w->source}; return true; } + +void pure_wire_answered(PureWire *w) { + pthread_mutex_lock(&w->lock); + Answer done = w->answers[w->answer_head++]; + --w->answer_count; + w->delivered = done.seq; /* trimmed after the owner applies it */ + pthread_mutex_unlock(&w->lock); + free(done.bytes); +} uint64_t pure_wire_queued(const PureWire *w) { return w->queued; } uint64_t pure_wire_delivered(PureWire *w) { @@ -771,7 +773,6 @@ void pure_wire_close(PureWire *w) { } for (size_t i = 0; i < w->answer_count; ++i) free(w->answers[w->answer_head + i].bytes); free(w->answers); - free(w->taken.bytes); free(w->queue); pthread_cond_destroy(&w->changed); pthread_mutex_destroy(&w->lock); diff --git a/serve/deliver.h b/serve/deliver.h index 71cc38e..a7ccac3 100644 --- a/serve/deliver.h +++ b/serve/deliver.h @@ -49,11 +49,10 @@ void pure_wire_close(PureWire *); * http_status is the answer's status. An answer that makes no message is * logged and the send tried again every 10 s. Answers go one per request. */ bool pure_wire_reply(PureWire *, uint32_t input, char *error, size_t size); -/* The owner's side: the next answer in order, or false. Taking it counts - * its send delivered (the owner applies the answer, then trims); what out - * points to is the wire's until the next call. The answer is applied with - * pure_app_deliver from the wire's source with the send's number, so an - * answer that comes again after a restart is applied once. */ +/* The owner's side: the next answer in order, or false; it stays the next, + * and what out points to the wire's, until pure_wire_answered. The answer + * is applied with pure_app_answer from the wire's source with the send's + * number, so an answer that comes again after a restart is applied once. */ typedef struct { uint32_t input; uint64_t seq; @@ -62,6 +61,9 @@ typedef struct { const char *source; } PureWireAnswer; bool pure_wire_answer(PureWire *, PureWireAnswer *out); +/* The next answer was applied, or refused by the app: it goes, and its send + * counts delivered (the owner trims it). */ +void pure_wire_answered(PureWire *); /* Counts for a host's report. */ void pure_wire_counts(PureWire *, uint64_t *delivered, uint64_t *attempts); #endif diff --git a/serve/serve.c b/serve/serve.c index 95ac37f..ab386ae 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -185,6 +185,8 @@ typedef struct { bool reads_first; /* the owner's: a round's reads before its writes */ uint64_t next[PURE_MODULE_INPUTS]; /* when each timer ticks next (monotonic clock, ms; 0: not yet begun) */ uint64_t ticks; + uint64_t memory_refused; /* events refused for the app's memory budget */ + bool answers_wait; /* an answer the app had no room for: tried again after ANSWER_WAIT_MS */ /* The owner's: the last save's snapshot (NULL: the owner shows views), * taken under the lock, and the app's changes it shows. */ bool publishes; @@ -517,11 +519,12 @@ static void answer(Connection *c, PureOutcome result) { } static const char *const fault_names[] = {"none", "divide", "non_finite", "clamp", "limit", "not_found", "encoding", - "exists"}; + "exists", "memory"}; /* An app route's answer: a view's value, or what an event replied. A failed - * event names its fault: 404 not_found, 409 exists, 422 for the others; - * a view without a value is 404 no_value. */ + * event names its fault: 404 not_found, 409 exists, 503 memory (the app's + * budget: again later), 422 for the others; a view without a value is 404 + * no_value. */ static void answer_app(Connection *c, PureOutcome result, bool view, PureFault fault, const char *type) { const char *named = fault != PURE_FAULT_NONE ? fault_names[fault] : NULL; if (result.status == PURE_OK) { @@ -532,6 +535,8 @@ static void answer_app(Connection *c, PureOutcome result, bool view, PureFault f respond(c, 404, "Not Found", view ? "no_value" : named ? named : "not_found", 0, (PureSpan){0}); } else if (result.status == PURE_CONFLICT) { respond(c, 409, "Conflict", named ? named : "conflict", 0, (PureSpan){0}); + } else if (fault == PURE_FAULT_MEMORY) { + respond(c, 503, "Service Unavailable", named, 0, (PureSpan){0}); } else if (named && (result.status == PURE_LIMIT || result.status == PURE_INVALID)) { respond(c, 422, "Unprocessable Content", named, 0, (PureSpan){0}); } else if (result.status == PURE_INVALID) { @@ -1346,6 +1351,10 @@ static PureResult show_given(PureApp *app, const Route *route, size_t available, return PURE_OK; } +/* The app refused a message it was given (a fault of its own): it was + * delivered, and is heard. Refused for memory it was not. */ +static bool app_refused(const PureApp *app) { return app->fault != PURE_FAULT_NONE && app->fault != PURE_FAULT_MEMORY; } + /* An input's event: a text, a key pressed (down, then up), or a message * whose reply (a field of it) is borrowed from the request. */ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect *effect, Applied *out) { @@ -1360,6 +1369,7 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect unsigned count = get_u32(delivery + 8); const unsigned char *at = effect->value.data + 4 + DELIVERY; out->gap = false; + out->fault = PURE_FAULT_NONE; for (unsigned i = 0; i < count; ++i) { uint32_t size = get_u32(at); PureSpan record = {at + 4, size}; @@ -1370,11 +1380,13 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect break; } PureResult status = pure_app_deliver(app, route->target, source, first + i, record, NULL); - if (status != PURE_OK && app->fault == PURE_FAULT_NONE) break; /* not applied: sent again */ + if (status != PURE_OK && !app_refused(app)) { /* not applied: sent again */ + out->fault = app->fault; + break; + } heard = first + i; } out->heard = pure_app_heard(app, source); - out->fault = PURE_FAULT_NONE; return PURE_OK; /* the save says whether they are stored */ } unsigned batch = header & DELIVERED ? get_u32(delivery + 8) : 1; @@ -1396,14 +1408,15 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect source[delivery[12]] = 0; status = pure_app_deliver(app, route->target, source, get_u64(delivery), payload, table ? placed : NULL); /* The rest of a batch, in order: an app's refusal was heard and the - * batch goes on; another failure stops it, to be sent again. */ - for (unsigned i = 1; i < batch && (status == PURE_OK || app->fault != PURE_FAULT_NONE); ++i) { + * batch goes on; another failure (the budget's too) stops it, to + * be sent again. */ + for (unsigned i = 1; i < batch && (status == PURE_OK || app_refused(app)); ++i) { const unsigned char *at = effect->value.data + lead + i * table; for (size_t k = 0; k < 2 * (size_t)shape->field_count; ++k) placed[k] = get_u32(at + 4 * k); PureSpan next = {payload.data + i * shape->prefix, payload.size - i * shape->prefix}; status = pure_app_deliver(app, route->target, source, get_u64(delivery) + i, next, placed); } - if (batch > 1 && app->fault != PURE_FAULT_NONE) status = PURE_OK; + if (batch > 1 && app_refused(app)) status = PURE_OK; } else status = table ? pure_app_message_fields(app, route->target, NULL, -1, payload, placed) : pure_app_message(app, route->target, NULL, -1, payload); @@ -1428,6 +1441,7 @@ static void apply_effect(Executor *ex, const Job *job, Applied *out) { if (job->route.kind == ROUTE_INPUT) { out->status = apply_input(app, &job->route, effect, out); if (out->status == PURE_OK) ++ex->burst_events; + ex->memory_refused += out->fault == PURE_FAULT_MEMORY; return; } bool read = effect->input == PURE_INPUT_READ; @@ -1549,22 +1563,34 @@ static void after_save(Executor *ex) { free(q.fresh); } -/* The answers the wires made become events, in order; then what the wires +enum { ANSWER_WAIT_MS = 100 }; + +/* The answers the wires made become events, in order, each taking its send + * out of the outbox; one the app has no room for waits, with the wire's + * after it, and is tried again after ANSWER_WAIT_MS. Then what the wires * delivered leaves the outbox (with the next save). True when an answer was * applied (the caller saves). */ static bool take_deliveries(Executor *ex) { bool answered = false; + ex->answers_wait = false; for (unsigned i = 0; i < ex->wire_count; ++i) { PureWireAnswer answer; while (pure_wire_answer(ex->wires[i], &answer)) { ex->app.now = wall_ms(); - PureResult result = pure_app_deliver(&ex->app, answer.input, answer.source, answer.seq, answer.payload, - answer.placed); + PureResult result = pure_app_answer(&ex->app, answer.input, pure_wire_output(ex->wires[i]), answer.source, + answer.seq, answer.payload, answer.placed); + if (result != PURE_OK && ex->app.fault == PURE_FAULT_MEMORY) { + pure_app_finish(&ex->app); + ++ex->memory_refused; + ex->answers_wait = true; + break; + } if (result != PURE_OK && ex->app.fault == PURE_FAULT_NONE) { ex->held[i] = true; /* not applied and not refused: sent again after a restart */ fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%" PRIu64 ",\"error\":\"its answer was not applied\"}\n", answer.source, answer.seq); } + pure_wire_answered(ex->wires[i]); pure_app_finish(&ex->app); answered = true; ++ex->answers; @@ -1711,7 +1737,7 @@ static void hand_back(void *context, Applied *applied, size_t count) { LinkJob *link = applied[i].link; if (!link) continue; /* records a link delivered: stored now, or not */ pure_link_done(link->listener, (PureLinkDone){link->token, applied[i].heard, applied[i].status == PURE_OK, - applied[i].gap, false}); + applied[i].gap, applied[i].fault == PURE_FAULT_MEMORY}); free(link->ingress); free(link); } @@ -1948,7 +1974,8 @@ static void *run_owner(void *context) { owner->delivered = false; pthread_mutex_unlock(&owner->lock); long due = run_timers(&owner->executor); - bool answered = delivered && take_deliveries(&owner->executor); + bool answered = (delivered || owner->executor.answers_wait) && take_deliveries(&owner->executor); + if (owner->executor.answers_wait && (due < 0 || due > ANSWER_WAIT_MS)) due = ANSWER_WAIT_MS; pthread_mutex_lock(&owner->lock); /* Answers are saved with the next round, or now if none waits. */ if (answered && !owner->count && !owner->stopping) { @@ -2193,6 +2220,8 @@ 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; ex->show = show; @@ -2203,6 +2232,7 @@ static bool start_executor(Executor *ex, const PureModule *module, uint32_t show return false; } pure_app_background_checkpoints(&ex->app, true); + pure_app_memory(&ex->app, memory_budget); return true; } @@ -2600,6 +2630,7 @@ int main(int argc, char **argv) { else if (!strcmp(argv[i], "--senders")) senders_path = argv[i + 1]; 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], "--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; else if (!strcmp(argv[i], "--threads") && number(argv[i + 1], 256, &value) && value) threads = (unsigned)value; @@ -2721,7 +2752,7 @@ int main(int argc, char **argv) { pure_link_listener_close(link_listener, &heard); /* after the owner, which tells it what it stored */ size_t active = 0, retained = 0; uint64_t accepted = 0, requests = 0, bursts = owner.executor.bursts, events = owner.executor.burst_events, expired = 0, - ticks = owner.executor.ticks, shown = 0, checkpoints = 0; + ticks = owner.executor.ticks, shown = 0, checkpoints = 0, memory_refused = owner.executor.memory_refused; for (unsigned i = 0; i < threads; ++i) { shown += hosts[i].shown; active += hosts[i].active; @@ -2731,6 +2762,7 @@ int main(int argc, char **argv) { bursts += hosts[i].local.bursts; events += hosts[i].local.burst_events; ticks += hosts[i].local.ticks; + memory_refused += hosts[i].local.memory_refused; expired += hosts[i].expired; stop_shard(&hosts[i]); } @@ -2750,6 +2782,7 @@ int main(int argc, char **argv) { sent.refused += c.refused; pure_link_close(owner.executor.links[i]); /* what was not stored stays in the outbox */ } + size_t state_bytes = owned ? pure_app_state_bytes(&owner.executor.app) : 0; if (owned) { checkpoints = owner.executor.checkpoints; stop_executor(&owner.executor); @@ -2768,12 +2801,13 @@ int main(int argc, char **argv) { ",\"link_refused\":%" PRIu64 ",\"listener_packets\":%" PRIu64 ",\"listener_received\":%" PRIu64 ",\"listener_deliveries\":%" PRIu64 ",\"listener_woken\":%" PRIu64 ",\"listener_remembered\":%" PRIu64 ",\"listener_swept\":%" PRIu64 ",\"listener_active\":%" PRIu64 ",\"listener_keyed\":%" PRIu64 - ",\"listener_secrets\":%" PRIu64 ",\"listener_epochs\":%" PRIu64 ",\"listener_refused\":%" PRIu64 "}\n", + ",\"listener_secrets\":%" PRIu64 ",\"listener_epochs\":%" PRIu64 ",\"listener_refused\":%" PRIu64 + ",\"state_bytes\":%zu,\"memory_refused\":%" PRIu64 "}\n", active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false", owner.executor.publications, shown, delivered, attempts, owner.executor.answers, groups, grouped, checkpoints, sent.packets, sent.records, sent.resent, sent.lost, sent.acknowledged, sent.stored, sent.refused, heard.packets, heard.received, heard.deliveries, heard.woken, heard.remembered, heard.swept, - heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused); + heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused, state_bytes, memory_refused); free(hosts); free(thread); return 0; @@ -2781,11 +2815,13 @@ 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] " + "[--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] [--link-idle-ms 30000] [--memory BYTES] " "[--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" " --send OUTPUT=http[s]://host:port/path: to anything else, over HTTP\n" - " --senders FILE: more senders the --link-port takes, a line each: NAME KEY\n", stderr); + " --senders FILE: more senders the --link-port takes, a line each: NAME KEY\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", stderr); return 2; } diff --git a/tests/check.py b/tests/check.py index 3c77dc4..e15a996 100644 --- a/tests/check.py +++ b/tests/check.py @@ -444,6 +444,20 @@ view leaders = each e in board { """ +def post(port, path, body): + """A JSON body posted on its own connection: status, X-Pure-Error.""" + import http.client + connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30) + try: + data = json.dumps(body).encode() + connection.request('POST', path, data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'}) + response = connection.getresponse() + response.read() + return response.status, response.getheader('X-Pure-Error') + finally: + connection.close() + + def http(port, method, path, body=b''): """One request on its own connection: status, X-Pure-Error, content type, body.""" import http.client @@ -1074,6 +1088,131 @@ def server_links_refused(aup, pure, work): return {'said': said[0]['error'], 'packets': stats['a']['link_packets']} + +MEMORY_RECEIVER = """record Order { + item: text + count = 0 +} +state kept: list of Order +state got = 0 +input order = message { + item: text + count: u32 + to: { + city: text + } +} +input clean = message { + below: u32 +} +on order { + kept.append(Order { item: order.item, count: order.count }) + got = got + 1 +} +on clean { + for o in kept where o.count < clean.below { + kept.remove(o) + } +} +view count = got +view held = kept.count +""" + + +def server_memory(aup, pure, work): + """What an app holds, bounded (--memory, docs/design/013-memory-budget.md). + A sender whose link's receiver is down fills its outbox: its next buy is + refused, 503 memory, and changes nothing, while its views still read; + once the receiver is up and stores what waited, it takes buys again. The + receiver, started again with a budget below what it keeps, refuses the + link's deliveries (and a client's order, 503 memory) but not the clean + that frees them; meanwhile its link hands the records over a few times a + second, not at every turn, and once freed it takes them all, once.""" + import select + import socket + import time + sender_module, receiver_module = work / 'memory-sender.pmod', work / 'memory-receiver.pmod' + for module, text in ((sender_module, LINK_SENDER), (receiver_module, MEMORY_RECEIVER)): + source = module.with_suffix('.aup') + source.write_text(text) + compile_module(aup, source, module) + + def free_port(kind=socket.SOCK_STREAM): + with socket.socket(socket.AF_INET, kind) as probe: + probe.bind(('127.0.0.1', 0)) + return probe.getsockname()[1] + + def start(args, port): + process = subprocess.Popen([str(a) for a in args] + ['--port', str(port)], stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + ready, _, _ = select.select([process.stdout], [], [], 10) + line = process.stdout.readline() if ready else b'' + assert line, f'host did not start: {process.stderr.read(2000) if process.poll() is not None else ""}' + return process, json.loads(line) + + def stop(process): + process.terminate() + _, error = process.communicate(timeout=30) + assert process.returncode == 0, error[-1000:] + return [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] + + def number(port, view): + return int(http(port, 'GET', f'/views/{view}')[3]) + + def until(test, seconds): # test once more at most after the deadline: it may post + deadline = time.time() + seconds + while not test(): + if time.time() > deadline: + return False + time.sleep(0.02) + return True + + keys = {} + for name in ('sender', 'receiver'): + process, info = start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'memory-key', + '--key', work / f'memory-{name}.key'], 0) + keys[name] = info['public_key'] + stop(process) + port_a, port_b, link_port = free_port(), free_port(), free_port(socket.SOCK_DGRAM) + receiving = [pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'memory-b', + '--key', work / 'memory-receiver.key', '--link-port', link_port, + '--sender', f'memory-sender.ordered={keys["sender"]}'] + sender = start([pure / 'build/pure-serve', '--module', sender_module, '--store', work / 'memory-a', + '--key', work / 'memory-sender.key', '--send', f'ordered=pure://127.0.0.1:{link_port}/order', + '--seal', f'ordered={keys["receiver"]}', '--memory', 1000], port_a)[0] + receiver, stats, item = None, {}, 'x' * 40 + try: + accepted, refused = 0, None + while accepted < 200 and not refused: + status = post(port_a, '/inputs/buy', {'item': item, 'count': 1}) + accepted += status[0] == 200 + refused = status if status[0] != 200 else None + assert refused == (503, 'memory') and 3 <= accepted < 200, (accepted, refused) + assert number(port_a, 'sent') == accepted + receiver = start(receiving, port_b)[0] + assert until(lambda: number(port_b, 'count') == accepted, 20), number(port_b, 'count') + assert until(lambda: post(port_a, '/inputs/buy', {'item': item, 'count': 1})[0] == 200, 10) + assert until(lambda: number(port_b, 'count') == accepted + 1, 10) + stop(receiver) + receiver = start(receiving + ['--memory', 300], port_b)[0] # below what it keeps + order = {'item': item, 'count': 1, 'to': {'city': 'Seoul'}} + assert post(port_b, '/inputs/order', order) == (503, 'memory') + for _ in range(3): + assert post(port_a, '/inputs/buy', {'item': item, 'count': 1})[0] == 200 + time.sleep(2.0) + assert number(port_b, 'count') == accepted + 1 and number(port_a, 'sent') == accepted + 4 + assert post(port_b, '/inputs/clean', {'below': 100})[0] == 200 + assert until(lambda: number(port_b, 'count') == accepted + 4, 5), number(port_b, 'count') + assert number(port_b, 'held') == 3 + finally: + for name, process in (('b', receiver), ('a', sender)): + if process: + stats[name] = stop(process) + assert stats['a']['memory_refused'] >= 1 and stats['a']['state_bytes'] <= 1000, stats['a'] + assert 2 <= stats['b']['memory_refused'] <= 40 and stats['b']['state_bytes'] <= 300, stats['b'] + return {'sends_held': accepted, 'receiver_refused': stats['b']['memory_refused'], + 'link_resent': stats['a']['link_resent']} + def server_links_independent(aup, pure, work): """A receiver written from docs/design/012-link-protocol.md alone, in Python with the cryptography package and no code of link/: a Pure node @@ -2312,6 +2451,113 @@ def server_answers(aup, pure, work): return {'orders': sent, 'paid': paid, 'declined': declined, 'answers_after_restart': stats['answers']} + +RECEIPT_SOURCE = """record Receipt { + order: text + note: text + status = 0 +} +state receipts: list of Receipt +state sent = 0 +state answered = 0 +input buy = message { + order: text +} +output charge = message { + order: text +} +input charged = message { + order: text + note: text + http_status: u32 +} +input clean = message { + below: u32 +} +on buy { + sent = sent + 1 + send charge { order: buy.order } +} +on charged { + answered = answered + 1 + receipts.append(Receipt { order: charged.order, note: charged.note, status: charged.http_status }) +} +on clean { + for r in receipts where r.status < clean.below { + receipts.remove(r) + } +} +view sent_view = sent +view answered_view = answered +view receipts_view = receipts.count +""" + + +def server_answers_memory(aup, pure, work): + """Answers (--reply) under a memory budget: an answer keeps a receipt + larger than its send, so once receipts fill the budget the next answer + waits (it is not lost, and not applied twice) and buys are refused, 503 + memory. A clean that frees the receipts lets the answers in, each once.""" + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + import threading + import time + module = work / 'receipt.pmod' + module.with_suffix('.aup').write_text(RECEIPT_SOURCE) + compile_module(aup, module.with_suffix('.aup'), module) + seen, lock = {}, threading.Lock() + + class Api(BaseHTTPRequestHandler): + protocol_version = 'HTTP/1.1' + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) + with lock: + seen[body['order']] = seen.get(body['order'], 0) + 1 + data = json.dumps({'note': 'n' * 200}).encode() + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Api) + threading.Thread(target=server.serve_forever, daemon=True).start() + host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'receipt-store', '--memory', 3000, + '--send', f'charge=http://127.0.0.1:{server.server_address[1]}/charge', '--reply', 'charge=charged']) + + def number(view): + return int(http(host.port, 'GET', f'/views/{view}')[3]) + + try: + bought, refusals = 0, 0 + while refusals < 3 and bought < 200: + status = post(host.port, '/inputs/buy', {'order': f'o{bought}'}) + if status[0] == 200: + bought += 1 + else: + assert status == (503, 'memory'), status + refusals += 1 + time.sleep(0.02) + time.sleep(0.5) + assert 3 <= bought < 200 and number('sent_view') == bought, bought + waiting = bought - number('answered_view') + assert waiting >= 1, (bought, number('answered_view')) + assert post(host.port, '/inputs/clean', {'below': 1000})[0] == 200 + deadline = time.time() + 10 + while number('answered_view') < bought and time.time() < deadline: + time.sleep(0.02) + assert number('answered_view') == bought and number('receipts_view') == waiting, (number('answered_view'), waiting) + finally: + stats = host.stop() + server.shutdown() + with lock: + assert len(seen) == bought and all(count == 1 for count in seen.values()), seen + assert stats[-1]['memory_refused'] >= refusals + 1, stats[-1] + return {'bought': bought, 'answers_waited': waiting, 'refused': stats[-1]['memory_refused']} + BEAT_SOURCE = """state beats = 0 state last = 0 state stamped = 0 @@ -2605,6 +2851,7 @@ def main(): report['server_links_udp'] = server_links_udp(aup, pure, work) report['server_links_udp_lossy'] = server_links_udp_lossy(aup, pure, work) report['server_links_refused'] = server_links_refused(aup, pure, work) + report['server_memory'] = server_memory(aup, pure, work) report['server_links_independent'] = server_links_independent(aup, pure, work) report['server_links_moved'] = server_links_moved(aup, pure, work) report['server_webhook'] = server_webhook(aup, pure, work) @@ -2614,6 +2861,7 @@ def main(): report['server_rows'] = server_rows(aup, pure, work) report['server_arguments'] = server_arguments(aup, pure, work) report['server_answers'] = server_answers(aup, pure, work) + report['server_answers_memory'] = server_answers_memory(aup, pure, work) print(json.dumps(report)) |