diff options
Diffstat (limited to 'link')
| -rw-r--r-- | link/link-protocol.c | 147 | ||||
| -rw-r--r-- | link/link-protocol.h | 10 |
2 files changed, 145 insertions, 12 deletions
diff --git a/link/link-protocol.c b/link/link-protocol.c index c8b5118..a7133e0 100644 --- a/link/link-protocol.c +++ b/link/link-protocol.c @@ -4,6 +4,7 @@ #include "link-protocol.h" #include <fcntl.h> #include <malloc.h> +#include <math.h> #include <stdio.h> #include <stdlib.h> #include <string.h> @@ -17,6 +18,12 @@ enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PIN 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 { 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. */ +enum { REORDER_FIRST = 3, REORDER_PACKETS_MOST = 256, REORDER_QUARTERS_MOST = 8, CALM = 16 }; +/* CUBIC's constants (RFC 9438): the cubic's scale, and the window kept at a + * congestion event. */ +static const double CUBIC_C = 0.4, CUBIC_BETA = 0.7; #define MS UINT64_C(1000000) /* RFC 9002's values, but a first RTT of 100 ms (a LAN's, not the 333 ms of * the Internet at large); a probe at least every second while a receiver is @@ -150,8 +157,11 @@ static Piece *piece_of(Record *r, uint32_t i) { return r->pieces == 1 ? &r->one typedef struct { uint64_t seq; uint32_t piece; } Unit; typedef struct { uint64_t pn, sent; /* its number; when it went (ns) */ + uint64_t largest_then; /* lost: the largest number acknowledged when it was called lost */ + uint64_t episode; /* lost: the congestion episode it counted in */ uint32_t bytes; bool eliciting, done; /* acknowledged or lost */ + bool lost, by_time; /* called lost (by time, or by later ones acknowledged), and not acknowledged since */ uint16_t count; Unit units[UNITS]; } Flight; @@ -177,6 +187,28 @@ struct PureLinkSender { uint64_t largest_acked, latest_rtt, smoothed_rtt, rtt_var, min_rtt; bool acked_any, sampled; uint64_t loss_time, last_eliciting, credit, stored_changed; + /* Reordering: later packets acknowledged before a packet is lost (3 at + * first), and the time past the RTT, in quarters of the least RTT (0: + * RFC 9002's eighth of the RTT); both grow when a packet called lost + * turns out late. Congestion events one after another while losses of + * the first are still unsettled make an episode; one whose losses all + * turn out late is undone. Its window, threshold and recovery before + * it; its losses unsettled; whether none of them was real. */ + uint64_t reorder_packets, reorder_quarters, calm_events; + uint64_t episode, undo_window, undo_threshold, undo_recovery, undo_losses; + bool undo_ok; + /* Congestion avoidance is CUBIC (RFC 9438), in datagrams and seconds: + * the window before the last event, the time the cubic takes to reach + * it again, when the epoch began (0: none yet), and what Reno would have + * (the window never grows slower). As they were before an episode, for + * its undoing. */ + double cubic_max, cubic_k, cubic_reno; + uint64_t cubic_epoch; + double undo_cubic_max, undo_cubic_k, undo_cubic_reno; + uint64_t undo_cubic_epoch; + /* The window went unused (nothing to send, or no credit) when the + * sender last looked: it does not grow then (RFC 9002 ยง7.8). */ + bool unused; uint64_t received; /* pieces received and not yet stored */ uint64_t unsent_from; /* no record before this number has a piece unsent */ unsigned pto_count, probes; @@ -301,8 +333,11 @@ static void settle(PureLinkSender *l, Flight *f) { } /* A packet lost: what it carried and was not received otherwise goes again. */ -static void lose(PureLinkSender *l, Flight *f, uint64_t now) { +static void lose(PureLinkSender *l, Flight *f, uint64_t now, bool by_time) { settle(l, f); + f->lost = true; + f->by_time = by_time; + f->largest_then = l->largest_acked; ++l->counts.lost; for (uint16_t u = 0; u < f->count; ++u) { Record *r = record_of(l, f->units[u].seq); @@ -313,22 +348,85 @@ static void lose(PureLinkSender *l, Flight *f, uint64_t now) { } } if (f->sent > l->recovery_start) { /* one congestion event per round trip */ + if (!l->undo_losses) { /* an episode begins */ + ++l->episode; + l->undo_window = l->window; + l->undo_threshold = l->threshold; + l->undo_recovery = l->recovery_start; + l->undo_cubic_max = l->cubic_max, l->undo_cubic_k = l->cubic_k, l->undo_cubic_reno = l->cubic_reno; + l->undo_cubic_epoch = l->cubic_epoch; + l->undo_ok = true; + } l->recovery_start = now; - l->window = most(l->window / 2, WINDOW_LEAST); + double w = (double)l->window / DATAGRAM; + l->cubic_max = w < l->cubic_max ? w * (1 + CUBIC_BETA) / 2 : w; /* fast convergence: another flow came */ + l->window = most((uint64_t)((double)l->window * CUBIC_BETA), WINDOW_LEAST); l->threshold = l->window; + l->cubic_k = cbrt(l->cubic_max * (1 - CUBIC_BETA) / CUBIC_C); + l->cubic_epoch = now; + l->cubic_reno = (double)l->window / DATAGRAM; + if (++l->calm_events >= CALM) { /* many events, none of them late packets: less patience */ + l->reorder_packets = most(REORDER_FIRST, l->reorder_packets / 2); + l->reorder_quarters /= 2; + l->calm_events = 0; + } + } + f->episode = l->episode; + ++l->undo_losses; +} + +/* A loss of the episode settled: it came late, or it was real. */ +static void settle_loss(PureLinkSender *l, const Flight *f, bool real) { + if (f->episode != l->episode || !l->undo_losses) return; + if (real) l->undo_ok = false; + if (!--l->undo_losses && l->undo_ok) { + l->window = most(l->window, l->undo_window); + l->threshold = l->undo_threshold; + l->recovery_start = l->undo_recovery; + l->cubic_max = l->undo_cubic_max, l->cubic_k = l->undo_cubic_k, l->cubic_reno = l->undo_cubic_reno; + l->cubic_epoch = l->undo_cubic_epoch; + ++l->counts.undone; + } +} + +/* A packet called lost acknowledged after all: it came late. What it + * carried was received (not sent again, if it waits to be); later packets + * are awaited longer; and when every loss of the last congestion event + * came late, the event is undone. */ +static void came_late(PureLinkSender *l, Flight *f, uint64_t now) { + f->lost = false; + ++l->counts.late; + for (uint16_t u = 0; u < f->count; ++u) { + Record *r = record_of(l, f->units[u].seq); + if (r && (*piece_of(r, f->units[u].piece)).state != RECEIVED) { + set_piece(l, r, f->units[u].piece, RECEIVED); + l->stored_changed = now; + } } + if (f->by_time) l->reorder_quarters = least(l->reorder_quarters + 1, REORDER_QUARTERS_MOST); + else l->reorder_packets = least(most(l->reorder_packets, l->largest_acked - f->pn + 1), REORDER_PACKETS_MOST); + l->calm_events = 0; + settle_loss(l, f, false); +} + +/* How long past the RTT a packet may come before it is called lost. */ +static uint64_t reorder_time(const PureLinkSender *l, uint64_t rtt) { + uint64_t wait = rtt / 8; + if (l->reorder_quarters && l->sampled) wait = most(wait, l->min_rtt / 4 * l->reorder_quarters); + return least(wait, rtt); } static void detect_lost(PureLinkSender *l, uint64_t now) { l->loss_time = 0; if (!l->acked_any) return; uint64_t rtt = most(l->latest_rtt, l->sampled ? l->smoothed_rtt : FIRST_RTT); - uint64_t delay = most(rtt * 9 / 8, GRANULARITY); + 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]; if (f->pn >= l->largest_acked) break; if (f->done) continue; - if (f->sent + delay <= now || l->largest_acked >= f->pn + 3) lose(l, f, now); + bool late = f->sent + delay <= now; + if (late || l->largest_acked >= f->pn + l->reorder_packets) lose(l, f, now, late); else if (!l->loss_time || f->sent + delay < l->loss_time) l->loss_time = f->sent + delay; } } @@ -349,16 +447,42 @@ static void sample_rtt(PureLinkSender *l, uint64_t latest, uint64_t ack_delay) { l->smoothed_rtt = (7 * l->smoothed_rtt + adjusted) / 8; } +/* The window after bytes acknowledged: slow start below the threshold, + * then CUBIC. The cubic's target is where it will be an RTT on, and the + * window closes on it by the share of the window acknowledged. */ +static void grow(PureLinkSender *l, uint64_t bytes, uint64_t now) { + if (l->window < l->threshold) { + l->window += bytes; + return; + } + double w = (double)l->window / DATAGRAM, acked = (double)bytes / DATAGRAM; + if (!l->cubic_epoch) { /* no event yet: the cubic starts where the window is */ + l->cubic_epoch = now; + l->cubic_max = l->cubic_reno = w; + l->cubic_k = 0; + } + double t = (double)(now - l->cubic_epoch) / 1e9, rtt = (double)(l->sampled ? l->smoothed_rtt : FIRST_RTT) / 1e9; + l->cubic_reno += (l->cubic_reno < l->cubic_max ? 3 * (1 - CUBIC_BETA) / (1 + CUBIC_BETA) : 1) * acked / w; + double d = t - l->cubic_k, at = CUBIC_C * d * d * d + l->cubic_max; + d += rtt; + double target = CUBIC_C * d * d * d + l->cubic_max; + target = target < w ? w : target > 1.5 * w ? 1.5 * w : target; + double next = at < l->cubic_reno ? l->cubic_reno : w + (target - w) * acked / w; + l->window = most(l->window, (uint64_t)(next * DATAGRAM)); +} + static void on_ack(PureLinkSender *l, const unsigned char *p, size_t size, uint64_t now) { if (size < 13 || size < 13 + 16u * p[12]) return; uint64_t largest = get64(p), delay = (uint64_t)get32(p + 8) * 1000; Flight *top = NULL; + if (!l->acked_any || largest > l->largest_acked) l->largest_acked = largest; for (unsigned k = 0; k < p[12]; ++k) { uint64_t first = get64(p + 13 + 16 * k), last = get64(p + 21 + 16 * k); if (first > last || !l->flight_count) continue; uint64_t oldest = l->flights[l->flight_head].pn, newest = oldest + l->flight_count - 1; for (uint64_t pn = most(first, oldest); pn <= least(last, newest); ++pn) { Flight *f = flight_of(l, pn); + if (f && f->lost) came_late(l, f, now); if (!f || f->done) continue; settle(l, f); ++l->counts.acknowledged; @@ -369,17 +493,21 @@ static void on_ack(PureLinkSender *l, const unsigned char *p, size_t size, uint6 l->stored_changed = now; /* waits for its save from now */ } } - if (f->eliciting && f->sent > l->recovery_start) - l->window += l->window < l->threshold ? f->bytes : (uint64_t)DATAGRAM * f->bytes / l->window; + if (f->eliciting && f->sent > l->recovery_start && !l->unused) grow(l, f->bytes, now); if (pn == largest) top = f; } } - if (!l->acked_any || largest > l->largest_acked) l->largest_acked = largest; l->acked_any = true; if (top && top->eliciting) sample_rtt(l, now - top->sent, delay); l->pto_count = 0; detect_lost(l, now); - while (l->flight_count && l->flights[l->flight_head].done) { + /* Done ones go; one called lost stays a while, so it is known if it + * comes late. */ + uint64_t keep = 2 * most(l->latest_rtt, l->sampled ? l->smoothed_rtt : FIRST_RTT); + while (l->flight_count) { + 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_count; } @@ -571,11 +699,13 @@ static size_t transmit(PureLinkSender *l, Flight *f, const unsigned char *body, size_t pure_link_sender_packet(PureLinkSender *l, unsigned char out[PURE_LINK_DATAGRAM], bool whole, uint64_t now, uint64_t *number) { bool probe = l->probes > 0; + if (!probe && l->in_flight + DATAGRAM > l->window) l->unused = false; if (!l->out || (!probe && l->in_flight + DATAGRAM > l->window) || !flight_room(l)) return 0; unsigned char body[BODY]; Flight f = {.pn = l->next_pn, .eliciting = true}; size_t size = fill(l, body, &f, whole); if (!size) { + l->unused = true; if (!probe) return 0; body[0] = F_PING; /* nothing to send again: a probe all the same */ size = 1; @@ -653,6 +783,7 @@ PureLinkSender *pure_link_sender(const char *source, const char *input, uint32_t l->window = WINDOW_FIRST; l->threshold = UINT64_MAX; l->credit = CREDIT; + l->reorder_packets = REORDER_FIRST; return l; } diff --git a/link/link-protocol.h b/link/link-protocol.h index 3037926..3f71809 100644 --- a/link/link-protocol.h +++ b/link/link-protocol.h @@ -57,11 +57,13 @@ enum { PURE_LINK_DATAGRAM = 1200, PURE_LINK_NAME = 64, PURE_LINK_ADDRESS = 28 }; /* Counts for a host's report: packets and records sent, records sent again, * packets lost and acknowledged, packets received, deliveries handed over, - * the last number stored; a receiver's senders woken (their active state - * made), of them with a secret remembered (no X25519), let go when idle, - * and active now. */ + * the last number stored; packets called lost that came late, and + * congestion events undone for it; a receiver's senders woken (their + * active state made), of them with a secret remembered (no X25519), let go + * when idle, and active now. */ typedef struct { - uint64_t packets, records, resent, lost, acknowledged, received, deliveries, stored, woken, remembered, swept, active; + uint64_t packets, records, resent, lost, acknowledged, received, deliveries, stored, late, undone, woken, remembered, + swept, active; } PureLinkCounts; /* ---- The sending side: one output's saved sends to a receiver's input, as |