From 56d6876f906f91db9f48f0f27b7f456ee601efd9 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Tue, 29 Sep 2026 15:26:30 +0100 Subject: Cut what a small record costs a link: no division a record, no walk a delivery Profiling both sides in one thread, without sockets (61-byte records): the rings of records and flights divided for each place, the record a number names was found three times over, a record in one part was searched for its part, and the receiver walked the line record by record to hand it over and again to let it go. The rings are powers of two long and now masked; a record is found once; a one-part record's part is the first; a line of a batch or less goes whole without a walk, and a delivery applied whole lets its bytes go at once. The protocol alone: 0.171 to 0.162 us a record. M5 (the link and listener threads, sockets too), five rounds against the commit before: 61 B 0.298 to 0.287 us, 1 KiB 2.63-2.90 to 2.48-2.79 us, 16 KiB the same or less; quinn measured 0.31, 3.85 and 60 us in the same rounds. test-link's results are the same to the byte. --- link/link-protocol.c | 54 ++++++++++++++++++++++++++++++++++------------------ link/link-protocol.h | 2 +- 2 files changed, 36 insertions(+), 20 deletions(-) (limited to 'link') diff --git a/link/link-protocol.c b/link/link-protocol.c index fe62b5f..e4661cb 100644 --- a/link/link-protocol.c +++ b/link/link-protocol.c @@ -123,7 +123,8 @@ static bool derive(const unsigned char secret[32], const unsigned char from[PURE /* The secret of epoch n: the X25519 of the two keys made for it (fresh), * with the secret before it (epoch 0's is the nodes' static secret). */ -static bool epoch_secret_of(const unsigned char before[32], const unsigned char fresh[32], uint32_t n, unsigned char out[32]) { +static bool epoch_secret_of(const unsigned char before[32], const unsigned char fresh[32], uint32_t n, + unsigned char out[32]) { unsigned char info[17 + 4 + 32]; memcpy(info, "pure link 0 epoch", 17); put32(info + 17, n); @@ -297,12 +298,14 @@ static bool seal_in(PureLinkSender *l, uint32_t n, const unsigned char secret[32 static bool offer(PureLinkSender *l, uint32_t n); static void begin_epoch(PureLinkSender *l, const unsigned char their[PURE_SEAL_KEY], uint64_t now); -static Record *record_at(PureLinkSender *l, size_t k) { return &l->records[(l->head + k) % l->capacity]; } +/* The rings of records and flights are powers of two long (grown by + * doubling): a place is masked, not divided. */ +static Record *record_at(PureLinkSender *l, size_t k) { return &l->records[(l->head + k) & (l->capacity - 1)]; } /* The queued record numbered seq, or NULL (stored already, or not queued). */ static Record *record_of(PureLinkSender *l, uint64_t seq) { - if (!l->count || seq < record_at(l, 0)->seq || seq - record_at(l, 0)->seq >= l->count) return NULL; - return record_at(l, (size_t)(seq - record_at(l, 0)->seq)); + uint64_t first = l->count ? l->records[l->head].seq : 0; + return l->count && seq >= first && seq - first < l->count ? record_at(l, (size_t)(seq - first)) : NULL; } /* Every change of a part's state goes through here: the link knows at once @@ -355,7 +358,7 @@ static uint32_t part_at(Record *r, uint32_t offset) { static unsigned carried(PureLinkSender *l, const Unit *u, uint64_t pn, bool acknowledged, uint64_t now) { Record *r = record_of(l, u->seq); unsigned unsent = 0; - for (uint32_t i = r ? part_at(r, u->offset) : 0; r && i < r->parts; ++i) { + for (uint32_t i = r && r->more ? part_at(r, u->offset) : 0; r && i < r->parts; ++i) { Part *p = part_of(r, i); if (p->offset >= u->offset + u->length && u->length) break; if (acknowledged && p->state != RECEIVED) { @@ -434,7 +437,7 @@ static Flight *flight_of(PureLinkSender *l, uint64_t pn) { if (!l->flight_count) return NULL; uint64_t first = l->flights[l->flight_head].pn; if (pn < first || pn - first >= l->flight_count) return NULL; - return &l->flights[(l->flight_head + (size_t)(pn - first)) % l->flight_capacity]; + return &l->flights[(l->flight_head + (size_t)(pn - first)) & (l->flight_capacity - 1)]; } /* Room for one more flight: made before a packet is, so every number sent @@ -444,7 +447,7 @@ static bool flight_room(PureLinkSender *l) { size_t capacity = l->flight_capacity ? 2 * l->flight_capacity : 8; Flight *grown = malloc(capacity * sizeof(*grown)); if (!grown) return false; - for (size_t k = 0; k < l->flight_count; ++k) grown[k] = l->flights[(l->flight_head + k) % l->flight_capacity]; + for (size_t k = 0; k < l->flight_count; ++k) grown[k] = l->flights[(l->flight_head + k) & (l->flight_capacity - 1)]; free(l->flights); l->flights = grown; l->flight_capacity = capacity; @@ -453,7 +456,7 @@ static bool flight_room(PureLinkSender *l) { } static void push_flight(PureLinkSender *l, const Flight *f) { - l->flights[(l->flight_head + l->flight_count++) % l->flight_capacity] = *f; + l->flights[(l->flight_head + l->flight_count++) & (l->flight_capacity - 1)] = *f; } static void settle(PureLinkSender *l, Flight *f) { @@ -540,7 +543,7 @@ static void detect_lost(PureLinkSender *l, uint64_t now) { uint64_t rtt = most(l->latest_rtt, l->sampled ? l->smoothed_rtt : FIRST_RTT); uint64_t delay = most(rtt + reorder_time(l, rtt), GRANULARITY); for (size_t k = 0; k < l->flight_count; ++k) { - Flight *f = &l->flights[(l->flight_head + k) % l->flight_capacity]; + Flight *f = &l->flights[(l->flight_head + k) & (l->flight_capacity - 1)]; if (f->pn >= l->largest_acked) break; if (f->done) continue; bool late = f->sent + delay <= now; @@ -632,7 +635,7 @@ static void on_ack(PureLinkSender *l, const unsigned char *p, size_t size, uint6 Flight *f = &l->flights[l->flight_head]; if (!f->done || (f->lost && f->sent + keep > now)) break; if (f->lost) settle_loss(l, f, true); /* not late: lost */ - l->flight_head = (l->flight_head + 1) % l->flight_capacity; + l->flight_head = (l->flight_head + 1) & (l->flight_capacity - 1); --l->flight_count; } } @@ -649,7 +652,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->head = (l->head + 1) % l->capacity; + l->head = (l->head + 1) & (l->capacity - 1); --l->count; } l->stored = heard; @@ -1166,6 +1169,8 @@ typedef struct Stream { size_t line_head, line_size, line_capacity; uint64_t line_first; unsigned line_count; + unsigned out_count; /* the delivery out from the line: its records and bytes (let go at once when all applied) */ + size_t out_bytes; bool tail; uint32_t tail_size, tail_got; Held *held; /* records not yet stored, by number: held_count from held_first */ @@ -1223,7 +1228,8 @@ struct PureLinkReceiver { uint64_t ours_at; double accept_tokens; uint64_t accept_at; - struct { Run *run; uint64_t id; } *offers; /* runs whose offers wait for the turn (a slot may hold another run by then) */ + /* Runs whose offers wait for the turn (a slot may hold another run by then). */ + struct { Run *run; uint64_t id; } *offers; size_t offer_count, offer_capacity; PureAead *aead; /* keyed by each packet's run */ unsigned char open_key[PURE_AEAD_KEY], seal_key[PURE_AEAD_KEY]; /* the keys it has now */ @@ -1439,11 +1445,13 @@ static uint32_t memo_check(const unsigned char key[PURE_SEAL_KEY], const unsigne } static unsigned char *memo_place(const PureLinkReceiver *s, uint32_t peer) { - return s->memo && MEMO_HEAD + ((size_t)peer + 1) * MEMO_PLACE <= s->memo_size ? s->memo + MEMO_HEAD + (size_t)peer * MEMO_PLACE - : NULL; + size_t at = MEMO_HEAD + (size_t)peer * MEMO_PLACE; + return s->memo && at + MEMO_PLACE <= s->memo_size ? s->memo + at : NULL; } -static uint32_t epoch_check(const unsigned char *e) { return fnv(fnv(2166136261u, e, 12), e + 16, MEMO_EPOCH_SIZE - 16); } +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 const unsigned char *memo_epoch(const PureLinkReceiver *s, uint32_t peer, uint64_t run) { @@ -1811,12 +1819,14 @@ static bool hand_over(PureLinkReceiver *s, Stream *st) { if (st->known) { /* the line as it is, up to BATCH records */ line_pull(st); if (!st->line_count) return true; - unsigned n = 0; - size_t bytes = 0; - for (; n < st->line_count && n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes); + unsigned n = st->line_count; + size_t bytes = st->line_size - st->line_head; /* the whole line, when it is a batch or less */ + if (n > BATCH) + for (n = 0, bytes = 0; n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes); PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st}; if (!s->host.deliver(s->host.context, &d)) return false; st->out = true; + st->out_count = n, st->out_bytes = bytes; ++s->counts.deliveries; return true; } @@ -2206,7 +2216,13 @@ static void apply_done(PureLinkReceiver *s, const PureLinkDone *done, size_t cou if (!done[k].stored) continue; /* not saved: handed over again */ if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard; if (!st->known) st->line_first = st->heard + 1; /* the line begins */ - line_drop(st, st->heard); + if (st->out_count && st->heard == st->line_first + st->out_count - 1) { /* the delivery applied whole */ + st->line_head += st->out_bytes; + st->line_first += st->out_count; + st->line_count -= st->out_count; + } else + line_drop(st, st->heard); + st->out_count = 0; while (st->held_count && held_at(st, 0)->seq <= st->heard) drop_first(st); st->known = true; st->gap = done[k].gap; diff --git a/link/link-protocol.h b/link/link-protocol.h index 16fd683..e7ca38c 100644 --- a/link/link-protocol.h +++ b/link/link-protocol.h @@ -187,7 +187,7 @@ typedef struct { const char *source; uint64_t first; unsigned count; - const unsigned char *records; /* count of: u32le size, bytes */ + const unsigned char *records; /* count of: u32le size, bytes (the receiver's: the host's during the call) */ size_t size; void *token; /* for its done */ } PureLinkDelivery; -- cgit v1.2.3