Measure a link on a path of loss, reordering and a bottleneck (M6 harness)

020-lossy-paths/run.py runs each case in an unprivileged namespace: a
50 Mbit/s tbf bottleneck under netem's delay, jitter and loss, both
ways. A link may send without UDP GSO (pure_link_gso), as netem takes a
burst as one packet; quinn the same (QUIC_GSO=0). transport.c and the
quinn harness give the records handed over in each 100 ms, and at a
steady rate the time each took; TCP CUBIC flows run beside them.

3 files changed, 374 insertions, 31 deletions

diff --git a/research/experiments/019-idle-links/quic/src/main.rs b/research/experiments/019-idle-links/quic/src/main.rs
index 80e663f..050be25 100644
--- a/research/experiments/019-idle-links/quic/src/main.rs
+++ b/research/experiments/019-idle-links/quic/src/main.rs
@@ -20,10 +20,15 @@
 //!       connects again and sends it again. Then a line of JSON: orders made
 //!       and stored, connections made again, the orders stored in each
 //!       100 ms, the time from an order made to it stored
-//!   idle-quic drain PORT COUNT SIZE
+//!   idle-quic drain PORT COUNT SIZE [RATE [SECONDS]]
 //!       one client sends COUNT messages of SIZE bytes over one stream
 //!       (each after its u32 length) and waits until the service says it
-//!       read them all; a line of JSON: how long
+//!       read them all; a line of JSON: how long. RATE: messages a second
+//!       (0: as fast as it may), each with the time it was written, so the
+//!       service knows how long each took; SECONDS: stop writing after that
+//!       long. The service writes a line of JSON for each such stream: the
+//!       messages read in each 100 ms, and how long they took.
+//!   QUIC_GSO=0: no UDP GSO (a path of netem takes a burst as one packet)
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::{Arc, Mutex};
@@ -40,6 +45,9 @@ fn transport() -> Arc<TransportConfig> {
     let mut t = TransportConfig::default();
     t.max_idle_timeout(Some(IdleTimeout::from(VarInt::from_u32(3_600_000)))); // an hour: connections at rest stay
     t.keep_alive_interval(None);
+    if std::env::var("QUIC_GSO").as_deref() == Ok("0") {
+        t.enable_segmentation_offload(false);
+    }
     Arc::new(t)
 }
 
@@ -80,8 +88,8 @@ fn serve(cert: Vec<u8>, key: Vec<u8>, port: u16) {
                     if recv.read_exact(&mut first).await.is_err() {
                         return;
                     }
-                    if first == *DRAIN {
-                        drain(&mut send, &mut recv).await;
+                    if first == *DRAIN || first == *PACED {
+                        drain(&mut send, &mut recv, first == *PACED).await;
                         continue;
                     }
                     let Ok(rest) = recv.read_to_end(1024).await else { return };
@@ -114,12 +122,21 @@ fn serve(cert: Vec<u8>, key: Vec<u8>, port: u16) {
 }
 
 /// A drain's stream begins with these bytes; an order's never does (a client number that large).
+/// A paced drain's messages each begin with the time they were written (ns since the epoch).
 const DRAIN: &[u8; 4] = b"pure";
+const PACED: &[u8; 4] = b"pace";
+
+fn epoch_ns() -> u64 {
+    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() as u64
+}
 
 /// Messages each after its u32 length until a length of 0, read a chunk at
-/// a time as they come; then how many, as the answer.
-async fn drain(send: &mut quinn::SendStream, recv: &mut quinn::RecvStream) {
+/// a time as they come; then how many, as the answer, and a line of JSON:
+/// the messages read in each 100 ms, and (paced) how long each took.
+async fn drain(send: &mut quinn::SendStream, recv: &mut quinn::RecvStream, paced: bool) {
     let (mut count, mut pending, mut done) = (0u64, Vec::<u8>::new(), false);
+    let began = Instant::now();
+    let (mut timeline, mut waited) = (Vec::<u32>::new(), Vec::<u64>::new());
     while !done {
         let Ok(Some(chunk)) = recv.read_chunk(1 << 20, true).await else { return };
         pending.extend_from_slice(&chunk.bytes);
@@ -133,11 +150,25 @@ async fn drain(send: &mut quinn::SendStream, recv: &mut quinn::RecvStream) {
             if pending.len() - at - 4 < length {
                 break;
             }
+            let bucket = (began.elapsed().as_millis() / 100) as usize;
+            if timeline.len() <= bucket {
+                timeline.resize(bucket + 1, 0);
+            }
+            timeline[bucket] += 1;
+            if paced && length >= 8 {
+                let stamp = u64::from_le_bytes(pending[at + 4..at + 12].try_into().unwrap());
+                waited.push(epoch_ns().saturating_sub(stamp));
+            }
             at += 4 + length;
             count += 1;
         }
         pending.drain(..at);
     }
+    waited.sort_unstable();
+    let pick = |q: usize| waited.get((waited.len() * q / 100).min(waited.len().saturating_sub(1))).copied().unwrap_or(0) as f64 / 1e6;
+    println!("{{\"drained\":{count},\"waited_ms_p50\":{:.3},\"waited_ms_p99\":{:.3},\"waited_ms_max\":{:.3},\"timeline_100ms\":[{}]}}",
+             pick(50), pick(99), waited.last().copied().unwrap_or(0) as f64 / 1e6,
+             timeline.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(","));
     let _ = send.write_all(&count.to_le_bytes()).await;
     let _ = send.finish();
 }
@@ -351,8 +382,9 @@ fn steady(port: u16, first: u64, count: u64, period: Duration, seconds: u64, win
     });
 }
 
-/// One client, COUNT messages of SIZE bytes over one stream, and the answer.
-fn drain_client(port: u16, count: u64, size: usize) {
+/// One client, COUNT messages of SIZE bytes over one stream (at rate a
+/// second, 0: as fast as it may; for at most seconds), and the answer.
+fn drain_client(port: u16, count: u64, size: usize, rate: f64, seconds: f64) {
     let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
     runtime.block_on(async move {
         let mut endpoint = Endpoint::client(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
@@ -365,9 +397,27 @@ fn drain_client(port: u16, count: u64, size: usize) {
         let per_chunk = ((64 << 10) / frame.len()).max(1);
         let chunk: Vec<u8> = frame.iter().copied().cycle().take(frame.len() * per_chunk).collect();
         let began = Instant::now();
-        send.write_all(DRAIN).await.unwrap();
+        let limit = if seconds > 0.0 { Duration::from_secs_f64(seconds) } else { Duration::MAX };
         let mut left = count as usize;
-        while left > 0 {
+        if rate > 0.0 {
+            send.write_all(PACED).await.unwrap();
+            let mut message = frame.clone();
+            for i in 0..count {
+                let due = Duration::from_secs_f64(i as f64 / rate);
+                if due >= limit {
+                    break;
+                }
+                if let Some(wait) = due.checked_sub(began.elapsed()) {
+                    tokio::time::sleep(wait).await;
+                }
+                message[4..12].copy_from_slice(&epoch_ns().to_le_bytes());
+                send.write_all(&message).await.unwrap();
+            }
+            left = 0;
+        } else {
+            send.write_all(DRAIN).await.unwrap();
+        }
+        while left > 0 && began.elapsed() < limit {
             let n = left.min(per_chunk);
             send.write_all(&chunk[..n * frame.len()]).await.unwrap();
             left -= n;
@@ -388,8 +438,10 @@ fn main() {
         Some("serve") if args.len() == 4 || args.len() == 5 => serve(std::fs::read(&args[2]).unwrap(),
                                                                     std::fs::read(&args[3]).unwrap(),
                                                                     args.get(4).map_or(0, |p| p.parse().unwrap())),
-        Some("drain") if args.len() == 5 => drain_client(args[2].parse().unwrap(), args[3].parse().unwrap(),
-                                                         args[4].parse().unwrap()),
+        Some("drain") if (5..=7).contains(&args.len()) => drain_client(args[2].parse().unwrap(), args[3].parse().unwrap(),
+                                                                       args[4].parse().unwrap(),
+                                                                       args.get(5).map_or(0.0, |r| r.parse().unwrap()),
+                                                                       args.get(6).map_or(0.0, |s| s.parse().unwrap())),
         Some("steady") if args.len() == 8 => steady(args[2].parse().unwrap(), args[3].parse().unwrap(),
                                                     args[4].parse().unwrap(),
                                                     Duration::from_millis(args[5].parse().unwrap()),
@@ -398,7 +450,7 @@ fn main() {
                                                       args[4].parse().unwrap(), args[5].parse().unwrap()),
         _ => {
             eprintln!("idle-quic serve CERT.der KEY.der [PORT] | idle-quic connect PORT FIRST COUNT WINDOW | \
-                       idle-quic steady PORT FIRST COUNT PERIOD_MS SECONDS WINDOW | idle-quic drain PORT COUNT SIZE");
+                       idle-quic steady PORT FIRST COUNT PERIOD_MS SECONDS WINDOW | idle-quic drain PORT COUNT SIZE [RATE [SECONDS]]");
             std::process::exit(2);
         }
     }
diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c
index 5827311..8ff34a4 100644
--- a/research/experiments/019-idle-links/transport.c
+++ b/research/experiments/019-idle-links/transport.c
@@ -1,15 +1,20 @@
 #define _GNU_SOURCE
 /* A link and a listener in one process, with no app and no store: what the
- * transport alone costs a record (M5 of docs/design/011-pure-transport.md).
+ * transport alone costs a record (M5 of docs/design/011-pure-transport.md),
+ * and how it fares on a path of loss and queues (M6).
  * The listener says each delivery stored as it hands it over.
  *
- *   transport SENDER.pmod RECEIVER.pmod COUNT SIZE [WINDOW]
+ *   transport SENDER.pmod RECEIVER.pmod COUNT SIZE [WINDOW [RATE [SECONDS]]]
  *       COUNT records of SIZE item bytes (an order of the sender's output
  *       "ordered", to the receiver's input "order"), at most WINDOW waiting
  *       at once (100,000); a line of JSON: how long, records a second, and
  *       CPU a record by thread (the link's, the listener's, the queuing
- *       one's). WINDOW 0: all are queued before the link starts (a backlog
- *       drained). Built beside pure-crypto.so (build/). */
+ *       one's), the records handed over in each 100 ms. WINDOW 0: all are
+ *       queued before the link starts (a backlog drained). RATE: records
+ *       queued a second (0: as fast as the window allows), and the time
+ *       from each queued to handed over. SECONDS: stop after that long
+ *       (0: once all are handed over). TRANSPORT_GSO=0: no UDP GSO bursts.
+ *       Built beside pure-crypto.so (build/). */
 #include "link.h"
 #include "module.h"
 #include "seal.h"
@@ -30,6 +35,11 @@ static atomic_uint_fast64_t delivered;
 static unsigned char *payload;
 static size_t payload_size;
 static uint64_t heard, wrong; /* the last record taken; deliveries not as sent */
+static uint64_t began, *queued_at, *waited; /* by number: when queued, how long until handed over (ns) */
+enum { BUCKETS = 36000 };                   /* an hour of 100 ms */
+static uint32_t timeline[BUCKETS];
+
+static uint64_t now_ns(void);
 
 /* Records as sent, each once, in order: the one after the last taken on
  * (one taken before would be a delivery again), of the payload's bytes. */
@@ -41,6 +51,10 @@ static bool deliver(void *context, const PureLinkDelivery *d) {
         if (d->first + k != heard + 1 + k || size != payload_size || memcmp(at + 4, payload, size)) ++wrong;
         at += 4 + size;
     }
+    uint64_t now = now_ns();
+    if (queued_at)
+        for (unsigned k = 0; k < d->count; ++k) waited[d->first + k] = now - queued_at[d->first + k];
+    if ((now - began) / 100000000u < BUCKETS) timeline[(now - began) / 100000000u] += d->count;
     if ((size_t)(at - d->records) != d->size || d->first > heard + 1) ++wrong;
     if (d->first + d->count - 1 > heard) heard = d->first + d->count - 1;
     atomic_fetch_add(&delivered, d->count);
@@ -63,6 +77,11 @@ static uint64_t now_ns(void) {
     return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec;
 }
 
+static int by_value(const void *a, const void *b) {
+    uint64_t x = *(const uint64_t *)a, y = *(const uint64_t *)b;
+    return x < y ? -1 : x > y;
+}
+
 /* CPU ns of this process's threads named name (NULL: all). */
 static uint64_t thread_cpu(const char *name) {
     uint64_t total = 0;
@@ -117,6 +136,8 @@ int main(int argc, char **argv) {
     }
     char error[256], dir[] = "/tmp/pure-transport-XXXXXX", path[512], hex[2 * PURE_SEAL_KEY + 1];
     uint64_t count = strtoull(argv[3], NULL, 10), window = argc > 5 ? strtoull(argv[5], NULL, 10) : 100000;
+    double rate = argc > 6 ? strtod(argv[6], NULL) : 0, limit = argc > 7 ? strtod(argv[7], NULL) : 0;
+    const char *gso = getenv("TRANSPORT_GSO");
     size_t size = strtoull(argv[4], NULL, 10);
     if (pure_module_read(argv[1], &sender, error, sizeof(error)) != PURE_OK ||
         pure_module_read(argv[2], &receiver, error, sizeof(error)) != PURE_OK || !mkdtemp(dir)) {
@@ -150,6 +171,9 @@ int main(int argc, char **argv) {
                                                  public_a, error, sizeof(error));
     if (!link) return fprintf(stderr, "transport: %s\n", output < 0 ? "no output 'ordered'" : error), 1;
     pure_link_peer(link, public_b);
+    if (gso && !strcmp(gso, "0")) pure_link_gso(link, false);
+    if (rate > 0 && (!(queued_at = calloc(count + 1, sizeof(*queued_at))) || !(waited = calloc(count + 1, sizeof(*waited)))))
+        return fputs("transport: out of memory\n", stderr), 1;
     payload = malloc(size + 256);
     payload_size = order(&sender, (uint32_t)output, size, payload);
     for (size_t i = 0; i < payload_size; ++i) payload[i] ^= (unsigned char)(i * 7); /* bytes that differ by place */
@@ -157,32 +181,52 @@ int main(int argc, char **argv) {
     if (!window) /* a backlog: all queued before the link starts */
         for (; first <= count; ++first)
             if (!pure_link_queue(link, first, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1;
-    uint64_t began = now_ns(), cpu_link = 0, cpu_listen = thread_cpu("pure-listen"), cpu_all = thread_cpu(NULL);
+    began = now_ns();
+    uint64_t cpu_link = 0, cpu_listen = thread_cpu("pure-listen"), cpu_all = thread_cpu(NULL);
+    uint64_t end = limit > 0 ? began + (uint64_t)(limit * 1e9) : UINT64_MAX;
     if (!pure_link_start(link, notify, NULL, error, sizeof(error))) return fprintf(stderr, "transport: %s\n", error), 1;
-    for (uint64_t seq = first; seq <= count; ++seq) {
-        while (seq - pure_link_delivered(link) > window) {
+    for (uint64_t seq = first; seq <= count && now_ns() < end; ++seq) {
+        uint64_t due = rate > 0 ? began + (uint64_t)((double)(seq - first) * 1e9 / rate) : 0;
+        while ((seq - pure_link_delivered(link) > window || now_ns() < due) && now_ns() < end) {
             struct timespec pause = {0, 50000};
             nanosleep(&pause, NULL);
         }
+        if (now_ns() >= end) break;
+        if (queued_at) queued_at[seq] = now_ns();
         if (!pure_link_queue(link, seq, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1;
     }
-    while (pure_link_delivered(link) < count) {
+    while (pure_link_delivered(link) < count && now_ns() < end) {
         struct timespec pause = {0, 50000};
         nanosleep(&pause, NULL);
     }
-    double seconds = (double)(now_ns() - began) / 1e9;
+    double seconds = (double)((end < now_ns() ? end : now_ns()) - began) / 1e9;
     uint64_t link_ns = thread_cpu("pure-link") - cpu_link, listen_ns = thread_cpu("pure-listen") - cpu_listen,
              all_ns = thread_cpu(NULL) - cpu_all;
-    PureLinkCounts sent;
+    PureLinkCounts sent, got;
     pure_link_counts(link, &sent);
-    printf("{\"records\":%llu,\"record_bytes\":%zu,\"seconds\":%.3f,\"per_second\":%.0f,\"packets\":%llu,"
-           "\"wrong\":%llu,\"resent\":%llu,\"link_us_per_record\":%.3f,\"listener_us_per_record\":%.3f,\"other_us_per_record\":%.3f}\n",
-           (unsigned long long)count, payload_size, seconds, (double)count / seconds, (unsigned long long)sent.packets,
-           (unsigned long long)wrong, (unsigned long long)sent.resent, (double)link_ns / 1e3 / (double)count, (double)listen_ns / 1e3 / (double)count,
-           (double)(all_ns - link_ns - listen_ns) / 1e3 / (double)count);
     pure_link_close(link);
-    PureLinkCounts heard;
-    pure_link_listener_close(listener, &heard);
+    pure_link_listener_close(listener, &got);
+    uint64_t done = heard; /* in order, as the listener handed them over */
+    if (!done) done = 1;
+    printf("{\"records\":%llu,\"delivered\":%llu,\"record_bytes\":%zu,\"seconds\":%.3f,\"per_second\":%.0f,\"mbit\":%.2f,"
+           "\"packets\":%llu,\"wrong\":%llu,\"resent\":%llu,\"lost\":%llu,\"late\":%llu,\"undone\":%llu,"
+           "\"link_us_per_record\":%.3f,\"listener_us_per_record\":%.3f,\"other_us_per_record\":%.3f",
+           (unsigned long long)count, (unsigned long long)heard, payload_size, seconds, (double)heard / seconds,
+           (double)heard * (double)payload_size * 8 / seconds / 1e6, (unsigned long long)sent.packets,
+           (unsigned long long)wrong, (unsigned long long)sent.resent, (unsigned long long)sent.lost,
+           (unsigned long long)sent.late, (unsigned long long)sent.undone, (double)link_ns / 1e3 / (double)done,
+           (double)listen_ns / 1e3 / (double)done, (double)(all_ns - link_ns - listen_ns) / 1e3 / (double)done);
+    if (queued_at && heard) { /* the waits of those handed over */
+        qsort(waited + 1, heard, sizeof(*waited), by_value);
+        printf(",\"waited_ms_p50\":%.3f,\"waited_ms_p99\":%.3f,\"waited_ms_max\":%.3f", (double)waited[1 + heard / 2] / 1e6,
+               (double)waited[1 + heard * 99 / 100] / 1e6, (double)waited[heard] / 1e6);
+    }
+    size_t last = 0;
+    for (size_t b = 0; b < BUCKETS; ++b)
+        if (timeline[b]) last = b;
+    printf(",\"timeline_100ms\":[");
+    for (size_t b = 0; b <= last; ++b) printf("%s%u", b ? "," : "", timeline[b]);
+    printf("]}\n");
     snprintf(path, sizeof(path), "rm -rf %s", dir);
-    return system(path) || wrong ? 1 : 0;
+    return system(path) || wrong || (!limit && heard < count) ? 1 : 0;
 }
diff --git a/research/experiments/020-lossy-paths/run.py b/research/experiments/020-lossy-paths/run.py
new file mode 100644
index 0000000..4b98d6b
--- /dev/null
+++ b/research/experiments/020-lossy-paths/run.py
@@ -0,0 +1,247 @@
+#!/usr/bin/env python3
+"""M6 of docs/design/011-pure-transport.md: a link on a path of delay, loss,
+reordering and a bottleneck, against quinn, and beside TCP CUBIC.
+
+Each case runs in `unshare -rn` (an unprivileged network namespace) whose
+loopback is the path, both ways: MTU 1500, and TCP's segments no larger
+(gso_max_size), so each is a packet; a 50 Mbit/s bottleneck (tbf) whose
+queue holds about a round trip of it; under it netem's delay (half the
+RTT), jitter (packets reorder) and random loss. Pure and quinn send without
+UDP GSO: a qdisc takes a GSO burst as one packet.
+
+  bulk   a backlog of 1 KiB records for --seconds (Pure: a link and a
+         listener, transport.c; quinn: one stream); the goodput from the
+         receiver's records in each 100 ms, the first 2 s left out.
+  paced  50 records a second for --seconds (well within what each carries); the time each took from
+         queued (written) to handed over (read): p50, p99, max.
+  share  Pure, quinn or a second TCP flow beside a TCP CUBIC flow, both
+         for --seconds; each one's share of the goodput from 5 s on.
+
+    python3 research/experiments/020-lossy-paths/run.py [--rounds 3] [--out FILE]
+"""
+import argparse
+import json
+import os
+import shutil
+import socket
+import statistics
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+HERE = Path(__file__).resolve().parent
+QUINN = ROOT / 'research/experiments/019-idle-links/quic/target/release/idle-quic'
+TC = shutil.which('tc') or '/usr/sbin/tc'
+IP = shutil.which('ip') or '/usr/sbin/ip'
+RATE_MBIT, SIZE = 50, 1000
+sys.path.insert(0, str(ROOT / 'tests'))
+
+
+def build(work):
+    """The transport harness, quinn's, the two modules and a certificate."""
+    import check
+    import importlib.util
+    spec = importlib.util.spec_from_file_location('idle_links', ROOT / 'research/experiments/019-idle-links/run.py')
+    idle_links = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(idle_links)
+    subprocess.run(['make', '-s', '-C', str(ROOT), 'build/libpure.a', 'build/libpure-link.a', 'build/pure-crypto.so'],
+                   check=True)
+    subprocess.run(['cc', '-O2', '-g', '-std=c17', '-Wall', '-Wextra', '-Werror', f'-I{ROOT}/core', f'-I{ROOT}/link',
+                    str(ROOT / 'research/experiments/019-idle-links/transport.c'), str(ROOT / 'build/libpure-link.a'),
+                    str(ROOT / 'build/libpure.a'), '-lm', '-pthread', '-ldl', '-o', str(ROOT / 'build/link-transport')],
+                   check=True)
+    subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=QUINN.parents[2], check=True)
+    for name, text in (('sender', check.LINK_SENDER), ('receiver', check.LINK_RECEIVER)):
+        (work / f'{name}.aup').write_text(text)
+        check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), work / f'{name}.aup',
+                             work / f'{name}.pmod')
+    idle_links.make_certificate(work)
+
+
+# ---- Inside the namespace
+
+
+def path(rtt_ms, loss, jitter_ms):
+    """The loopback as the path."""
+    subprocess.run([IP, 'link', 'set', 'lo', 'up', 'mtu', '1500', 'gso_max_size', '1500', 'gso_max_segs', '1'],
+                   check=True)
+    queue = max(30000, RATE_MBIT * 1000000 // 8 * rtt_ms // 1000)  # a round trip of the bottleneck, in bytes
+    subprocess.run([TC, 'qdisc', 'add', 'dev', 'lo', 'root', 'handle', '1:', 'tbf', 'rate', f'{RATE_MBIT}mbit',
+                    'burst', '15000', 'limit', str(queue)], check=True)
+    netem = ['delay', f'{rtt_ms / 2}ms'] + ([f'{jitter_ms}ms'] if jitter_ms else []) + \
+            (['loss', f'{loss}%'] if loss else []) + ['limit', '100000']
+    subprocess.run([TC, 'qdisc', 'add', 'dev', 'lo', 'parent', '1:1', 'handle', '10:', 'netem'] + netem, check=True)
+
+
+def goodput(timeline, record_bytes, skip_s=2.0, until_s=None):
+    """Mbit/s of records from skip_s on (to until_s)."""
+    first = int(skip_s * 10)
+    last = len(timeline) if until_s is None else min(len(timeline), int(until_s * 10))
+    counted = timeline[first:last]
+    return round(sum(counted) * record_bytes * 8 / (len(counted) / 10) / 1e6, 2) if counted else 0.0
+
+
+def pure(work, seconds, rate=0, count=2000000):
+    done = subprocess.run([str(ROOT / 'build/link-transport'), str(work / 'sender.pmod'), str(work / 'receiver.pmod'),
+                           str(count), str(SIZE), '20000', str(rate), str(seconds)], capture_output=True, text=True,
+                          env={**os.environ, 'TRANSPORT_GSO': '0'}, timeout=seconds + 120)
+    assert done.returncode == 0, done.stderr[-2000:]
+    return json.loads(done.stdout)
+
+
+def quinn_server(work):
+    server = subprocess.Popen([str(QUINN), 'serve', str(work / 'cert.der'), str(work / 'key.der')],
+                              stdout=subprocess.PIPE, text=True, env={**os.environ, 'QUIC_GSO': '0'})
+    port = json.loads(server.stdout.readline())['port']
+    return server, port
+
+
+def quinn(work, seconds, rate=0, count=2000000):
+    server, port = quinn_server(work)
+    try:
+        # The bytes of Pure's record: its prefix (count, item's length), the item and the city ("Seoul").
+        client = subprocess.run([str(QUINN), 'drain', str(port), str(count), str(SIZE + 8 + 5), str(rate), str(seconds)],
+                                capture_output=True, text=True, env={**os.environ, 'QUIC_GSO': '0'},
+                                timeout=seconds + 300)
+        assert client.returncode == 0, client.stderr[-2000:]
+        line = server.stdout.readline()
+    finally:
+        server.kill()
+        server.wait()
+    return json.loads(line)
+
+
+class Tcp:
+    """A TCP CUBIC flow: a sender writing as fast as it may for seconds, a
+    receiver counting bytes in each 100 ms from when it starts."""
+
+    def __init__(self, seconds):
+        self.seconds = seconds
+        self.listener = socket.socket()
+        self.listener.bind(('127.0.0.1', 0))
+        self.listener.listen(1)
+        self.timeline = []
+        self.began = None
+
+    def start(self):
+        self.began = time.monotonic()
+        self.receiver = threading.Thread(target=self.receive)
+        self.sender = threading.Thread(target=self.send)
+        self.receiver.start()
+        self.sender.start()
+
+    def receive(self):
+        connection, _ = self.listener.accept()
+        with connection:
+            while True:
+                data = connection.recv(1 << 20)
+                if not data:
+                    break
+                bucket = int((time.monotonic() - self.began) * 10)
+                while len(self.timeline) <= bucket:
+                    self.timeline.append(0)
+                self.timeline[bucket] += len(data)
+
+    def send(self):
+        s = socket.socket()
+        s.setsockopt(socket.IPPROTO_TCP, socket.TCP_CONGESTION, b'cubic')
+        s.connect(self.listener.getsockname())
+        chunk = b'x' * 65536
+        while time.monotonic() - self.began < self.seconds:
+            s.sendall(chunk)
+        s.close()
+
+    def join(self):
+        self.sender.join()
+        self.receiver.join()
+        self.listener.close()
+        return self.timeline
+
+
+def inner(case):
+    """One case, in the namespace: a line of JSON."""
+    work = Path(case['work'])
+    path(case['rtt_ms'], case['loss'], case['jitter_ms'])
+    seconds, kind, who = case['seconds'], case['kind'], case['who']
+    row = dict(case)
+    del row['work']
+    if kind in ('bulk', 'paced'):
+        rate = 50 if kind == 'paced' else 0
+        result = pure(work, seconds, rate) if who == 'pure' else quinn(work, seconds, rate)
+        timeline = result['timeline_100ms']
+        row['mbit'] = goodput(timeline, SIZE + 13, 2.0, seconds)
+        for key in ('waited_ms_p50', 'waited_ms_p99', 'waited_ms_max', 'resent', 'lost', 'late', 'undone', 'packets'):
+            if key in result:
+                row[key] = result[key]
+    else:  # share: who beside a TCP CUBIC flow
+        tcp = Tcp(seconds)
+        tcp.start()
+        if who == 'tcp':
+            other = Tcp(seconds)
+            other.start()
+            theirs = [b / (SIZE + 13) for b in other.join()]
+        else:
+            theirs = (pure(work, seconds) if who == 'pure' else quinn(work, seconds))['timeline_100ms']
+        mine = tcp.join()
+        row['mbit'] = goodput(theirs, SIZE + 13, 5.0, seconds - 1)
+        row['tcp_mbit'] = goodput(mine, 1, 5.0, seconds - 1)
+        total = row['mbit'] + row['tcp_mbit']
+        row['share'] = round(row['mbit'] / total, 3) if total else 0.0
+    return row
+
+
+# ---- Outside
+
+
+def cases(seconds, share_seconds):
+    for rtt in (20, 100):
+        for loss in (0, 1, 5):
+            for jitter in (0, 3):
+                for who in ('pure', 'quinn'):
+                    yield {'kind': 'bulk', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': jitter, 'seconds': seconds}
+        for loss in (0, 1, 5):
+            for who in ('pure', 'quinn'):
+                yield {'kind': 'paced', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': 0, 'seconds': seconds}
+        for who in ('pure', 'quinn', 'tcp'):
+            yield {'kind': 'share', 'who': who, 'rtt_ms': rtt, 'loss': 0, 'jitter_ms': 0, 'seconds': share_seconds}
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--rounds', type=int, default=1)
+    parser.add_argument('--seconds', type=int, default=20)
+    parser.add_argument('--share-seconds', type=int, default=30)
+    parser.add_argument('--only', default=None, help='kinds to run, for example bulk,share')
+    parser.add_argument('--out', type=Path, default=None)
+    parser.add_argument('--inner', default=None, help=argparse.SUPPRESS)
+    args = parser.parse_args()
+    if args.inner:
+        print(json.dumps(inner(json.loads(args.inner))), flush=True)
+        return
+    with tempfile.TemporaryDirectory(prefix='pure-m6-') as directory:
+        work = Path(directory)
+        build(work)
+        only = set(args.only.split(',')) if args.only else None
+        for round_ in range(args.rounds):
+            for case in cases(args.seconds, args.share_seconds):
+                if only and case['kind'] not in only:
+                    continue
+                case = {**case, 'round': round_, 'work': str(work)}
+                done = subprocess.run(['unshare', '-rn', sys.executable, __file__, '--inner', json.dumps(case)],
+                                      capture_output=True, text=True, timeout=case['seconds'] + 600)
+                if done.returncode:
+                    print(json.dumps({**case, 'error': done.stderr[-1500:]}), flush=True)
+                    continue
+                line = done.stdout.strip().splitlines()[-1]
+                print(line, flush=True)
+                if args.out:
+                    with open(args.out, 'a') as out:
+                        out.write(line + '\n')
+
+
+if __name__ == '__main__':
+    main()