aboutsummaryrefslogtreecommitdiffstats
path: root/link
diff options
context:
space:
mode:
Diffstat (limited to 'link')
-rw-r--r--link/link-protocol.c44
-rw-r--r--link/test-link.c90
2 files changed, 122 insertions, 12 deletions
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);
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 12:30:32 +0000