From 393804bee74311ef00e39502d7979ca00bd38cb2 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Tue, 29 Sep 2026 16:42:07 +0100 Subject: Give each of a peer's runs a place, however many send at once A receiver kept two runs a peer: the sender's, and the one before it restarted. A node with three outputs to another sends in three runs with one key, and they pushed one another out: each packet keyed its run again, and an answer marked for one went to the run that took its place. Now more places are made beside when the run seen longest ago is still live (up to 64), and one is reused only once its run is quiet the idle period. test-link's three outputs of one node key three runs, not seven, and are all stored in 16 ms, not 1016. --- link/link-protocol.c | 44 ++++++++++++++++++------- link/test-link.c | 90 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 122 insertions(+), 12 deletions(-) (limited to 'link') diff --git a/link/link-protocol.c b/link/link-protocol.c index 62def6f..6120deb 100644 --- a/link/link-protocol.c +++ b/link/link-protocol.c @@ -26,7 +26,11 @@ enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PIN * (a sender may offer an epoch again with another key). */ enum { EPOCH_MASK = 15, EPOCH_FRAME = 1 + 4 + PURE_SEAL_KEY, ANSWERS = 4, ACCEPT_FRAME = EPOCH_FRAME + ANSWERS }; enum { RECORDS_HEAD = 10 }; /* type, the first number, the count */ -enum { RANGES = 32, RUNS = 2, UNITS = 64, CREDIT = 1024, BATCH = 256, RECORD_HEAD = 19 }; +enum { RANGES = 32, UNITS = 64, CREDIT = 1024, BATCH = 256, RECORD_HEAD = 19 }; +/* A receiver keeps a peer's runs in RUNS places (its sender's, and the one + * before a restart), and more beside when more of them send at once (a + * node's outputs to this one, each a sender of its own), RUNS_MOST in all. */ +enum { RUNS = 2, RUNS_MOST = 64 }; enum { WINDOW_FIRST = 10 * DATAGRAM, WINDOW_LEAST = 2 * DATAGRAM }; /* Reordering awaited (see PureLinkSender): at first and at most; congestion * events with no late packet before the patience halves. */ @@ -1238,7 +1242,8 @@ struct Active { uint64_t seen; uint64_t accepted_at; /* its last new epoch: one a second, whatever its runs */ size_t place; /* among the active */ - Run runs[RUNS]; + Run runs[RUNS], **more; /* its runs: more_count more beside, each made once and kept */ + unsigned more_count; Stream **streams; unsigned stream_count, stream_capacity; }; @@ -1606,6 +1611,11 @@ static void let_go(PureLinkReceiver *s, Active *a) { free(a->streams[k]); } free(a->streams); + for (unsigned i = 0; i < a->more_count; ++i) { + explicit_bzero(a->more[i], sizeof(*a->more[i])); + free(a->more[i]); + } + free(a->more); s->peers[a->peer].active = NULL; s->actives[a->place] = s->actives[--s->active_count]; s->actives[a->place]->place = a->place; @@ -1613,14 +1623,16 @@ static void let_go(PureLinkReceiver *s, Active *a) { free(a); } +static Run *run_at(const Active *a, unsigned i) { return i < RUNS ? (Run *)&a->runs[i] : a->more[i - RUNS]; } + /* Whether any of a peer's records or answers waits here. */ static bool busy(const Active *a) { for (unsigned k = 0; k < a->stream_count; ++k) if (a->streams[k]->out || a->streams[k]->held_count || a->streams[k]->line_count || a->streams[k]->tail || a->streams[k]->ready) return true; - for (unsigned i = 0; i < RUNS; ++i) - if (a->runs[i].due) return true; + for (unsigned i = 0; i < RUNS + a->more_count; ++i) + if (run_at(a, i)->due) return true; return false; } @@ -1661,19 +1673,27 @@ static void mark_due(PureLinkReceiver *s, Run *r) { } static Run *run_found(Active *a, uint64_t run) { - for (unsigned i = 0; i < RUNS; ++i) - if (a->runs[i].used && a->runs[i].run == run) return &a->runs[i]; + for (unsigned i = 0; i < RUNS + a->more_count; ++i) + if (run_at(a, i)->used && run_at(a, i)->run == run) return run_at(a, i); return NULL; } -/* A new run's keys, in the slot unused longest: only once its first packet - * opened with them, so a forged one displaces nothing. */ -static Run *install_run(Active *a, const Run *made) { +/* A new run's keys, in the place of the run heard from longest ago when that + * one is quiet the idle period (a sender that started again), else in a + * place more (up to RUNS_MOST, then that one's all the same): only once its + * first packet opened with them, so a forged one displaces nothing. A place + * is kept while the peer is active, so what points to it stays good. */ +static Run *install_run(const PureLinkReceiver *s, Active *a, const Run *made, uint64_t now) { Run *slot = &a->runs[0]; - for (unsigned i = 0; i < RUNS; ++i) { - Run *r = &a->runs[i]; + for (unsigned i = 0; i < RUNS + a->more_count; ++i) { + Run *r = run_at(a, i); if (!r->used || (slot->used && r->seen < slot->seen)) slot = r; } + if (slot->used && now - slot->seen < s->idle && RUNS + a->more_count < RUNS_MOST) { + Run **more = realloc(a->more, (a->more_count + 1) * sizeof(*more)), *r = more ? calloc(1, sizeof(*r)) : NULL; + if (more) a->more = more; + if (r) a->more[a->more_count++] = slot = r; + } bool due = slot->due; /* the answer list may hold the slot */ *slot = *made; slot->due = due; @@ -2132,7 +2152,7 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack } if (woke) ++s->counts.woken, s->counts.remembered += a->remembered; /* counted once it opens */ bool fresh = !r; - if (!r) r = install_run(a, &made); + if (!r) r = install_run(s, a, &made, now); explicit_bzero(&made, sizeof(made)); if (uses_next) confirm(s, peer, r); if (!fresh && had(r, pn)) { /* again (opened first: nothing unsealed answers): its acknowledgment was lost, perhaps */ diff --git a/link/test-link.c b/link/test-link.c index 9b709ce..7024260 100644 --- a/link/test-link.c +++ b/link/test-link.c @@ -652,6 +652,95 @@ static void flood(const char *dir) { free(w.packets); } +/* Three outputs of one node to this one: three senders with one key, each a + * run of its own, sending at once. Each run keeps a place of its own (none + * pushes another out, which sent its answers to the wrong run), each keyed + * once, and every stream's records are delivered once, in order. */ +typedef struct { uint64_t now, heard[3]; PureLinkSender *senders[3]; PureLinkDone done[64]; size_t done_count; } Outputs; +static int output_input(void *context, const char *name, uint32_t shape) { + (void)context; + return shape != 0x5eed ? -1 : !strcmp(name, "one") ? 0 : !strcmp(name, "two") ? 1 : !strcmp(name, "three") ? 2 : -1; +} +static bool output_deliver(void *context, const PureLinkDelivery *d) { + Outputs *o = context; + if (o->done_count == sizeof(o->done) / sizeof(o->done[0])) return false; + uint64_t heard = o->heard[d->input]; + bool gap = false; + const unsigned char *at = d->records; + for (unsigned k = 0; k < d->count; ++k) { + uint32_t size = (uint32_t)at[0] | (uint32_t)at[1] << 8 | (uint32_t)at[2] << 16 | (uint32_t)at[3] << 24; + uint64_t seq = d->first + k; + if (seq > heard + 1) { + gap = true; + break; + } + if (seq == heard + 1) { + bool same = size == record_size(seq); + for (size_t i = 0; same && i < size; ++i) same = at[4 + i] == record_byte(seq, i); + CHECK(same, "outputs: record %llu of input %u is not as it was sent", (unsigned long long)seq, d->input); + heard = seq; + } + at += 4 + size; + } + o->heard[d->input] = heard; + o->done[o->done_count++] = (PureLinkDone){d->token, heard, true, gap}; + return true; +} +static bool output_answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to, + size_t to_size) { + (void)to, (void)to_size; + Outputs *o = context; + for (unsigned i = 0; i < 3; ++i) pure_link_sender_take(o->senders[i], packet, size, o->now); /* its run's opens */ + return true; +} +static void outputs(const char *dir) { + unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY], out[PURE_LINK_DATAGRAM_MOST]; + char error[256]; + static unsigned char bytes[20200]; + PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b); + Outputs o = {0}; + PureLinkReceiver *r = pure_link_receiver(b, public_b, &(PureLinkHost){output_input, output_deliver, &o}, error, + sizeof(error)); + if (!r || !pure_link_allow(r, "test.ordered", public_a, error, sizeof(error))) exit(1); + const char *inputs[3] = {"one", "two", "three"}; + for (unsigned i = 0; i < 3; ++i) { + PureLinkSender *s = o.senders[i] = pure_link_sender("test.ordered", inputs[i], 0x5eed, a, public_a); + if (!s || !pure_link_sender_peer(s, public_b)) exit(1); + for (uint64_t seq = 1; seq <= 1000; ++seq) { + size_t size = record_size(seq); + for (size_t k = 0; k < size; ++k) bytes[k] = record_byte(seq, k); + if (!pure_link_sender_queue(s, seq, bytes, size)) exit(1); + } + } + uint64_t number; + bool all = false; + for (; o.now < 60000 * MS && !all; o.now += MS) { + for (unsigned i = 0; i < 3; ++i) + for (size_t size; (size = pure_link_sender_packet(o.senders[i], out, false, o.now, &number));) + pure_link_receiver_take(r, out, size, "sender0", 7, o.now); + PureLinkDone told[64]; + size_t count = o.done_count; + memcpy(told, o.done, count * sizeof(*told)); + o.done_count = 0; + pure_link_receiver_turn(r, told, count, o.now, output_answer, &o); + all = true; + for (unsigned i = 0; i < 3; ++i) { + pure_link_sender_tick(o.senders[i], o.now); + all = all && pure_link_sender_stored(o.senders[i]) == 1000; + } + } + PureLinkCounts heard; + pure_link_receiver_counts(r, &heard); + CHECK(all && o.heard[0] == 1000 && o.heard[1] == 1000 && o.heard[2] == 1000 && heard.keyed == 3, + "outputs: heard %llu, %llu, %llu of 1000 each, %llu runs keyed (3 runs)", (unsigned long long)o.heard[0], + (unsigned long long)o.heard[1], (unsigned long long)o.heard[2], (unsigned long long)heard.keyed); + printf("{\"case\":\"outputs\",\"runs\":3,\"records\":3000,\"virtual_ms\":%llu,\"keyed\":%llu,\"packets\":%llu}\n", + (unsigned long long)(o.now / MS), (unsigned long long)heard.keyed, (unsigned long long)heard.received); + for (unsigned i = 0; i < 3; ++i) pure_link_sender_free(o.senders[i]); + pure_link_receiver_free(r); + pure_seal_key_free(a), pure_seal_key_free(b); +} + /* The largest send (PURE_LINK_RECORD_MOST) goes whole, over a receiver that * never heard of the stream before; one byte more is not taken to send. */ typedef struct { World *w; PureLinkSender *s; } Largest; @@ -768,6 +857,7 @@ int main(void) { o.hole ? "opened" : "never opened", o.datagram); impostors(dir); flood(dir); + outputs(dir); largest(dir); char command[600]; snprintf(command, sizeof(command), "rm -rf %s", dir); -- cgit v1.2.3