diff options
| -rw-r--r-- | serve/serve.c | 307 | ||||
| -rw-r--r-- | tests/check.py | 84 |
2 files changed, 353 insertions, 38 deletions
diff --git a/serve/serve.c b/serve/serve.c index ec75fec..b5112d2 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -213,10 +213,15 @@ typedef struct { unsigned link_count; uint32_t link_output[PURE_MODULE_INPUTS]; uint64_t link_trimmed[PURE_MODULE_INPUTS]; - /* Each outlet's place in the outbox, wires first: its sends before it - * are queued (feed_outlets); and the app's sends_gone when they were - * last moved back. How often an outlet's window was full. */ - size_t outlet_from[2 * PURE_MODULE_INPUTS]; + /* The owner's: outputs handed to other apps of the host (send=OUTPUT=APP/INPUT), and how far each was trimmed. */ + struct Local *locals[PURE_MODULE_INPUTS]; + unsigned local_count; + uint64_t local_trimmed[PURE_MODULE_INPUTS], handed; + /* Each outlet's place in the outbox, wires first, then links, then + * locals: its sends before it are queued (feed_outlets); and the app's + * sends_gone when they were last moved back. How often an outlet's + * window was full. */ + size_t outlet_from[3 * PURE_MODULE_INPUTS]; uint64_t gone_seen, windows_full; struct Writer *writer; /* the thread writing the app's checkpoints, once one was due */ uint64_t checkpoints; /* images it wrote */ @@ -306,7 +311,7 @@ static void stop_writer(Executor *ex) { /* A request made ready: its effect, what it names, and the thread that * answers it (or, for a surface's event, the waiter it wakes). */ -typedef struct LinkJob { PureLinkListener *listener; void *token; unsigned char *ingress; } LinkJob; +typedef struct LinkJob { PureLinkListener *listener; void *token; unsigned char *ingress; struct Local *local; } LinkJob; typedef struct { Host *host; PureEffect effect; Route route; Waiter *waiter; LinkJob *link; } Job; /* The owner of an app with states served by several connection threads: @@ -1644,8 +1649,166 @@ static void publish(Executor *ex) { enum { WINDOW_SENDS = 65536 }; static size_t window_bytes = 4u << 20; +/* An output's sends handed to another app of the host (send=OUTPUT=APP/INPUT, + * docs/design/017's stage 2): no socket, seal or packet. The sender's turn + * queues its saved sends here (feed_outlets) and hands them, a batch at a + * time, to the receiving app's queue as records a link delivers; the + * receiver's turn applies them after its last number heard and, its save + * over, says what it heard (local_done); the sender's next turn trims its + * outbox so far. What the receiver did not store is queued again from the + * outbox, after 100 ms when it had no room. */ +enum { LOCAL_RETRY_MS = 100 }; +typedef struct Local { + Owner *from, *to; + uint32_t output, input; /* the sender's output, the receiver's input */ + char source[PURE_SOURCE + 1]; + unsigned char fields[PURE_FIELDS_NAMED]; /* the sender's, when the input takes them by name */ + size_t fields_size; + pthread_mutex_t lock; /* the receiver's turn says what it stored */ + uint64_t queued, delivered; /* the last number queued, the last the receiver stored */ + unsigned char *batch; /* queued, not handed over yet: a u32le size and the bytes of each */ + size_t batch_size, batch_capacity, out_bytes; + uint64_t first, out_last; /* the batch's first number; the last of the one handed over */ + unsigned count; + bool out, rewind; /* a batch is with the receiver; it did not store all of one */ + uint64_t retry; /* when one it had no room for goes again (the monotonic clock in ms) */ +} Local; + +static uint64_t local_queued(Local *l) { + pthread_mutex_lock(&l->lock); + uint64_t queued = l->queued; + pthread_mutex_unlock(&l->lock); + return queued; +} + +static uint64_t local_delivered(Local *l) { + pthread_mutex_lock(&l->lock); + uint64_t delivered = l->delivered; + pthread_mutex_unlock(&l->lock); + return delivered; +} + +static size_t local_held(Local *l) { + pthread_mutex_lock(&l->lock); + size_t held = l->batch_size + l->out_bytes; + pthread_mutex_unlock(&l->lock); + return held; +} + +static bool local_queue(Local *l, uint64_t seq, PureSpan payload) { + pthread_mutex_lock(&l->lock); + bool room = grow(&l->batch, &l->batch_capacity, l->batch_size + 4 + payload.size, 1); + if (room) { + put_u32(l->batch + l->batch_size, (uint32_t)payload.size); + if (payload.size) memcpy(l->batch + l->batch_size + 4, payload.data, payload.size); + if (!l->count) l->first = seq; + l->batch_size += 4 + payload.size; + ++l->count; + l->queued = seq; + } + pthread_mutex_unlock(&l->lock); + return room; +} + +/* The batch queued goes to the receiver, unless one is there or it waits + * to try again; as a link's delivery is laid out (link_deliver). */ +static void local_hand(Executor *ex, Local *l, uint64_t now) { + pthread_mutex_lock(&l->lock); + if (l->out || !l->count || (l->retry && now < l->retry)) { + pthread_mutex_unlock(&l->lock); + return; + } + size_t length = strlen(l->source), size = 4 + DELIVERY + l->batch_size + l->fields_size; + unsigned char *ingress = malloc(size); + LinkJob *link = malloc(sizeof(*link)); + if (!ingress || !link) { + pthread_mutex_unlock(&l->lock); + free(ingress); + free(link); + return; + } + memset(ingress, 0, 4 + DELIVERY); + put_u32(ingress, (uint32_t)l->batch_size | DELIVERED | STORED_RECORDS); + put_u64(ingress + 4, l->first); + put_u32(ingress + 12, l->count); + ingress[16] = (unsigned char)length; + memcpy(ingress + 17, l->source, length); + memcpy(ingress + 4 + DELIVERY, l->batch, l->batch_size); + if (l->fields_size) memcpy(ingress + 4 + DELIVERY + l->batch_size, l->fields, l->fields_size); + *link = (LinkJob){.ingress = ingress, .local = l}; + l->out = true; + l->out_bytes = l->batch_size; + l->out_last = l->first + l->count - 1; + l->batch_size = 0; + l->count = 0; + l->retry = 0; + pthread_mutex_unlock(&l->lock); + Job job = {.route = {.kind = ROUTE_INPUT, .target = l->input}, .link = link, + .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {ingress, size}}}; + Owner *to = l->to; + pthread_mutex_lock(&to->lock); + bool queued = !to->stopping && grow(&to->jobs, &to->capacity, to->count + 1, sizeof(*to->jobs)); + if (queued) { + to->jobs[to->count++] = job; + ++to->arrivals; + pthread_cond_signal(&to->waiting); + } + pthread_mutex_unlock(&to->lock); + if (queued) { + ++ex->handed; + schedule(to); + return; + } + free(ingress); + free(link); + pthread_mutex_lock(&l->lock); + l->out = false; + l->rewind = true; /* queued again from the outbox */ + pthread_mutex_unlock(&l->lock); +} + +/* The receiver's turn, its save over: what it heard from the batch. All of + * it, and the sender trims so far; else what it did not store is queued + * again (after LOCAL_RETRY_MS when it had no room). The sender's turn + * comes either way. */ +static void local_done(Local *l, uint64_t heard, bool stored, bool full) { + pthread_mutex_lock(&l->lock); + l->out = false; + l->out_bytes = 0; + if (stored && heard > l->delivered) l->delivered = heard; + if (!stored || heard < l->out_last) { + l->rewind = true; + if (full) l->retry = now_ms() + LOCAL_RETRY_MS; + } + pthread_mutex_unlock(&l->lock); + Owner *from = l->from; + pthread_mutex_lock(&from->lock); + from->delivered = true; + pthread_cond_signal(&from->waiting); + pthread_mutex_unlock(&from->lock); + schedule(from); +} + +/* The milliseconds until a batch an app's local outlet holds back goes again (-1: none waits). */ +static long locals_due(Executor *ex) { + long due = -1; + uint64_t now = now_ms(); + for (unsigned i = 0; i < ex->local_count; ++i) { + Local *l = ex->locals[i]; + pthread_mutex_lock(&l->lock); + if (l->retry && (l->count || l->rewind)) { + long wait = l->retry > now ? (long)(l->retry - now) : 0; + if (due < 0 || wait < due) due = wait; + } + pthread_mutex_unlock(&l->lock); + } + return due; +} + static uint32_t outlet_output(Executor *ex, unsigned i) { - return i < ex->wire_count ? pure_wire_output(ex->wires[i]) : ex->link_output[i - ex->wire_count]; + return i < ex->wire_count ? pure_wire_output(ex->wires[i]) + : i < ex->wire_count + ex->link_count ? ex->link_output[i - ex->wire_count] + : ex->locals[i - ex->wire_count - ex->link_count]->output; } /* The saved sends not yet queued go to their outlets, in order, as far as @@ -1654,16 +1817,17 @@ static uint32_t outlet_output(Executor *ex, unsigned i) { * that left the outbox since (they may have been before it), so the sends * it queued before are not looked at again, however many wait. */ static void feed_outlets(Executor *ex) { - unsigned outlets = ex->wire_count + ex->link_count; + unsigned outlets = ex->wire_count + ex->link_count + ex->local_count; uint64_t gone = ex->app.sends_gone - ex->gone_seen; ex->gone_seen = ex->app.sends_gone; for (unsigned i = 0; i < outlets; ++i) { PureWire *wire = i < ex->wire_count ? ex->wires[i] : NULL; - PureLink *link = wire ? NULL : ex->links[i - ex->wire_count]; + PureLink *link = !wire && i < ex->wire_count + ex->link_count ? ex->links[i - ex->wire_count] : NULL; + Local *local = !wire && !link ? ex->locals[i - ex->wire_count - ex->link_count] : NULL; uint32_t output = outlet_output(ex, i); - uint64_t queued = wire ? pure_wire_queued(wire) : pure_link_queued(link), - delivered = wire ? pure_wire_delivered(wire) : pure_link_delivered(link); - size_t held = wire ? pure_wire_held(wire) : pure_link_held(link), at = ex->outlet_from[i]; + uint64_t queued = wire ? pure_wire_queued(wire) : link ? pure_link_queued(link) : local_queued(local), + delivered = wire ? pure_wire_delivered(wire) : link ? pure_link_delivered(link) : local_delivered(local); + size_t held = wire ? pure_wire_held(wire) : link ? pure_link_held(link) : local_held(local), at = ex->outlet_from[i]; at = at > gone ? at - (size_t)gone : 0; for (; at < ex->app.sent_saved; ++at) { uint32_t of; @@ -1674,13 +1838,16 @@ static void feed_outlets(Executor *ex) { ++ex->windows_full; break; } - if (wire ? !pure_wire_queue(wire, seq, payload) : !pure_link_queue(link, seq, payload.data, payload.size)) + if (wire ? !pure_wire_queue(wire, seq, payload) + : link ? !pure_link_queue(link, seq, payload.data, payload.size) : !local_queue(local, seq, payload)) break; queued = seq; held += payload.size; } ex->outlet_from[i] = at; } + uint64_t now = ex->local_count ? now_ms() : 0; + for (unsigned i = 0; i < ex->local_count; ++i) local_hand(ex, ex->locals[i], now); } static void after_save(Executor *ex) { @@ -1732,6 +1899,22 @@ static bool take_deliveries(Executor *ex) { pure_app_trim(&ex->app, ex->link_output[i], stored); ex->link_trimmed[i] = stored; } + for (unsigned i = 0; i < ex->local_count; ++i) { + Local *l = ex->locals[i]; + pthread_mutex_lock(&l->lock); + if (l->rewind && !l->out) { /* from what it stored, again */ + l->queued = l->delivered; + l->batch_size = 0; + l->count = 0; + l->rewind = false; + ex->outlet_from[ex->wire_count + ex->link_count + i] = 0; + } + uint64_t stored = l->delivered; + pthread_mutex_unlock(&l->lock); + if (stored <= ex->local_trimmed[i]) continue; + pure_app_trim(&ex->app, l->output, stored); + ex->local_trimmed[i] = stored; + } feed_outlets(ex); /* their windows moved on */ return answered; } @@ -1865,9 +2048,11 @@ static void hand_back(void *context, Applied *applied, size_t count) { for (size_t i = 0; i < count; ++i) { events = events || applied[i].waiter; 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, applied[i].fault == PURE_FAULT_MEMORY}); + if (!link) continue; /* records a link (or another app of the host) delivered: stored now, or not */ + if (link->local) local_done(link->local, applied[i].heard, applied[i].status == PURE_OK, applied[i].fault == PURE_FAULT_MEMORY); + else + pure_link_done(link->listener, (PureLinkDone){link->token, applied[i].heard, applied[i].status == PURE_OK, + applied[i].gap, applied[i].fault == PURE_FAULT_MEMORY}); free(link->ingress); free(link); } @@ -2171,6 +2356,9 @@ static bool owner_turn(Owner *owner, bool alone, long *due) { *due = run_timers(ex); bool answered = (delivered || ex->answers_wait) && take_deliveries(ex); if (ex->answers_wait && (*due < 0 || *due > ANSWER_WAIT_MS)) *due = ANSWER_WAIT_MS; + long held_back = ex->local_count ? locals_due(ex) : -1; + if (held_back >= 0 && (*due < 0 || held_back < *due)) *due = held_back; + if (held_back == 0) feed_outlets(ex); /* its time came */ pthread_mutex_lock(&owner->lock); /* Answers are saved with the requests waiting, or now if none waits. */ if (answered && !owner->count) { @@ -2784,7 +2972,7 @@ static bool link_deliver(void *context, const PureLinkDelivery *d) { memcpy(ingress + 17, d->source, length); memcpy(ingress + 4 + DELIVERY, d->records, d->size); if (d->fields_size) memcpy(ingress + 4 + DELIVERY + d->size, d->fields, d->fields_size); /* laid out with those */ - *link = (LinkJob){link_listener, d->token, ingress}; + *link = (LinkJob){.listener = link_listener, .token = d->token, .ingress = ingress}; Job job = {.route = {.kind = ROUTE_INPUT, .target = d->input % PURE_MODULE_INPUTS}, .link = link, .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {ingress, size}}}; pthread_mutex_lock(&owner->lock); @@ -2814,7 +3002,58 @@ static void wire_delivered(void *context) { /* --send OUTPUT=URL: that output's saved sends go to url, from NAME.OUTPUT. */ static unsigned char own_public[PURE_SEAL_KEY]; /* this node's key's public half, when it has one */ -static bool open_wires(Executor *ex, const PureModule *m, const char *const *sends, unsigned count, const char *name) { +/* send=OUTPUT=APP/INPUT in a host of apps: that output's sends handed to + * another app of it (Local), its input taking the output's shape or, by + * name, its fields. */ +static bool open_local(App *app, uint32_t output, const char *to, const char *source, char *error, size_t size) { + const char *slash = strchr(to, '/'); + size_t length = slash ? (size_t)(slash - to) : 0; + App *receiver = NULL; + for (unsigned a = 0; named_apps && slash && a < app_count && !receiver; ++a) + if (strlen(apps[a].name) == length && !memcmp(apps[a].name, to, length)) receiver = &apps[a]; + if (!receiver || !receiver->running || !receiver->owned) { + snprintf(error, size, "no running app '%.*s' in this host", (int)length, to); + return false; + } + const PureModule *m = &app->module, *r = &receiver->module; + int input = pure_module_input(r, slash + 1); + if (input < 0 || r->input[input].kind != PURE_KIND_MESSAGE || r->input[input].keyed) { + snprintf(error, size, "%s has no input '%s' taking messages", receiver->name, slash + 1); + return false; + } + Executor *ex = &app->owner.executor; + Local *l = calloc(1, sizeof(*l)); + if (!l) { + snprintf(error, size, "no memory"); + return false; + } + *l = (Local){.from = &app->owner, .to = &receiver->owner, .output = output, .input = (uint32_t)input}; + if (pthread_mutex_init(&l->lock, NULL)) { + free(l); + snprintf(error, size, "no lock"); + return false; + } + size_t named = strnlen(source, PURE_SOURCE); + memcpy(l->source, source, named); + l->source[named] = 0; + if (pure_module_shape(r, (uint32_t)input) != pure_module_shape(m, output)) { /* taken by name, or not at all */ + l->fields_size = pure_module_fields(m, output, l->fields); + if (!pure_module_takes(r, (uint32_t)input, l->fields, l->fields_size)) { + pthread_mutex_destroy(&l->lock); + free(l); + snprintf(error, size, "%s's input %s does not take the output's fields", receiver->name, slash + 1); + return false; + } + } + ex->locals[ex->local_count++] = l; + return true; +} + +static bool open_wires(App *app, const char *name) { + Executor *ex = &app->owner.executor; + const PureModule *m = &app->module; + const char *const *sends = app->sends; + unsigned count = app->send_count; for (unsigned i = 0; i < count; ++i) { const char *equals = strchr(sends[i], '='); char output[PURE_MODULE_NAME], source[PURE_SOURCE + 2], error[160]; @@ -2825,7 +3064,25 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen } memcpy(output, sends[i], length); output[length] = 0; + /* An output goes one way: its outbox is trimmed as far as its outlet's receiver stored. */ + int wired = pure_module_input(m, output); + for (unsigned k = 0; wired >= 0 && k < ex->wire_count + ex->link_count + ex->local_count; ++k) + if (outlet_output(ex, k) == (uint32_t)wired) { + fprintf(stderr, "send=%s: the output is wired already\n", sends[i]); + return false; + } int written = snprintf(source, sizeof(source), "%s.%s", name, output); + if (!strstr(equals + 1, "://")) { /* another app of the host: APP/INPUT */ + int index = pure_module_input(m, output); + if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) snprintf(error, sizeof(error), "no output '%s'", output); + else if (written <= 0 || written > PURE_SOURCE) snprintf(error, sizeof(error), "the name and output are longer than 64 bytes"); + if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT || written <= 0 || written > PURE_SOURCE || + !open_local(app, (uint32_t)index, equals + 1, source, error, sizeof(error))) { + fprintf(stderr, "send=%s: %s\n", sends[i], error); + return false; + } + continue; + } if (!strncmp(equals + 1, "pure://", 7)) { /* a link: no connection, sealed with the node's key */ int index = pure_module_input(m, output); if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) snprintf(error, sizeof(error), "no output '%s'", output); @@ -3230,7 +3487,7 @@ int main(int argc, char **argv) { App *app = &apps[a]; Executor *ex = &app->owner.executor; if (!app->running || !app->owned) continue; - if (open_wires(ex, &app->module, app->sends, app->send_count, named_apps ? app->name : name) && + if (open_wires(app, named_apps ? app->name : name) && wire_replies(ex, &app->module, app->replies, app->reply_count) && wire_seals(ex, &app->module, app->seals, app->seal_count, seal_key ? own_public : NULL)) continue; @@ -3238,7 +3495,8 @@ int main(int argc, char **argv) { snprintf(app->why, sizeof(app->why), "Its sends are wired wrong (said above)"); for (unsigned i = 0; i < ex->wire_count; ++i) pure_wire_close(ex->wires[i]); for (unsigned i = 0; i < ex->link_count; ++i) pure_link_close(ex->links[i]); - ex->wire_count = ex->link_count = 0; + for (unsigned i = 0; i < ex->local_count; ++i) free(ex->locals[i]); + ex->wire_count = ex->link_count = ex->local_count = 0; stop_executor(ex); app->running = false; --owners; @@ -3345,7 +3603,7 @@ int main(int argc, char **argv) { PureLinkCounts sent = {0}, heard = {0}; /* what links sent, and what the listener took */ pure_link_listener_close(link_listener, &heard); /* after the owner, which tells it what it stored */ size_t active = 0, retained = 0, state_bytes = 0; - uint64_t moved = 0, accepted = 0, requests = 0, bursts = 0, events = 0, expired = 0, ticks = 0, shown = 0, checkpoints = 0, + uint64_t moved = 0, handed = 0, accepted = 0, requests = 0, bursts = 0, events = 0, expired = 0, ticks = 0, shown = 0, checkpoints = 0, memory_refused = 0, windows_full = 0, publications = 0, answers = 0, delivered = 0, attempts = 0; for (unsigned i = 0; i < threads; ++i) { shown += hosts[i].shown; @@ -3386,6 +3644,12 @@ int main(int argc, char **argv) { sent.refused += c.refused; pure_link_close(ex->links[i]); /* what was not stored stays in the outbox */ } + for (unsigned i = 0; i < ex->local_count; ++i) { + free(ex->locals[i]->batch); + pthread_mutex_destroy(&ex->locals[i]->lock); + free(ex->locals[i]); + } + handed += ex->handed; state_bytes += pure_app_state_bytes(&ex->app); stop_executor(ex); checkpoints += ex->checkpoints; @@ -3413,7 +3677,8 @@ int main(int argc, char **argv) { heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused, state_bytes, memory_refused, windows_full); if (named_apps) { /* what each app did */ - fprintf(stderr, ",\"workers\":%u,\"moved_home\":%" PRIu64 ",\"apps\":[", owners ? workers : 0, moved); + fprintf(stderr, ",\"workers\":%u,\"moved_home\":%" PRIu64 ",\"handed\":%" PRIu64 ",\"apps\":[", + owners ? workers : 0, moved, handed); for (unsigned a = 0; a < app_count; ++a) { const Owner *o = &apps[a].owner; fprintf(stderr, "%s{\"name\":", a ? "," : ""); diff --git a/tests/check.py b/tests/check.py index f75204a..bb7ec25 100644 --- a/tests/check.py +++ b/tests/check.py @@ -3851,9 +3851,15 @@ twin {work / 'apps.pmod'} {work / 'store-a'} LINKS_SOURCE = """state n = 0 input put = message { k: u32 } output out = message { k: u32 } +output here = message { k: u32 } +output named = message { k: u32 } +output small = message { k: u32 } on put { n = n + 1 send out { k: put.k } + send here { k: put.k } + send named { k: put.k } + send small { k: put.k } } view count = n """ @@ -3875,17 +3881,34 @@ view count = count(g in got where g.k > 0) view sum = sum(g.k for g in got) view in_order = ordered """ +# The same, its input laid out otherwise (a field before k): it takes the sends by name. +LINKS_NAMED = LINKS_TAKER.replace('input take = message { k: u32 }', 'input take = message {\n note: text\n k: u32\n}') +# The same, and it can let go of what it got (and what it heard stays). +LINKS_SMALL = LINKS_TAKER + """input clear = message { all: u32 } +on clear { + for g in got { + got.remove(g) + } +} +view held = count(g in got where g.k > 0) +""" def server_apps_links(aup, pure, work): - """Apps of a host send (docs/design/017, stage 2): an app's send= wires - its output, its sends' source APP.OUTPUT, sealed with the host's key; an - app of another host takes them on that host's one --link-port, named - APP/INPUT. Every send arrives once and in order, and the stop line counts - the link's packets.""" + """Apps of a host send (docs/design/017, stage 2). An app's send= wires + each output: to an app of another host over a link (its sends' source + APP.OUTPUT, sealed with the host's key, taken on that host's one + --link-port as APP/INPUT), and to apps of its own host with no socket + (send=OUTPUT=APP/INPUT): one taking the same shape, one taking the + fields by name, one whose memory budget fills. Every send arrives once + and in order, across a kill -9 of the sending host between requests; + the full app's sends wait in the sender's outbox until it lets go of + what it holds, and none is lost.""" + import signal import socket import time - for name, text in (('links-src', LINKS_SOURCE), ('links-taker', LINKS_TAKER)): + for name, text in (('links-src', LINKS_SOURCE), ('links-taker', LINKS_TAKER), ('links-named', LINKS_NAMED), + ('links-small', LINKS_SMALL)): (work / f'{name}.aup').write_text(text) compile_module(aup, work / f'{name}.aup', work / f'{name}.pmod') serve = pure / 'build/pure-serve' @@ -3899,27 +3922,54 @@ def server_apps_links(aup, pure, work): keys[side] = host.info['public_key'] host.stop() (work / 'links-far.conf').write_text(f"far {work / 'links-taker.pmod'} {work / 'links-far'}\n") - (work / 'links-near.conf').write_text(f"src {work / 'links-src.pmod'} {work / 'links-src'} " - f"send=out=pure://127.0.0.1:{link_port}/far/take seal=out={keys['receiver']}\n") + (work / 'links-near.conf').write_text( + f"src {work / 'links-src.pmod'} {work / 'links-src'} send=out=pure://127.0.0.1:{link_port}/far/take " + f"seal=out={keys['receiver']} send=here=here/take send=named=byname/take send=small=small/take\n" + f"here {work / 'links-taker.pmod'} {work / 'links-here'}\n" + f"byname {work / 'links-named.pmod'} {work / 'links-byname'}\n" + f"small {work / 'links-small.pmod'} {work / 'links-small'} memory=1200\n") far = HttpHost([serve, '--apps', work / 'links-far.conf', '--key', work / 'links-receiver.key', '--link-port', link_port, '--sender', f"src.out={keys['sender']}"]) - near = HttpHost([serve, '--apps', work / 'links-near.conf', '--key', work / 'links-sender.key']) - sent = 300 + near_args = [serve, '--apps', work / 'links-near.conf', '--key', work / 'links-sender.key', '--threads', 2] + near = HttpHost(near_args) + sent, killed_at = 400, 150 try: for k in range(1, sent + 1): + if k == killed_at: # between requests: every send before it is stored, some not yet handed on + near.process.send_signal(signal.SIGKILL) + near.process.wait() + near = HttpHost(near_args) status, why, _, _ = http(near.port, 'POST', f'/src/inputs/put?k={k}') assert status == 200, (k, status, why) + + def views(port, app): + return [http(port, 'GET', f'/{app}/views/{view}')[3] for view in ('count', 'sum', 'in_order')] + + whole = [str(sent).encode(), str(sent * (sent + 1) // 2).encode(), b'true'] deadline = time.time() + 30 - while time.time() < deadline and http(far.port, 'GET', '/far/views/count')[3] != str(sent).encode(): + while time.time() < deadline and (views(far.port, 'far') != whole or views(near.port, 'here') != whole or + views(near.port, 'byname') != whole): time.sleep(0.05) - assert http(far.port, 'GET', '/far/views/count')[3] == str(sent).encode() - assert http(far.port, 'GET', '/far/views/sum')[3] == str(sent * (sent + 1) // 2).encode() - assert http(far.port, 'GET', '/far/views/in_order')[3] == b'true' + assert views(far.port, 'far') == whole, views(far.port, 'far') + assert views(near.port, 'here') == whole, views(near.port, 'here') + assert views(near.port, 'byname') == whole, views(near.port, 'byname') + # The small app is full: it holds some, the rest wait in src's outbox; it lets go, and they come. + held = int(http(near.port, 'GET', '/small/views/held')[3]) + assert 0 < held < sent, held + taken = held + for _ in range(sent): + assert http(near.port, 'POST', '/small/inputs/clear?all=1')[0] == 200 + time.sleep(0.3) # a held-back batch goes again after 100 ms + now = int(http(near.port, 'GET', '/small/views/held')[3]) + if not now: + break + taken += now + assert taken == sent and http(near.port, 'GET', '/small/views/in_order')[3] == b'true', taken finally: - packets = near.stop()[-1]['link_packets'] + near_stats = near.stop()[-1] far.stop() - assert packets > 0, packets - return {'sent': sent, 'link_packets': packets} + assert near_stats['handed'] > 0 and near_stats['link_packets'] > 0, near_stats + return {'sent': sent, 'killed_at': killed_at, 'handed': near_stats['handed'], 'small_held': held} def main(): |