Wait for late packets, undo what they cost, and grow as CUBIC

A packet called lost stays known a while; acknowledged after all, it
came late: what it carried counts as received, the reordering awaited
grows (later packets acknowledged, and time past the RTT, as RACK), and
an episode of congestion events whose losses all came late is undone.
The window grows only when it was used (RFC 9002 7.8), and congestion
avoidance is CUBIC (RFC 9438), as quinn's.

In a path of 50 Mbit/s whose packets reorder by up to 5 ms, with no
loss, a link carried 1.5 Mbit/s at an RTT of 20 ms and 10 at 100 ms; now
43.6 and 40.1, as without reordering (research/experiments/020-lossy-paths,
virtual time). With 1% random loss: 4.9 to 6.1 at 20 ms, 1.0 to 1.3 at
100 ms. The transport's CPU a record is unchanged.

4 files changed, 311 insertions, 13 deletions

diff --git a/Makefile b/Makefile
index e3e26c5..e5d8451 100644
--- a/Makefile
+++ b/Makefile
@@ -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;
+}