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.

2 files changed, 127 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;
 }