diff options
| author | eunhhu <[email protected]> | 2026-09-29 07:44:50 +0100 |
|---|---|---|
| committer | eunhhu <[email protected]> | 2026-09-29 07:44:50 +0100 |
| commit | 6a72b5c8164ceb73994ab231a075cfb0ad1d7192 (patch) | |
| tree | 9769d2d41a82f6b07adf07d4ab27906e6a53e7be | |
| parent | ed123556cf1c10369959fa96e1262050db63f7ce (diff) | |
| download | pure-6a72b5c8164ceb73994ab231a075cfb0ad1d7192.tar.gz pure-6a72b5c8164ceb73994ab231a075cfb0ad1d7192.zip | |
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.
| -rw-r--r-- | link/link.c | 6 | ||||
| -rw-r--r-- | link/link.h | 4 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/quic/src/main.rs | 78 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/transport.c | 80 | ||||
| -rw-r--r-- | research/experiments/020-lossy-paths/run.py | 247 |
5 files changed, 382 insertions, 33 deletions
diff --git a/link/link.c b/link/link.c index 0588633..449ea60 100644 --- a/link/link.c +++ b/link/link.c @@ -112,7 +112,7 @@ struct PureLink { size_t burst_size; unsigned burst_count; uint64_t burst_pn[BURST]; - bool gso; + bool gso, no_gso; }; PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own, @@ -148,6 +148,8 @@ PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, co return l; } +void pure_link_gso(PureLink *l, bool on) { l->no_gso = !on; } + void pure_link_peer(PureLink *l, const unsigned char peer[PURE_SEAL_KEY]) { l->peered = true; l->keyed = pure_link_sender_peer(l->sender, peer); @@ -279,7 +281,7 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e setsockopt(l->fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer)); } int segment = DATAGRAM; - l->gso = ok && !setsockopt(l->fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment)); + l->gso = ok && !l->no_gso && !setsockopt(l->fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment)); ok = ok && l->keyed && (l->burst = malloc((size_t)BURST * DATAGRAM)) && (l->taking = malloc(sizeof(*l->taking))) && (l->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) >= 0; l->notify = notify; diff --git a/link/link.h b/link/link.h index 5237262..afe1430 100644 --- a/link/link.h +++ b/link/link.h @@ -22,6 +22,10 @@ bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_ uint64_t pure_link_queued(PureLink *); /* The last number the receiver saved; any thread. */ uint64_t pure_link_delivered(PureLink *); +/* Packets go in bursts with UDP GSO where the system has it (at first); + * off, each in a call of its own: for a path whose queues take a burst as + * one packet (netem does). Before pure_link_start. */ +void pure_link_gso(PureLink *, bool on); /* Send on a thread of the link's own, calling notify when more was saved. */ bool pure_link_start(PureLink *, void (*notify)(void *), void *context, char *error, size_t size); void pure_link_counts(PureLink *, PureLinkCounts *); 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() |
