aboutsummaryrefslogtreecommitdiffstats
path: root/link
diff options
context:
space:
mode:
Diffstat (limited to 'link')
-rw-r--r--link/link-protocol.c72
-rw-r--r--link/link-protocol.h4
-rw-r--r--link/test-link.c122
3 files changed, 147 insertions, 51 deletions
diff --git a/link/link-protocol.c b/link/link-protocol.c
index 1106ac5..318441e 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -1544,23 +1544,28 @@ static void drop_first(Stream *st) {
if (!--st->held_count) st->held_first = 0;
}
-/* The memo of secrets (pure_link_remember): a header, then a place a peer
- * in the order allowed, each the first 28 bytes of its key, a check (FNV-1a
- * of its whole key and the secret) and the secret. A place whose check
- * fails (never made, another key, cut short by a crash) is made again.
- * From MEMO_EPOCH on, the epoch last accepted for the peer's run: the run,
- * the epoch, a check (FNV-1a of the rest), its secret, the one before, the
- * key the sender offered and ours; written secrets first, the run last. A
- * node that starts again goes on in the sender's epoch, no key made. */
+/* The memo of secrets (pure_link_remember): a header, MEMO_MORE places
+ * beside (below), then a place a peer in the order allowed, each the first
+ * 28 bytes of its key, a check (FNV-1a of its whole key and the secret) and
+ * the secret. A place whose check fails (never made, another key, cut short
+ * by a crash) is made again. From MEMO_EPOCH on, the epoch last accepted
+ * for the peer's run: the run, the epoch, a check (FNV-1a of the rest), its
+ * secret, the one before, the key the sender offered and ours; written
+ * secrets first, the run last. A node that starts again goes on in the
+ * sender's epoch, no key made. A peer's other runs that send while the one
+ * its place has does (a node's outputs to this one) keep theirs beside:
+ * the peer (+ 1), then an epoch as a place has it, in one of two places
+ * its peer and run find. */
enum { MEMO_HEAD = 64, MEMO_PLACE = 256, MEMO_EPOCH = 64, MEMO_EPOCH_SIZE = 144 };
-static const unsigned char memo_magic[8] = {'P', 'L', 'M', '2', 0, 0, 0, 0};
+enum { MEMO_MORE = 1024, MEMO_MORE_PLACE = 16 + MEMO_EPOCH_SIZE, MEMO_PLACES = MEMO_HEAD + MEMO_MORE * MEMO_MORE_PLACE };
+static const unsigned char memo_magic[8] = {'P', 'L', 'M', '3', 0, 0, 0, 0};
static uint32_t memo_check(const unsigned char key[PURE_SEAL_KEY], const unsigned char secret[32]) {
return fnv(fnv(2166136261u, key, PURE_SEAL_KEY), secret, 32);
}
static unsigned char *memo_place(const PureLinkReceiver *s, uint32_t peer) {
- size_t at = MEMO_HEAD + (size_t)peer * MEMO_PLACE;
+ size_t at = MEMO_PLACES + (size_t)peer * MEMO_PLACE;
return s->memo && at + MEMO_PLACE <= s->memo_size ? s->memo + at : NULL;
}
@@ -1568,18 +1573,53 @@ static uint32_t epoch_check(const unsigned char *e) {
return fnv(fnv(2166136261u, e, 12), e + 16, MEMO_EPOCH_SIZE - 16);
}
-/* The epoch the memo has for the peer's run (its place's epoch part), or NULL. */
+static bool epoch_of(const unsigned char *e, uint64_t run) { return get64(e) == run && get32(e + 12) == epoch_check(e); }
+
+/* The two places beside where a peer's run may be kept. */
+static void memo_beside(const PureLinkReceiver *s, uint32_t peer, uint64_t run, unsigned char *ways[2]) {
+ unsigned char key[12];
+ put32(key, peer), put64(key + 4, run);
+ size_t k = fnv(2166136261u, key, sizeof(key)) & (MEMO_MORE - 1);
+ ways[0] = s->memo + MEMO_HEAD + k * MEMO_MORE_PLACE;
+ ways[1] = s->memo + MEMO_HEAD + (k ^ 1) * MEMO_MORE_PLACE;
+}
+
+/* The epoch the memo has for the peer's run (its place's epoch part, or
+ * beside), or NULL. */
static const unsigned char *memo_epoch(const PureLinkReceiver *s, uint32_t peer, uint64_t run) {
const unsigned char *place = memo_place(s, peer), *e = place ? place + MEMO_EPOCH : NULL;
- return e && get64(e) == run && get32(e + 12) == epoch_check(e) ? e : NULL;
+ if (!e || epoch_of(e, run)) return e;
+ unsigned char *ways[2];
+ memo_beside(s, peer, run, ways);
+ for (unsigned w = 0; w < 2; ++w)
+ if (get32(ways[w]) == peer + 1 && epoch_of(ways[w] + 16, run)) return ways[w] + 16;
+ return NULL;
}
-static void remember_epoch(PureLinkReceiver *s, uint32_t peer, const Run *r) {
- unsigned char *place = memo_place(s, peer);
+/* The epoch accepted for r, in the peer's place, or beside when the run
+ * the place has still sends: in the way that has r, else an empty one,
+ * else the first. What a run had beside goes once its place has it (an
+ * epoch's secrets are kept only until the next one's are). */
+static Run *run_found(Active *a, uint64_t run);
+static void remember_epoch(PureLinkReceiver *s, uint32_t peer, const Run *r, uint64_t now) {
+ unsigned char *place = memo_place(s, peer), *ways[2], *beside = NULL;
if (!place) return;
unsigned char *e = place + MEMO_EPOCH;
+ Active *a = s->peers[peer].active;
+ const Run *other = a && get64(e) != r->run && epoch_of(e, get64(e)) ? run_found(a, get64(e)) : NULL;
+ memo_beside(s, peer, r->run, ways);
+ if (other && now - other->seen < s->idle) {
+ for (unsigned w = 0; w < 2 && !beside; ++w)
+ if (get32(ways[w]) == peer + 1 && get64(ways[w] + 16) == r->run) beside = ways[w];
+ for (unsigned w = 0; w < 2 && !beside; ++w)
+ if (!epoch_of(ways[w] + 16, get64(ways[w] + 16))) beside = ways[w];
+ e = (beside = beside ? beside : ways[0]) + 16;
+ } else
+ for (unsigned w = 0; w < 2; ++w)
+ if (get32(ways[w]) == peer + 1 && get64(ways[w] + 16) == r->run) explicit_bzero(ways[w], MEMO_MORE_PLACE);
memset(e, 0, 16);
__atomic_thread_fence(__ATOMIC_RELEASE);
+ if (beside) put32(beside, peer + 1);
memcpy(e + 16, r->next.secret, 32);
memcpy(e + 48, r->now.secret, 32);
memcpy(e + 80, r->offered, PURE_SEAL_KEY);
@@ -1594,7 +1634,7 @@ bool pure_link_remember(PureLinkReceiver *s, const char *path, char *error, size
int fd = open(path, O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600);
struct stat info;
unsigned char head[MEMO_HEAD];
- size_t needed = MEMO_HEAD + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE;
+ size_t needed = MEMO_PLACES + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE;
bool ok = fd >= 0 && !fstat(fd, &info) && S_ISREG(info.st_mode) && !(info.st_mode & 077);
/* Another node's key, or none yet: begun again. */
bool ours = ok && pread(fd, head, MEMO_HEAD, 0) == MEMO_HEAD && !memcmp(head, memo_magic, 8) &&
@@ -2115,7 +2155,7 @@ static void accept_waiting(PureLinkReceiver *s, uint64_t now) {
if (base == a->secret && r->now.n && epoch_made(s, peer, r, &zero, 0, a->secret)) /* begun again from 0: so are its answers */
epoch_now(s, peer, r, &zero);
explicit_bzero(&zero, sizeof(zero));
- remember_epoch(s, a->peer, r);
+ remember_epoch(s, a->peer, r, now);
mark_due(s, r);
}
s->offer_count = 0;
diff --git a/link/link-protocol.h b/link/link-protocol.h
index df15246..f131e99 100644
--- a/link/link-protocol.h
+++ b/link/link-protocol.h
@@ -244,7 +244,9 @@ void pure_link_idle(PureLinkReceiver *, uint64_t ms);
* (path, 0600; begun again for another node's key): a place a sender in
* the order allowed, mapped, so a sender at rest takes no memory for it.
* A node that starts again then makes no X25519 for a sender it heard from
- * before. After the senders are allowed, before the first packet. */
+ * before, and goes on in the epoch of each of its runs (a key's runs that
+ * send beside its first, as a node's outputs to this one do, in 1,024
+ * places beside). After the senders are allowed, before the first packet. */
bool pure_link_remember(PureLinkReceiver *, const char *path, char *error, size_t size);
/* A packet that came from an address (from_size bytes, at most
* PURE_LINK_ADDRESS kept): answers to its sender go there. */
diff --git a/link/test-link.c b/link/test-link.c
index f609fe5..7655025 100644
--- a/link/test-link.c
+++ b/link/test-link.c
@@ -657,8 +657,15 @@ static void flood(const char *dir) {
/* 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;
+ * once, and every stream's records are delivered once, in order. Then,
+ * each in an epoch past 0, the receiver starts again (kill -9) with its
+ * memo: every run goes on in its epoch, none told LOST, no X25519 made. */
+typedef struct {
+ uint64_t now, heard[3], queued;
+ 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;
@@ -695,49 +702,96 @@ static bool output_answer(void *context, const unsigned char *packet, size_t siz
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];
+static PureLinkReceiver *outputs_receiver(Outputs *o, const PureSealKey *b, const unsigned char *public_b,
+ const unsigned char *public_a, const char *memo) {
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,
+ 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) {
+ if (!r || !pure_link_allow(r, "test.ordered", public_a, error, sizeof(error)) ||
+ !pure_link_remember(r, memo, error, sizeof(error))) {
+ fprintf(stderr, "test-link: %s\n", error);
+ exit(1);
+ }
+ return r;
+}
+/* Each sender's records to last queued. */
+static void outputs_queue(Outputs *o, uint64_t last) {
+ static unsigned char bytes[20200];
+ for (unsigned i = 0; i < 3; ++i)
+ for (uint64_t seq = o->queued + 1; seq <= last; ++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);
+ if (!pure_link_sender_queue(o->senders[i], seq, bytes, size)) exit(1);
}
- }
- uint64_t number;
- bool all = false;
- for (; o.now < 60000 * MS && !all; o.now += MS) {
+ o->queued = last;
+}
+/* Until each sender's records are stored, or a minute; when epochs, a record
+ * more each 50 ms until each is in an epoch past 0 (an offer goes in a
+ * packet of records). */
+static void outputs_until(Outputs *o, PureLinkReceiver *r, bool epochs) {
+ unsigned char out[PURE_LINK_DATAGRAM_MOST];
+ uint64_t number, end = o->now + 60000 * MS;
+ for (bool all = false; o->now < end && !all; o->now += MS) {
+ if (epochs && o->now % (50 * MS) == 0) outputs_queue(o, o->queued + 1);
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);
+ 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);
+ 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 c;
+ pure_link_sender_tick(o->senders[i], o->now);
+ pure_link_sender_counts(o->senders[i], &c);
+ all = all && pure_link_sender_stored(o->senders[i]) == o->queued && (!epochs || c.epochs >= 1);
}
}
- 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);
+}
+static void outputs(const char *dir) {
+ unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY];
+ char memo[512];
+ PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b);
+ snprintf(memo, sizeof(memo), "%s/b.key.links", dir);
+ unlink(memo);
+ Outputs o = {0};
+ PureLinkReceiver *r = outputs_receiver(&o, b, public_b, public_a, memo);
+ const char *inputs[3] = {"one", "two", "three"};
+ for (unsigned i = 0; i < 3; ++i)
+ if (!(o.senders[i] = pure_link_sender("test.ordered", inputs[i], 0x5eed, a, public_a)) ||
+ !pure_link_sender_peer(o.senders[i], public_b))
+ exit(1);
+ outputs_queue(&o, 500);
+ outputs_until(&o, r, false);
+ PureLinkCounts first;
+ pure_link_receiver_counts(r, &first);
+ uint64_t took = o.now / MS;
+ outputs_until(&o, r, true); /* a new epoch a second for a peer: the third's after two */
+ pure_link_receiver_free(r); /* kill -9: what it held is gone, the memo stays */
+ o.done_count = 0;
+ r = outputs_receiver(&o, b, public_b, public_a, memo);
+ outputs_queue(&o, 1000);
+ outputs_until(&o, r, false);
+ PureLinkCounts again, sent = {0};
+ pure_link_receiver_counts(r, &again);
+ for (unsigned i = 0; i < 3; ++i) {
+ PureLinkCounts c;
+ pure_link_sender_counts(o.senders[i], &c);
+ sent.fallbacks += c.fallbacks;
+ CHECK(pure_link_sender_stored(o.senders[i]) == 1000 && o.heard[i] == 1000, "outputs: %u stored %llu, heard %llu",
+ i, (unsigned long long)pure_link_sender_stored(o.senders[i]), (unsigned long long)o.heard[i]);
+ }
+ CHECK(first.keyed == 3 && took < 1000, "outputs: %llu runs keyed (3 runs), stored in %llu ms",
+ (unsigned long long)first.keyed, (unsigned long long)took);
+ CHECK(!sent.fallbacks && !again.secrets && again.keyed == 3,
+ "outputs, started again: %llu epochs taken for lost, %llu secrets made, %llu runs keyed",
+ (unsigned long long)sent.fallbacks, (unsigned long long)again.secrets, (unsigned long long)again.keyed);
+ printf("{\"case\":\"outputs\",\"runs\":3,\"records\":3000,\"stored_ms\":%llu,\"keyed\":%llu,"
+ "\"keyed_again\":%llu,\"secrets_again\":%llu,\"fallbacks\":%llu}\n",
+ (unsigned long long)took, (unsigned long long)first.keyed, (unsigned long long)again.keyed,
+ (unsigned long long)again.secrets, (unsigned long long)sent.fallbacks);
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);
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 14:16:43 +0000