aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--core/module.c14
-rw-r--r--core/module.h6
-rw-r--r--link/link-protocol.c6
-rw-r--r--link/link-protocol.h2
-rw-r--r--link/link.c7
-rw-r--r--link/link.h2
-rw-r--r--serve/deliver.c15
-rw-r--r--serve/deliver.h3
-rw-r--r--serve/serve.c96
-rw-r--r--tests/check.py105
10 files changed, 196 insertions, 60 deletions
diff --git a/core/module.c b/core/module.c
index ac46fb3..b370f4b 100644
--- a/core/module.c
+++ b/core/module.c
@@ -2711,6 +2711,7 @@ static PureResult let_sends_go(PureApp *app, uint32_t output, uint64_t through)
PureResult removed = pure_list_remove(app->outbox, i, &app->undo);
if (removed != PURE_OK) return removed;
--app->sent_saved;
+ ++app->sends_gone;
app->links_changed = true;
}
return PURE_OK;
@@ -2957,7 +2958,7 @@ PureResult pure_app_message_fields(PureApp *app, uint32_t input, const char *key
}
/* The outbox's item at: its output (the module's, by name), number and payload. */
-static bool sent_at(const PureApp *app, size_t at, uint32_t *output, uint64_t *seq, PureSpan *payload) {
+bool pure_app_sent(const PureApp *app, size_t at, uint32_t *output, uint64_t *seq, PureSpan *payload) {
PureCell name = pure_list_get(app->outbox, at, 0), bytes = pure_list_get(app->outbox, at, 2);
const PureModule *m = app->module;
uint32_t i = 0;
@@ -2973,16 +2974,7 @@ void pure_app_outbox(const PureApp *app, PureSentVisit visit, void *context) {
uint32_t output;
uint64_t seq;
PureSpan payload;
- if (sent_at(app, i, &output, &seq, &payload) && !visit(context, output, seq, payload)) return;
- }
-}
-
-void pure_app_outbox_newest(const PureApp *app, PureSentVisit visit, void *context) {
- for (size_t i = app->sent_saved; i-- > 0;) {
- uint32_t output;
- uint64_t seq;
- PureSpan payload;
- if (sent_at(app, i, &output, &seq, &payload) && !visit(context, output, seq, payload)) return;
+ if (pure_app_sent(app, i, &output, &seq, &payload) && !visit(context, output, seq, payload)) return;
}
}
diff --git a/core/module.h b/core/module.h
index 69acd67..7034b0e 100644
--- a/core/module.h
+++ b/core/module.h
@@ -340,6 +340,7 @@ typedef struct {
* outputs a module dropped. */
PureList *outbox;
size_t sent_saved, durable_sent_saved, durable_outbox;
+ uint64_t sends_gone; /* how many ever left it: a host's places in it move back at most that many */
uint64_t last_sent[PURE_MODULE_INPUTS], saved_sent[PURE_MODULE_INPUTS];
struct PureCarried { char name[PURE_MODULE_NAME]; uint64_t last; } *carried;
unsigned carried_count;
@@ -438,8 +439,9 @@ PureResult pure_app_replace(PureApp *, const PureModule *);
* and trims what its receiver has saved; a receiver applies each once. */
typedef bool (*PureSentVisit)(void *context, uint32_t output, uint64_t seq, PureSpan payload);
void pure_app_outbox(const PureApp *, PureSentVisit visit, void *context);
-/* The same, newest first (a host looking only for what is new stops early). */
-void pure_app_outbox_newest(const PureApp *, PureSentVisit visit, void *context);
+/* The saved send at a place in the outbox (0 the oldest, below sent_saved);
+ * false when its output is not the module's. */
+bool pure_app_sent(const PureApp *, size_t at, uint32_t *output, uint64_t *seq, PureSpan *payload);
/* Forget output's saved sends numbered up to through (saved with the next save). */
void pure_app_trim(PureApp *, uint32_t output, uint64_t through);
/* Deliver a message event from source (a sender's name, 1 to 64 printable
diff --git a/link/link-protocol.c b/link/link-protocol.c
index 7940d90..12f9a24 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -251,7 +251,7 @@ struct PureLinkSender {
/* Their bytes, one after another as queued (a record's never cut by the
* ring's end): taken back in order as they are stored. */
unsigned char *ring;
- size_t ring_capacity;
+ size_t ring_capacity, held; /* held: the bytes of the records, together */
uint64_t queued, stored; /* the last queued; saved by the receiver */
uint64_t run, next_pn, in_run;
PureAead *out, *in; /* made once the receiver's key is known */
@@ -461,6 +461,7 @@ bool pure_link_sender_queue(PureLinkSender *l, uint64_t seq, const unsigned char
if (!ok) return false;
*record_at(l, l->count++) = r;
l->queued = seq;
+ l->held += size;
return true;
}
@@ -699,6 +700,7 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui
Record *r = record_at(l, 0);
for (uint32_t i = 0; i < r->parts; ++i) l->received -= part_of(r, i)->state == RECEIVED;
free(r->more); /* its bytes go back with the ring's head */
+ l->held -= r->size;
l->head = (l->head + 1) & (l->capacity - 1);
--l->count;
}
@@ -1190,6 +1192,8 @@ size_t pure_link_sender_datagram(const PureLinkSender *l) { return l->datagram;
uint64_t pure_link_sender_stored(const PureLinkSender *l) { return l->stored; }
+size_t pure_link_sender_held(const PureLinkSender *l) { return l->held; }
+
void pure_link_sender_counts(const PureLinkSender *l, PureLinkCounts *out) {
*out = l->counts;
out->stored = l->stored;
diff --git a/link/link-protocol.h b/link/link-protocol.h
index 96085c6..e4d3080 100644
--- a/link/link-protocol.h
+++ b/link/link-protocol.h
@@ -196,6 +196,8 @@ void pure_link_sender_epochs(PureLinkSender *, uint64_t every_ns, uint64_t bytes
/* The last number queued; the last the receiver stored. */
uint64_t pure_link_sender_queued(const PureLinkSender *);
uint64_t pure_link_sender_stored(const PureLinkSender *);
+/* The bytes of the sends queued and not yet stored. */
+size_t pure_link_sender_held(const PureLinkSender *);
void pure_link_sender_counts(const PureLinkSender *, PureLinkCounts *);
void pure_link_sender_free(PureLinkSender *);
diff --git a/link/link.c b/link/link.c
index 37a5a2a..ad9376c 100644
--- a/link/link.c
+++ b/link/link.c
@@ -210,6 +210,13 @@ uint64_t pure_link_delivered(PureLink *l) {
return stored;
}
+size_t pure_link_held(PureLink *l) {
+ pthread_mutex_lock(&l->lock);
+ size_t held = pure_link_sender_held(l->sender);
+ pthread_mutex_unlock(&l->lock);
+ return held;
+}
+
static void take_answers(PureLink *l, uint64_t now) {
for (unsigned rounds = 0, n = TAKE; rounds < 4 && n == TAKE; ++rounds) {
n = (unsigned)take_datagrams(l->fd, l->taking);
diff --git a/link/link.h b/link/link.h
index 3390a82..64846ce 100644
--- a/link/link.h
+++ b/link/link.h
@@ -22,6 +22,8 @@ bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_
uint64_t pure_link_queued(PureLink *);
/* The last number the receiver saved; any thread. */
uint64_t pure_link_delivered(PureLink *);
+/* The bytes of the sends queued and not yet saved; any thread. */
+size_t pure_link_held(PureLink *);
/* Packets go in bursts with UDP GSO where the system has it (at first);
* off, each in a call of its own: for a path whose queues take a burst as
* one packet (netem does). Before pure_link_start. */
diff --git a/serve/deliver.c b/serve/deliver.c
index 511a2cf..b00984b 100644
--- a/serve/deliver.c
+++ b/serve/deliver.c
@@ -35,6 +35,7 @@ struct PureWire {
pthread_mutex_t lock;
pthread_cond_t changed;
Pending *queue; /* the sends to deliver, in order, from head */
+ size_t held; /* their bytes */
size_t head, count, capacity;
uint64_t queued, delivered, attempts, deliveries;
unsigned batch; /* how many a request may carry, as the receiver said (X-Pure-Batch); 0: one */
@@ -269,6 +270,13 @@ uint64_t pure_wire_delivered(PureWire *w) {
return delivered;
}
+size_t pure_wire_held(PureWire *w) {
+ pthread_mutex_lock(&w->lock);
+ size_t held = w->held;
+ pthread_mutex_unlock(&w->lock);
+ return held;
+}
+
void pure_wire_counts(PureWire *w, uint64_t *delivered, uint64_t *attempts) {
pthread_mutex_lock(&w->lock);
*delivered = w->deliveries;
@@ -312,6 +320,7 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) {
if (sent && payload.size) memcpy(sent, payload.data, payload.size);
w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq, sent, payload.size};
w->queued = seq;
+ w->held += t.size + (sent ? payload.size : 0);
pthread_cond_signal(&w->changed);
pthread_mutex_unlock(&w->lock);
return true;
@@ -717,8 +726,10 @@ static void *deliver(void *context) {
}
if (outcome == DELIVERED) {
for (unsigned i = 0; i < count; ++i) {
- free(w->queue[w->head + i].body);
- free(w->queue[w->head + i].sent);
+ const Pending *p = &w->queue[w->head + i];
+ w->held -= p->size + (p->sent ? p->sent_size : 0);
+ free(p->body);
+ free(p->sent);
}
w->head += count;
w->count -= count;
diff --git a/serve/deliver.h b/serve/deliver.h
index a7ccac3..244363a 100644
--- a/serve/deliver.h
+++ b/serve/deliver.h
@@ -37,6 +37,9 @@ bool pure_wire_queue(PureWire *, uint64_t seq, PureSpan payload);
uint64_t pure_wire_queued(const PureWire *);
/* The last number delivered; any thread. */
uint64_t pure_wire_delivered(PureWire *);
+/* The bytes of the sends queued and not yet delivered (their JSON, and the
+ * payloads kept for answers); any thread. */
+size_t pure_wire_held(PureWire *);
/* Deliver on a thread of the wire's own, calling notify after each delivery. */
bool pure_wire_start(PureWire *, void (*notify)(void *), void *context);
/* Stop (a delivery under way is finished or abandoned) and free. */
diff --git a/serve/serve.c b/serve/serve.c
index e140705..a9dd9d4 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -206,6 +206,11 @@ 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];
+ 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 */
} Executor;
@@ -1509,59 +1514,56 @@ static void publish(Executor *ex) {
if (ex->surface) surface_changed(ex->surface);
}
-/* The saved sends of wired outputs not queued yet go to their wires, in
- * order; a wire that cannot take one takes the rest at a later save. The
- * outbox is looked at newest first, only as far back as the new sends go,
- * however many older ones wait for their receivers. */
-typedef struct { uint32_t wire; uint64_t seq; PureSpan payload; } Fresh;
-typedef struct { Executor *executor; Fresh *fresh; size_t count, capacity; uint64_t done; bool failed; } Queueing;
+/* An output's outlets: its wires first, then its links. Each holds a
+ * window of the output's saved sends, WINDOW_SENDS and --window bytes past
+ * what its receiver has (at least one): the outbox keeps the rest, and the
+ * outlet takes them as its receiver takes what it holds. */
+enum { WINDOW_SENDS = 65536 };
+static size_t window_bytes = 4u << 20;
-/* An output's outlet: its wires first, then its links. */
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];
}
-static uint64_t outlet_queued(Executor *ex, unsigned i) {
- return i < ex->wire_count ? pure_wire_queued(ex->wires[i]) : pure_link_queued(ex->links[i - ex->wire_count]);
-}
-static bool find_fresh(void *context, uint32_t output, uint64_t seq, PureSpan payload) {
- Queueing *q = context;
- Executor *ex = q->executor;
+/* The saved sends not yet queued go to their outlets, in order, as far as
+ * each window goes; an outlet that cannot take one takes it later. Each
+ * outlet looks on from its place in the outbox, moved back by the sends
+ * 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;
+ uint64_t gone = ex->app.sends_gone - ex->gone_seen;
+ ex->gone_seen = ex->app.sends_gone;
for (unsigned i = 0; i < outlets; ++i) {
- if (outlet_output(ex, i) != output || (q->done >> i & 1)) continue;
- if (seq <= outlet_queued(ex, i)) {
- q->done |= UINT64_C(1) << i; /* older ones were queued before */
- break;
- }
- if (q->count == q->capacity) {
- size_t capacity = q->capacity ? 2 * q->capacity : 64;
- Fresh *grown = realloc(q->fresh, capacity * sizeof(*grown));
- if (!grown) { q->failed = true; return false; }
- q->fresh = grown;
- q->capacity = capacity;
+ PureWire *wire = i < ex->wire_count ? ex->wires[i] : NULL;
+ PureLink *link = wire ? NULL : ex->links[i - ex->wire_count];
+ 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];
+ at = at > gone ? at - (size_t)gone : 0;
+ for (; at < ex->app.sent_saved; ++at) {
+ uint32_t of;
+ uint64_t seq;
+ PureSpan payload;
+ if (!pure_app_sent(&ex->app, at, &of, &seq, &payload) || of != output || seq <= queued) continue;
+ if (queued > delivered && (queued - delivered >= WINDOW_SENDS || held + payload.size > window_bytes)) {
+ ++ex->windows_full;
+ break;
+ }
+ if (wire ? !pure_wire_queue(wire, seq, payload) : !pure_link_queue(link, seq, payload.data, payload.size))
+ break;
+ queued = seq;
+ held += payload.size;
}
- q->fresh[q->count++] = (Fresh){i, seq, payload};
- break;
+ ex->outlet_from[i] = at;
}
- return q->done != (outlets >= 64 ? UINT64_MAX : (UINT64_C(1) << outlets) - 1);
}
static void after_save(Executor *ex) {
publish(ex);
checkpoint_round(ex);
- if (!ex->wire_count && !ex->link_count) return;
- Queueing q = {.executor = ex};
- pure_app_outbox_newest(&ex->app, find_fresh, &q);
- bool blocked[2 * PURE_MODULE_INPUTS] = {false};
- for (size_t k = q.failed ? 0 : q.count; k-- > 0;) {
- const Fresh *f = &q.fresh[k];
- if (blocked[f->wire]) continue;
- blocked[f->wire] = f->wire < ex->wire_count ? !pure_wire_queue(ex->wires[f->wire], f->seq, f->payload)
- : !pure_link_queue(ex->links[f->wire - ex->wire_count], f->seq,
- f->payload.data, f->payload.size);
- }
- free(q.fresh);
+ feed_outlets(ex);
}
enum { ANSWER_WAIT_MS = 100 };
@@ -1607,6 +1609,7 @@ static bool take_deliveries(Executor *ex) {
pure_app_trim(&ex->app, ex->link_output[i], stored);
ex->link_trimmed[i] = stored;
}
+ feed_outlets(ex); /* their windows moved on */
return answered;
}
@@ -2630,6 +2633,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], "--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;
else if (!strcmp(argv[i], "--threads") && number(argv[i + 1], 256, &value) && value) threads = (unsigned)value;
@@ -2751,7 +2755,8 @@ 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, memory_refused = owner.executor.memory_refused;
+ ticks = owner.executor.ticks, shown = 0, checkpoints = 0, memory_refused = owner.executor.memory_refused,
+ windows_full = owner.executor.windows_full;
for (unsigned i = 0; i < threads; ++i) {
shown += hosts[i].shown;
active += hosts[i].active;
@@ -2801,12 +2806,13 @@ int main(int argc, char **argv) {
",\"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
- ",\"state_bytes\":%zu,\"memory_refused\":%" PRIu64 "}\n",
+ ",\"state_bytes\":%zu,\"memory_refused\":%" PRIu64 ",\"windows_full\":%" 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, state_bytes, memory_refused);
+ heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused, state_bytes, memory_refused,
+ windows_full);
free(hosts);
free(thread);
return 0;
@@ -2815,12 +2821,14 @@ usage:
"[--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] "
- "[--surface LIBRARY [-- ARGUMENTS]]\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"
" --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"
" --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);
+ " past it is refused, 503 memory (docs/design/013-memory-budget.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 379982d..ea4f9f6 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -1213,6 +1213,110 @@ def server_memory(aup, pure, work):
return {'sends_held': accepted, 'receiver_refused': stats['b']['memory_refused'],
'link_resent': stats['a']['link_resent']}
+
+WINDOW_SENDER = """state n = 0
+input buy = message {
+ item: text
+ count: u32
+}
+output ordered = message {
+ item: text
+ count: u32
+ to: {
+ city: text
+ }
+}
+output copied = message {
+ item: text
+ count: u32
+ to: {
+ city: text
+ }
+}
+on buy {
+ n = n + 1
+ send ordered { item: buy.item, count: buy.count, to: { city: "Seoul" } }
+ send copied { item: buy.item, count: buy.count, to: { city: "Busan" } }
+}
+view sent = n
+"""
+
+
+def server_windows(aup, pure, work):
+ """Each --send holds a window of its sends (--window bytes past what its
+ receiver has), not the whole outbox (docs/design/013-memory-budget.md,
+ step 2). A sender's two outputs, one linked (pure://) and one over HTTP,
+ both receivers down: 300 buys wait, each outlet holding some 2,000
+ bytes of them. When the receivers come up, the outlets take the rest
+ from the outbox as their receivers store what they hold, and each
+ receiver applies all 300, once each."""
+ import select
+ import socket
+ import time
+ sender_module, receiver_module = work / 'window-sender.pmod', work / 'window-receiver.pmod'
+ for module, text in ((sender_module, WINDOW_SENDER), (receiver_module, LINK_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])
+
+ keys = {}
+ for name in ('sender', 'receiver'):
+ process, info = start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'window-key',
+ '--key', work / f'window-{name}.key'], 0)
+ keys[name] = info['public_key']
+ stop(process)
+ port_a, port_b, port_c, link_port = free_port(), free_port(), free_port(), free_port(socket.SOCK_DGRAM)
+ sender = start([pure / 'build/pure-serve', '--module', sender_module, '--store', work / 'window-a',
+ '--key', work / 'window-sender.key', '--window', 2000,
+ '--send', f'ordered=pure://127.0.0.1:{link_port}/order', '--seal', f'ordered={keys["receiver"]}',
+ '--send', f'copied=http://127.0.0.1:{port_c}/inputs/order'], port_a)[0]
+ receivers, stats, total = [], {}, 0
+ try:
+ for i in range(300):
+ count = i % 50 + 1
+ assert post(port_a, '/inputs/buy', {'item': f'i{i:03}', 'count': count})[0] == 200
+ total += count
+ time.sleep(0.5)
+ receivers.append(start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'window-b',
+ '--key', work / 'window-receiver.key', '--link-port', link_port,
+ '--sender', f'window-sender.ordered={keys["sender"]}'], port_b)[0])
+ receivers.append(start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'window-c'],
+ port_c)[0])
+ deadline = time.time() + 60
+ while (number(port_b, 'count') < 300 or number(port_c, 'count') < 300) and time.time() < deadline:
+ time.sleep(0.05)
+ for port in (port_b, port_c):
+ assert number(port, 'count') == 300 and number(port, 'total') == total, (port, number(port, 'count'))
+ finally:
+ for name, process in zip(('b', 'c'), receivers):
+ stats[name] = stop(process)
+ stats['a'] = stop(sender)
+ assert stats['a']['windows_full'] >= 1, stats['a']
+ return {'windows_full': stats['a']['windows_full'], 'wire_deliveries': stats['a']['delivered'],
+ 'link_records': stats['a']['link_records']}
+
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
@@ -2918,6 +3022,7 @@ def main():
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_windows'] = server_windows(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)
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 13:53:00 +0000