diff options
| -rw-r--r-- | Makefile | 2 | ||||
| -rw-r--r-- | link/link-protocol.c | 147 | ||||
| -rw-r--r-- | link/link-protocol.h | 10 | ||||
| -rw-r--r-- | research/experiments/020-lossy-paths/sim.c | 165 |
4 files changed, 311 insertions, 13 deletions
@@ -78,7 +78,7 @@ $(BUILD)/test-module: core/test-module.c $(MODEL) $(STORE) $(HEADERS) # A link's two sides through a network of the test's own (pure-crypto.so beside it). $(BUILD)/test-link: link/test-link.c $(BUILD)/libpure-link.a $(BUILD)/pure-crypto.so $(wildcard link/*.h) @mkdir -p $(BUILD) - $(CC) $(CPPFLAGS) $(CFLAGS) link/test-link.c $(BUILD)/libpure-link.a $(LDFLAGS) -pthread -ldl $(LDLIBS) -o $@ + $(CC) $(CPPFLAGS) $(CFLAGS) link/test-link.c $(BUILD)/libpure-link.a $(LDFLAGS) -lm -pthread -ldl $(LDLIBS) -o $@ $(BUILD)/test-message-json: serve/test-message-json.c serve/message-json.c $(BUILD)/libpure.a $(HEADERS) @mkdir -p $(BUILD) 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 diff --git a/research/experiments/020-lossy-paths/sim.c b/research/experiments/020-lossy-paths/sim.c new file mode 100644 index 0000000..00f93d6 --- /dev/null +++ b/research/experiments/020-lossy-paths/sim.c @@ -0,0 +1,165 @@ +/* M6 of docs/design/011-pure-transport.md in virtual time: one link + * through a path of the sim's own. Each direction: random loss, a + * bottleneck (rate, queue limit, drop-tail), a one-way delay and random + * extra delay (so packets reorder). Records of a size, queued all at once + * (a backlog), stored by the host at once. It takes in link-protocol.c + * whole, to read the sender's window and RTT. + * + * sim SIZE COUNT DELAY_MS LOSS_PERMIL JITTER_MS RATE_MBIT QUEUE_PACKETS SEED + * prints a line of JSON: virtual seconds, records a second, packets sent, + * dropped (by loss, by the queue), called lost, of them late, congestion + * episodes undone, sent again, and the window and RTT seen each 100 ms. + * + * cc -O2 -std=c17 -I link research/experiments/020-lossy-paths/sim.c link/crypto.c -lm -ldl -o build/sim + * PURE_CRYPTO=build/pure-crypto.so build/sim 1000 30000 10 10 5 50 100 1 */ +#include "link-protocol.c" +#include <math.h> + +typedef struct { + uint64_t at; + bool to_receiver; + size_t size; + unsigned char bytes[PURE_LINK_DATAGRAM]; +} Packet; + +static uint64_t sim_seed; +static uint64_t sim_random(void) { + sim_seed ^= sim_seed << 13, sim_seed ^= sim_seed >> 7, sim_seed ^= sim_seed << 17; + return sim_seed; +} + +static Packet *packets; +static size_t packet_count, packet_capacity; +static uint64_t now, delay_ns, jitter, loss_permil, rate_bps, queue_limit; +static uint64_t link_free[2]; /* when each direction's bottleneck is free */ +static uint64_t dropped_loss, dropped_queue, sent_count; +static uint64_t heard, records, size; +static PureLinkDone dones[64]; +static size_t done_count; + +static void put(bool to_receiver, const unsigned char *bytes, size_t n) { + ++sent_count; + if (sim_random() % 1000 < loss_permil) { + ++dropped_loss; + return; + } + int d = to_receiver ? 0 : 1; + uint64_t start = link_free[d] > now ? link_free[d] : now; + uint64_t serial = rate_bps ? (uint64_t)((n + 28) * 8) * 1000000000u / rate_bps : 0; + /* queue: packets waiting before this one, by time */ + if (rate_bps && serial && (start - now) / serial >= queue_limit) { + ++dropped_queue; + return; + } + link_free[d] = start + serial; + if (packet_count == packet_capacity) { + packet_capacity = packet_capacity ? 2 * packet_capacity : 1024; + packets = realloc(packets, packet_capacity * sizeof(*packets)); + } + Packet *p = &packets[packet_count++]; + p->at = start + serial + delay_ns + (jitter ? sim_random() % jitter : 0); + p->to_receiver = to_receiver; + p->size = n; + memcpy(p->bytes, bytes, n); +} + +static bool sim_answer(void *context, const unsigned char *packet, size_t n, const unsigned char *to, size_t to_size) { + (void)context, (void)to, (void)to_size; + put(false, packet, n); + return true; +} + +static int input_of(void *context, const char *name, uint32_t shape) { + (void)context, (void)name, (void)shape; + return 0; +} + +static bool sim_deliver(void *context, const PureLinkDelivery *d) { + (void)context; + if (done_count == 64) return false; + uint64_t h = heard; + for (unsigned k = 0; k < d->count; ++k) + if (d->first + k == h + 1) h = d->first + k; + heard = h; + dones[done_count++] = (PureLinkDone){d->token, heard, true, d->first > heard + 1}; + return true; +} + +int main(int argc, char **argv) { + if (argc < 9) return 2; + size = strtoull(argv[1], NULL, 10); + records = strtoull(argv[2], NULL, 10); + delay_ns = strtoull(argv[3], NULL, 10) * MS; + loss_permil = strtoull(argv[4], NULL, 10); + jitter = strtoull(argv[5], NULL, 10) * MS; + rate_bps = strtoull(argv[6], NULL, 10) * 1000000u; + queue_limit = strtoull(argv[7], NULL, 10); + sim_seed = strtoull(argv[8], NULL, 10) * 0x9e3779b97f4a7c15u | 1; + char dir[] = "/tmp/pure-sim-XXXXXX", path[512], hex[65], error[256]; + if (!mkdtemp(dir)) return 1; + unsigned char pa[32], pb[32]; + snprintf(path, sizeof(path), "%s/a", dir); + PureSealKey *a = pure_seal_key(path, hex, error, sizeof(error)); + pure_seal_peer(hex, pa); + snprintf(path, sizeof(path), "%s/b", dir); + PureSealKey *b = pure_seal_key(path, hex, error, sizeof(error)); + pure_seal_peer(hex, pb); + PureLinkSender *s = pure_link_sender("sim.out", "order", 1, a, pa); + pure_link_sender_peer(s, pb); + PureLinkReceiver *r = pure_link_receiver(b, pb, &(PureLinkHost){input_of, sim_deliver, NULL}, error, sizeof(error)); + pure_link_allow(r, "sim.out", pa, error, sizeof(error)); + unsigned char *bytes = calloc(1, size + 1); + for (uint64_t seq = 1; seq <= records; ++seq) + if (!pure_link_sender_queue(s, seq, bytes, size)) return 1; + unsigned char out[PURE_LINK_DATAGRAM]; + uint64_t sample_at = 0, window_sum = 0, rtt_sum = 0, samples = 0, window_least = UINT64_MAX; + PureLinkDone told[64]; + while (heard < records || s->stored < records) { + uint64_t number; + for (size_t n; (n = pure_link_sender_packet(s, out, false, now, &number));) put(true, out, n); + uint64_t next = UINT64_MAX, due = pure_link_sender_due(s), turn = pure_link_receiver_due(r); + for (size_t k = 0; k < packet_count; ++k) + if (packets[k].at < next) next = packets[k].at; + if (due && due < next) next = due; + if (turn && turn < next) next = turn; + if (done_count && now + 1000 < next) next = now + 1000; + if (next == UINT64_MAX || next > 3600000 * MS) { + fprintf(stderr, "sim: stuck at %llu of %llu\n", (unsigned long long)heard, (unsigned long long)records); + return 1; + } + now = next > now ? next : now; + while (now >= sample_at) { + window_sum += s->window, rtt_sum += s->smoothed_rtt, ++samples; + if (s->window < window_least) window_least = s->window; + sample_at += 100 * MS; + } + for (size_t k = 0; k < packet_count;) { + Packet *p = &packets[k]; + if (p->at > now) { + ++k; + continue; + } + if (p->to_receiver) pure_link_receiver_take(r, p->bytes, p->size, "s", 1, now); + else pure_link_sender_take(s, p->bytes, p->size, now); + *p = packets[--packet_count]; + } + pure_link_sender_tick(s, now); + size_t count = done_count; + memcpy(told, dones, count * sizeof(*dones)); + done_count = 0; + pure_link_receiver_turn(r, told, count, now, sim_answer, NULL); + } + double seconds = (double)now / 1e9; + printf("{\"size\":%llu,\"records\":%llu,\"delay_ms\":%llu,\"loss_permil\":%llu,\"jitter_ms\":%llu,\"rate_mbit\":%llu," + "\"queue\":%llu,\"seconds\":%.3f,\"per_second\":%.0f,\"mbit\":%.2f,\"sent\":%llu,\"dropped_loss\":%llu," + "\"dropped_queue\":%llu,\"declared_lost\":%llu,\"late\":%llu,\"undone\":%llu,\"resent\":%llu,\"window_mean\":%.0f,\"window_least\":%llu," + "\"srtt_mean_ms\":%.2f}\n", + (unsigned long long)size, (unsigned long long)records, (unsigned long long)(delay_ns / MS), + (unsigned long long)loss_permil, (unsigned long long)(jitter / MS), (unsigned long long)(rate_bps / 1000000), + (unsigned long long)queue_limit, seconds, records / seconds, records * (double)size * 8 / seconds / 1e6, + (unsigned long long)s->counts.packets, (unsigned long long)dropped_loss, (unsigned long long)dropped_queue, + (unsigned long long)s->counts.lost, (unsigned long long)s->counts.late, (unsigned long long)s->counts.undone, (unsigned long long)s->counts.resent, (double)window_sum / samples, + (unsigned long long)window_least, (double)rtt_sum / samples / 1e6); + snprintf(path, sizeof(path), "rm -rf %s", dir); + return system(path) ? 1 : 0; +} |