From b6db529c6f9f94ba3ec93714a193e6f78b29e9c1 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 2 Nov 2026 12:46:37 -0500 Subject: [PATCH 1/2] joule-moq-relay: a fan-out load test: one publisher, N subscribers, a bystander, the relay's own memory and CPU Co-Authored-By: Claude Opus 5.5 --- .../joule-moq-relay/examples/fanout_load.rs | 295 ++++++++++++++++++ 1 file changed, 295 insertions(+) create mode 100644 joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs new file mode 100644 index 000000000..c77e996de --- /dev/null +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs @@ -0,0 +1,295 @@ +//! A load test of the relay's fan-out: one publisher and N subscribers of one track, over native +//! QUIC on loopback, the relay a process of its own, so what it spends is its own. +//! +//! ```text +//! fanout_load --subscribers N [--payload B] [--rate R] [--seconds S] [--relay-threads T] [--relay-cpus 0-7] +//! [--keyframe-every K --keyframe-bytes B] +//! ``` +//! +//! It prints one JSON object: the relay's resident memory before the subscribers, with them idle +//! and at its peak; its CPU over the run; how many objects arrived of how many sent; and how late. +//! A subscriber's lateness is from the publisher's clock, which is the subscriber's: one host. Run +//! the driver on other CPUs than the relay's (`taskset -c 8-31 fanout_load ...`), so that the clients +//! never take the relay's time. Every wait is bounded, and the relay exits when the driver's end of +//! its standard input closes, so a killed run leaves nothing behind. Linux, for `/proc`. +//! +//! A **bystander** runs beside the audience: a track of its own with one subscriber, at the same +//! rate and with small objects. It shares nothing with the audience but the relay, so its lateness +//! is what the audience's fan-out costs everyone else on the relay. + +use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject}; +use joule_moq::draft22::{Parameters, UniEvent}; +use joule_moq::{TrackName, TrackNamespace}; +use joule_moq_relay::client::{Client, ClientOptions}; +use joule_moq_relay::transport::quic; +use joule_moq_relay::{RelayConfig, RelayServer}; +use std::io::{BufRead, Read, Write}; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use tokio::time::timeout; + +/// Subscribers that share one client UDP socket. +const PER_ENDPOINT: usize = 64; +/// Latency buckets of 50 µs, up to 10 s; the last holds everything later. +const BUCKET_US: u64 = 50; +const BUCKETS: usize = 200_000; +const STEP: Duration = Duration::from_secs(30); + +fn arg(args: &[String], flag: &str, default: T) -> T { + args.iter().position(|a| a == flag).and_then(|i| args.get(i + 1)).and_then(|v| v.parse().ok()).unwrap_or(default) +} + +fn relay_main(args: &[String]) { + let threads: usize = arg(args, "--relay-threads", 8); + let rt = tokio::runtime::Builder::new_multi_thread().worker_threads(threads).enable_all().build().expect("a runtime"); + rt.block_on(async { + let mut c = RelayConfig::loopback(); + c.websocket_listen = None; + c.limits.max_sessions = 1 << 20; + let server = RelayServer::start(c).await.expect("the relay starts"); + let cert: String = server.identity.chain[0].as_ref().iter().map(|b| format!("{b:02x}")).collect(); + println!("{} {cert}", server.quic_addr.expect("a UDP endpoint")); + std::io::stdout().flush().expect("stdout"); + // Until the driver goes. + tokio::task::spawn_blocking(|| { + let mut sink = Vec::new(); + let _ = std::io::stdin().read_to_end(&mut sink); + }) + .await + .expect("the wait"); + std::process::exit(0); + }); +} + +fn proc_kb(pid: u32, field: &str) -> u64 { + std::fs::read_to_string(format!("/proc/{pid}/status")).ok().and_then(|s| s.lines().find(|l| l.starts_with(field)).and_then(|l| l.split_whitespace().nth(1)?.parse().ok())).unwrap_or(0) +} + +/// The process's user and system CPU, in seconds. +fn proc_cpu(pid: u32) -> f64 { + let Ok(s) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { return 0.0 }; + let after = s.rsplit(')').next().unwrap_or(""); + let f: Vec<&str> = after.split_whitespace().collect(); + let t = |i: usize| f.get(i).and_then(|v| v.parse::().ok()).unwrap_or(0); + (t(11) + t(12)) as f64 / 100.0 +} + +fn now_us() -> u64 { + SystemTime::now().duration_since(UNIX_EPOCH).expect("after 1970").as_micros() as u64 +} + +fn spawn_relay(args: &[String]) -> (Child, std::net::SocketAddr, Vec) { + let exe = std::env::current_exe().expect("this binary"); + let threads: usize = arg(args, "--relay-threads", 8); + let cpus: String = arg(args, "--relay-cpus", "0-7".to_string()); + let mut child = Command::new("taskset") + .args(["-c", &cpus]) + .arg(&exe) + .args(["relay", "--relay-threads", &threads.to_string()]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .expect("taskset runs"); + let mut line = String::new(); + std::io::BufReader::new(child.stdout.take().expect("its stdout")).read_line(&mut line).expect("its address"); + let (addr, cert) = line.trim().split_once(' ').expect("an address and a certificate"); + let cert = (0..cert.len() / 2).map(|i| u8::from_str_radix(&cert[2 * i..2 * i + 2], 16).expect("hex")).collect(); + (child, addr.parse().expect("an address"), cert) +} + +struct Seen { + objects: AtomicU64, + buckets: Vec, +} + +impl Seen { + fn note(&self, late_us: u64) { + self.objects.fetch_add(1, Ordering::Relaxed); + self.buckets[((late_us / BUCKET_US) as usize).min(BUCKETS - 1)].fetch_add(1, Ordering::Relaxed); + } + + /// The latency, in milliseconds, at each fraction. + fn at(&self, fractions: &[f64]) -> Vec { + let counts: Vec = self.buckets.iter().map(|b| b.load(Ordering::Relaxed)).collect(); + let total: u64 = counts.iter().sum(); + fractions + .iter() + .map(|f| { + let want = ((total as f64) * f).ceil() as u64; + let mut sum = 0; + for (i, c) in counts.iter().enumerate() { + sum += c; + if sum >= want.max(1) { + return (i as u64 * BUCKET_US) as f64 / 1000.0; + } + } + 0.0 + }) + .collect() + } +} + +fn driver(args: &[String]) { + let subscribers: usize = arg(args, "--subscribers", 100); + let payload: usize = arg(args, "--payload", 8192); + let rate: u64 = arg(args, "--rate", 30); + let seconds: u64 = arg(args, "--seconds", 20); + let keyframe_every: u64 = arg(args, "--keyframe-every", 0); + let keyframe_bytes: usize = arg(args, "--keyframe-bytes", 0); + let (child, addr, cert) = spawn_relay(args); + let pid = child.id(); + let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("a runtime"); + rt.block_on(async move { + tokio::time::sleep(Duration::from_millis(300)).await; + let cert = rustls::pki_types::CertificateDer::from(cert); + let alpn = joule_moq::draft22::ALPN.to_vec(); + let bind: std::net::SocketAddr = "127.0.0.1:0".parse().expect("an address"); + let (space, track) = (TrackNamespace::from_strs(["load"]), TrackName::from_str("video")); + let base_kb = proc_kb(pid, "VmRSS"); + + let pub_ep = quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint"); + let conn = timeout(STEP, quic::connect(&pub_ep, addr, "localhost")).await.expect("in time").expect("a session"); + let publisher = Arc::new(Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP")); + let _published = publisher.publish(space.clone(), track.clone(), 1, Parameters::new(), Default::default()).await.expect("PUBLISH"); + + let seen = Arc::new(Seen { objects: AtomicU64::new(0), buckets: (0..BUCKETS).map(|_| AtomicU64::new(0)).collect() }); + let endpoints: Vec = (0..subscribers.div_ceil(PER_ENDPOINT)).map(|_| quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint")).collect(); + let connecting = Instant::now(); + let gate = Arc::new(tokio::sync::Semaphore::new(128)); + let mut joins = Vec::with_capacity(subscribers); + for i in 0..subscribers { + let ep = endpoints[i / PER_ENDPOINT].clone(); + let (gate, space, track, seen) = (gate.clone(), space.clone(), track.clone(), seen.clone()); + joins.push(tokio::spawn(async move { + let permit = gate.acquire_owned().await.expect("the gate"); + let conn = timeout(STEP, quic::connect(&ep, addr, "localhost")).await.expect("in time").expect("a session"); + let c = Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP"); + let held = c.subscribe(space, track, Parameters::new()).await.expect("SUBSCRIBE"); + drop(permit); + tokio::spawn(async move { + let _held = held; + while let Some(mut r) = c.next_data(Duration::from_secs(3600)).await { + while let Ok(Some(e)) = r.next().await { + if let UniEvent::SubgroupObject(o) = e { + let sent = u64::from_le_bytes(o.payload[..8].try_into().expect("a timestamp")); + seen.note(now_us().saturating_sub(sent)); + } + } + } + }); + })); + } + for j in joins { + j.await.expect("a subscriber"); + } + let connect_s = connecting.elapsed().as_secs_f64(); + + // The bystander: a publisher and a subscriber of their own track. + let aside = Arc::new(Seen { objects: AtomicU64::new(0), buckets: (0..BUCKETS).map(|_| AtomicU64::new(0)).collect() }); + let by_ep = quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint"); + let by_name = TrackName::from_str("bystander"); + let conn = timeout(STEP, quic::connect(&by_ep, addr, "localhost")).await.expect("in time").expect("a session"); + let by_pub = Arc::new(Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP")); + let _by_published = by_pub.publish(space.clone(), by_name.clone(), 2, Parameters::new(), Default::default()).await.expect("PUBLISH"); + let conn = timeout(STEP, quic::connect(&by_ep, addr, "localhost")).await.expect("in time").expect("a session"); + let by_sub = Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP"); + let by_held = by_sub.subscribe(space.clone(), by_name, Parameters::new()).await.expect("SUBSCRIBE"); + let by_seen = aside.clone(); + tokio::spawn(async move { + let _held = by_held; + while let Some(mut r) = by_sub.next_data(Duration::from_secs(3600)).await { + while let Ok(Some(e)) = r.next().await { + if let UniEvent::SubgroupObject(o) = e { + by_seen.note(now_us().saturating_sub(u64::from_le_bytes(o.payload[..8].try_into().expect("a timestamp")))); + } + } + } + }); + tokio::time::sleep(Duration::from_secs(3)).await; + let idle_kb = proc_kb(pid, "VmRSS"); + let idle_cpu_start = proc_cpu(pid); + tokio::time::sleep(Duration::from_secs(3)).await; + let idle_cores = (proc_cpu(pid) - idle_cpu_start) / 3.0; + + // The run: one object to a group, at the rate, each stamped with when it left. + let cpu_start = proc_cpu(pid); + let driver_start = proc_cpu(std::process::id()); + let started = Instant::now(); + let by_tick = Duration::from_nanos(1_000_000_000 / rate); + let by_run = Duration::from_secs(seconds); + let by_sender = tokio::spawn(async move { + let mut tick = tokio::time::interval(by_tick); + let (t0, mut g) = (Instant::now(), 0u64); + while t0.elapsed() < by_run { + tick.tick().await; + let mut b = vec![0u8; 1024]; + b[..8].copy_from_slice(&now_us().to_le_bytes()); + let h = SubgroupHeader { track_alias: 2, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true }; + let _ = timeout(STEP, by_pub.send_subgroup(h, &[SubgroupObject::normal(0, b)])).await; + g += 1; + } + }); + let mut tick = tokio::time::interval(Duration::from_nanos(1_000_000_000 / rate)); + let body = vec![0x5au8; payload.max(8)]; + let mut sent = 0u64; + let mut lagging = 0u64; + let mut peak_kb = 0; + while started.elapsed() < Duration::from_secs(seconds) { + tick.tick().await; + let mut b = if keyframe_every != 0 && sent % keyframe_every == 0 { vec![0x5au8; keyframe_bytes.max(8)] } else { body.clone() }; + b[..8].copy_from_slice(&now_us().to_le_bytes()); + let h = SubgroupHeader { track_alias: 1, group_id: sent, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true }; + let t0 = Instant::now(); + timeout(STEP, publisher.send_subgroup(h, &[SubgroupObject::normal(0, b)])).await.expect("in time").expect("a send"); + if t0.elapsed() > Duration::from_nanos(1_000_000_000 / rate) { + lagging += 1; + } + sent += 1; + if sent % rate == 0 { + peak_kb = peak_kb.max(proc_kb(pid, "VmRSS")); + } + } + let run_s = started.elapsed().as_secs_f64(); + let run_cpu = proc_cpu(pid) - cpu_start; + let driver_cores = (proc_cpu(std::process::id()) - driver_start) / run_s; + let _ = by_sender.await; + tokio::time::sleep(Duration::from_secs(3)).await; + let got = seen.objects.load(Ordering::Relaxed); + let want = sent * subscribers as u64; + let l = seen.at(&[0.5, 0.9, 0.99, 0.999]); + let by = aside.at(&[0.5, 0.99, 0.999]); + let probe = std::fs::read_to_string(std::env::var("FANOUT_PROBE_FILE").unwrap_or_default()).unwrap_or_default(); + println!( + "{{\"subscribers\":{subscribers},\"payload\":{payload},\"rate\":{rate},\"seconds\":{seconds},\"connect_s\":{connect_s:.1},\"rss_base_mb\":{:.1},\"rss_idle_mb\":{:.1},\"kb_per_idle_subscriber\":{:.1},\"idle_cores\":{idle_cores:.3},\"rss_peak_mb\":{:.1},\"relay_cores\":{:.2},\"cpu_us_per_delivery\":{:.1},\"sent\":{sent},\"send_lagged\":{lagging},\"delivered\":{got},\"delivered_ratio\":{:.4},\"gbps\":{:.2},\"p50_ms\":{:.2},\"p90_ms\":{:.2},\"p99_ms\":{:.2},\"p999_ms\":{:.2},\"driver_cores\":{driver_cores:.2},\"bystander_p50_ms\":{:.2},\"bystander_p99_ms\":{:.2},\"bystander_p999_ms\":{:.2},\"probe\":\"{}\"}}", + base_kb as f64 / 1024.0, + idle_kb as f64 / 1024.0, + idle_kb.saturating_sub(base_kb) as f64 / subscribers.max(1) as f64, + peak_kb.max(proc_kb(pid, "VmHWM")) as f64 / 1024.0, + run_cpu / run_s, + run_cpu * 1e6 / got.max(1) as f64, + got as f64 / want.max(1) as f64, + got as f64 * payload as f64 * 8.0 / run_s / 1e9, + l[0], + l[1], + l[2], + l[3], + by[0], + by[1], + by[2], + probe.trim(), + ); + std::process::exit(0); + }); +} + +fn main() { + let args: Vec = std::env::args().skip(1).collect(); + if args.first().map(String::as_str) == Some("relay") { + relay_main(&args); + } else { + driver(&args); + } +} -- 2.54.0 (Apple Git-157) From 88ff31680567f58ef1f6db3064052053b03024a1 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 2 Nov 2026 13:26:10 -0500 Subject: [PATCH 2/2] joule-moq-relay: an object is held once, and the socket keeps what arrives Under load the relay collapsed with cores to spare, for two reasons a load test showed. Its one UDP socket kept the kernel's default buffer, about 212 KB. Each fan-out's acknowledgements arrive together, so at 2,000 viewers of 2 Mbit/s the kernel dropped about a million packets in 20 seconds. A dropped acknowledgement reads as loss: a retransmission and a smaller congestion window. The buffer is now a setting, 4 MiB by default, for the sockets the relay listens on and dials from. The system caps it, so the relay warns when it gets less. Each subscriber also got its own copy of every payload, while the fan-out held the relay's one lock, and two more copies on the way to the wire. An object is now held once, from when it is read until the last QUIC stream has it acknowledged. The cache, every session's queue and every QUIC send buffer share one allocation, and a keyframe stops being N keyframes in memory. A transport that keeps no buffers still copies, once, where it writes, and a bridge to an earlier draft still gets an object's head and payload as one write. `examples/fanout_load.rs` is the load test. A bystander on its own track measures what the audience costs everyone else on the relay. Co-Authored-By: Claude Opus 5.5 --- joule-code/CHANGELOG.md | 15 +++++ .../crates/joule-moq-relay/Cargo.toml | 2 +- .../crates/joule-moq-relay/README.md | 12 ++++ .../joule-moq-relay/examples/fanout_load.rs | 41 +++++++++--- .../crates/joule-moq-relay/relay.example.toml | 1 + .../crates/joule-moq-relay/src/cascade.rs | 2 +- .../crates/joule-moq-relay/src/config.rs | 11 ++++ .../crates/joule-moq-relay/src/relay.rs | 13 ++-- .../crates/joule-moq-relay/src/sched.rs | 62 ++++++++++++++----- .../crates/joule-moq-relay/src/server.rs | 4 +- .../crates/joule-moq-relay/src/session.rs | 3 +- .../joule-moq-relay/src/transport/mod.rs | 46 ++++++++++++++ .../joule-moq-relay/src/transport/quic.rs | 47 +++++++++++++- .../crates/joule-moq/src/draft22/data.rs | 23 ++++++- 14 files changed, 244 insertions(+), 38 deletions(-) diff --git a/joule-code/CHANGELOG.md b/joule-code/CHANGELOG.md index acfbea7a7..ac3081698 100644 --- a/joule-code/CHANGELOG.md +++ b/joule-code/CHANGELOG.md @@ -4,6 +4,21 @@ All notable changes to the Joule Code standard are recorded here. The format fol ## [Unreleased] +### Changed — `joule-moq-relay`: one object held once, and a socket that keeps what arrives + +- **`limits.quic_socket_buffer_bytes`** (4 MiB by default, and for the links a relay dials too): the + size asked of the kernel for the UDP socket's buffers. Each fan-out's acknowledgements arrive + together, and past the buffer the kernel drops them, which reads as loss. The system caps the + size (`net.core.rmem_max` on Linux), and the relay logs a warning when it gets less. +- **An object is held once**, from when it is read until the last subscriber's QUIC stream has it + acknowledged: the cache, each session's queue and each QUIC send buffer share one allocation + (`SendStream::write_parts`, `SubgroupCodec::encode_object_head`). The fan-out no longer copies a + payload per subscriber while it holds the relay's lock. A WebTransport or WebSocket session, and + a session bridged to an earlier draft, still has its own copy in its transport. +- **`examples/fanout_load.rs`**: one publisher, N subscribers of one track and a bystander on its + own, reporting the relay's memory and CPU, what arrived and how late, and the packets the kernel + dropped at each end. + ### Added — `joule-moq-relay`: relay-to-relay cascading - **`[[upstream]]`**: relays this one dials, over native QUIC (`moqt://`), WebTransport diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/Cargo.toml b/joule-code/joule-code-rs/crates/joule-moq-relay/Cargo.toml index 4c8c21442..adb8e5c22 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/Cargo.toml +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/Cargo.toml @@ -32,7 +32,7 @@ tokio-rustls = { version = "0.26", default-features = false, features = [" # The platform's roots, for an upstream whose front holds a public certificate. rustls-native-certs = "0.8" tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time", "net", "io-util", "signal"] } -bytes = "1" +bytes = "1.9" futures = "0.3" async-trait = "0.1" tracing = "0.1" diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/README.md b/joule-code/joule-code-rs/crates/joule-moq-relay/README.md index e6873d31f..7c3a26b82 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/README.md +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/README.md @@ -326,6 +326,18 @@ not verify. unaltered and nothing is re-signed. A refused object ends its downstream streams rather than leaving a hole in them. +## Load + +`examples/fanout_load.rs` measures the fan-out: one publisher, N subscribers of one track over +native QUIC, and a bystander on a track of its own, whose lateness is what the audience costs +everyone else on the relay. It reports the relay's resident memory and CPU, how many objects +arrived, how late, and the packets the kernel dropped at each end. A deployment sets +`net.core.rmem_max` to at least `limits.quic_socket_buffer_bytes` (4 MiB by default): the relay +asks for that, Linux quietly gives no more than its limit, and the relay logs a warning. An object +is held once from when it is read until the last subscriber's QUIC stream has it acknowledged: the +cache, each session's queue and each QUIC send buffer share one allocation. A WebTransport or +WebSocket session, and a session bridged to an earlier draft, has its own copy in its transport. + ## Tokens and room grants `[[auth.grants]]` rules name a token by the SHA-256 of its value (Token Type 0), a diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs index c77e996de..20f3a6077 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs @@ -3,7 +3,7 @@ //! //! ```text //! fanout_load --subscribers N [--payload B] [--rate R] [--seconds S] [--relay-threads T] [--relay-cpus 0-7] -//! [--keyframe-every K --keyframe-bytes B] +//! [--keyframe-every K --keyframe-bytes B] [--per-endpoint E] //! ``` //! //! It prints one JSON object: the relay's resident memory before the subscribers, with them idle @@ -30,8 +30,6 @@ use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::time::timeout; -/// Subscribers that share one client UDP socket. -const PER_ENDPOINT: usize = 64; /// Latency buckets of 50 µs, up to 10 s; the last holds everything later. const BUCKET_US: u64 = 50; const BUCKETS: usize = 200_000; @@ -76,6 +74,26 @@ fn proc_cpu(pid: u32) -> f64 { (t(11) + t(12)) as f64 / 100.0 } +/// The packets the kernel dropped on UDP sockets on loopback: the relay's, at `port`, and the rest. +fn udp_drops(port: u16) -> (u64, u64) { + let Ok(s) = std::fs::read_to_string("/proc/net/udp") else { return (0, 0) }; + let (mut relay, mut others) = (0, 0); + for l in s.lines().skip(1) { + let f: Vec<&str> = l.split_whitespace().collect(); + let (Some(local), Some(drops)) = (f.get(1), f.last().and_then(|d| d.parse::().ok())) else { continue }; + let Some((host, p)) = local.split_once(':') else { continue }; + if host != "0100007F" { + continue; + } + if u16::from_str_radix(p, 16) == Ok(port) { + relay += drops; + } else { + others += drops; + } + } + (relay, others) +} + fn now_us() -> u64 { SystemTime::now().duration_since(UNIX_EPOCH).expect("after 1970").as_micros() as u64 } @@ -139,6 +157,7 @@ fn driver(args: &[String]) { let seconds: u64 = arg(args, "--seconds", 20); let keyframe_every: u64 = arg(args, "--keyframe-every", 0); let keyframe_bytes: usize = arg(args, "--keyframe-bytes", 0); + let per_endpoint: usize = arg(args, "--per-endpoint", 8); let (child, addr, cert) = spawn_relay(args); let pid = child.id(); let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("a runtime"); @@ -156,12 +175,12 @@ fn driver(args: &[String]) { let _published = publisher.publish(space.clone(), track.clone(), 1, Parameters::new(), Default::default()).await.expect("PUBLISH"); let seen = Arc::new(Seen { objects: AtomicU64::new(0), buckets: (0..BUCKETS).map(|_| AtomicU64::new(0)).collect() }); - let endpoints: Vec = (0..subscribers.div_ceil(PER_ENDPOINT)).map(|_| quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint")).collect(); + let endpoints: Vec = (0..subscribers.div_ceil(per_endpoint)).map(|_| quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint")).collect(); let connecting = Instant::now(); let gate = Arc::new(tokio::sync::Semaphore::new(128)); let mut joins = Vec::with_capacity(subscribers); for i in 0..subscribers { - let ep = endpoints[i / PER_ENDPOINT].clone(); + let ep = endpoints[i / per_endpoint].clone(); let (gate, space, track, seen) = (gate.clone(), space.clone(), track.clone(), seen.clone()); joins.push(tokio::spawn(async move { let permit = gate.acquire_owned().await.expect("the gate"); @@ -215,6 +234,7 @@ fn driver(args: &[String]) { let idle_cores = (proc_cpu(pid) - idle_cpu_start) / 3.0; // The run: one object to a group, at the rate, each stamped with when it left. + let drops_start = udp_drops(addr.port()); let cpu_start = proc_cpu(pid); let driver_start = proc_cpu(std::process::id()); let started = Instant::now(); @@ -239,7 +259,7 @@ fn driver(args: &[String]) { let mut peak_kb = 0; while started.elapsed() < Duration::from_secs(seconds) { tick.tick().await; - let mut b = if keyframe_every != 0 && sent % keyframe_every == 0 { vec![0x5au8; keyframe_bytes.max(8)] } else { body.clone() }; + let mut b = if keyframe_every != 0 && sent.is_multiple_of(keyframe_every) { vec![0x5au8; keyframe_bytes.max(8)] } else { body.clone() }; b[..8].copy_from_slice(&now_us().to_le_bytes()); let h = SubgroupHeader { track_alias: 1, group_id: sent, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true }; let t0 = Instant::now(); @@ -248,22 +268,22 @@ fn driver(args: &[String]) { lagging += 1; } sent += 1; - if sent % rate == 0 { + if sent.is_multiple_of(rate) { peak_kb = peak_kb.max(proc_kb(pid, "VmRSS")); } } let run_s = started.elapsed().as_secs_f64(); let run_cpu = proc_cpu(pid) - cpu_start; let driver_cores = (proc_cpu(std::process::id()) - driver_start) / run_s; + let drops_end = udp_drops(addr.port()); let _ = by_sender.await; tokio::time::sleep(Duration::from_secs(3)).await; let got = seen.objects.load(Ordering::Relaxed); let want = sent * subscribers as u64; let l = seen.at(&[0.5, 0.9, 0.99, 0.999]); let by = aside.at(&[0.5, 0.99, 0.999]); - let probe = std::fs::read_to_string(std::env::var("FANOUT_PROBE_FILE").unwrap_or_default()).unwrap_or_default(); println!( - "{{\"subscribers\":{subscribers},\"payload\":{payload},\"rate\":{rate},\"seconds\":{seconds},\"connect_s\":{connect_s:.1},\"rss_base_mb\":{:.1},\"rss_idle_mb\":{:.1},\"kb_per_idle_subscriber\":{:.1},\"idle_cores\":{idle_cores:.3},\"rss_peak_mb\":{:.1},\"relay_cores\":{:.2},\"cpu_us_per_delivery\":{:.1},\"sent\":{sent},\"send_lagged\":{lagging},\"delivered\":{got},\"delivered_ratio\":{:.4},\"gbps\":{:.2},\"p50_ms\":{:.2},\"p90_ms\":{:.2},\"p99_ms\":{:.2},\"p999_ms\":{:.2},\"driver_cores\":{driver_cores:.2},\"bystander_p50_ms\":{:.2},\"bystander_p99_ms\":{:.2},\"bystander_p999_ms\":{:.2},\"probe\":\"{}\"}}", + "{{\"subscribers\":{subscribers},\"payload\":{payload},\"rate\":{rate},\"seconds\":{seconds},\"connect_s\":{connect_s:.1},\"rss_base_mb\":{:.1},\"rss_idle_mb\":{:.1},\"kb_per_idle_subscriber\":{:.1},\"idle_cores\":{idle_cores:.3},\"rss_peak_mb\":{:.1},\"relay_cores\":{:.2},\"cpu_us_per_delivery\":{:.1},\"sent\":{sent},\"send_lagged\":{lagging},\"delivered\":{got},\"delivered_ratio\":{:.4},\"gbps\":{:.2},\"p50_ms\":{:.2},\"p90_ms\":{:.2},\"p99_ms\":{:.2},\"p999_ms\":{:.2},\"driver_cores\":{driver_cores:.2},\"udp_drops_relay\":{},\"udp_drops_clients\":{},\"bystander_p50_ms\":{:.2},\"bystander_p99_ms\":{:.2},\"bystander_p999_ms\":{:.2}}}", base_kb as f64 / 1024.0, idle_kb as f64 / 1024.0, idle_kb.saturating_sub(base_kb) as f64 / subscribers.max(1) as f64, @@ -276,10 +296,11 @@ fn driver(args: &[String]) { l[1], l[2], l[3], + drops_end.0.saturating_sub(drops_start.0), + drops_end.1.saturating_sub(drops_start.1), by[0], by[1], by[2], - probe.trim(), ); std::process::exit(0); }); diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml b/joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml index 6dd13c04a..5d6ff354e 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml @@ -47,6 +47,7 @@ max_relay_hops = 8 upstream_reconnect_ms = 1000 websocket_buffer_bytes = 67108864 idle_timeout_ms = 30000 +quic_socket_buffer_bytes = 4194304 # the UDP socket's buffers; the system caps them (net.core.rmem_max on Linux) quic_send_window_bytes = 1048576 quic_stream_window_bytes = 16777216 # a stream's flow-control window: more than the largest group a door holds quic_connection_window_bytes = 67108864 # a session's, all its streams together: what bounds the bytes it leaves unread; never below the stream window diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/cascade.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/cascade.rs index bde8c5b6a..85d669fef 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/cascade.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/cascade.rs @@ -349,7 +349,7 @@ async fn dial(relay: &Relay, link: &Link, url: &str) -> Result<(SharedConnection // among it: an upstream can leave this relay holding no more unread than a session // it accepted can. let t = crate::server::transport_config(cfg); - let ep = crate::transport::quic::client_endpoint_tls(bind, (*tls).clone(), Some(t)).map_err(|e| e.to_string())?; + let ep = crate::transport::quic::client_endpoint_tls(bind, (*tls).clone(), Some(t), cfg.limits.quic_socket_buffer_bytes).map_err(|e| e.to_string())?; let conn = tokio::time::timeout(step, crate::transport::quic::connect(&ep, addr, &host)).await.map_err(|_| "no QUIC session in time".to_string())?.map_err(|e| e.to_string())?; Ok((Arc::new(conn), Some(ep))) } diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs index 1447d7268..c2070e86c 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs @@ -245,6 +245,13 @@ pub struct LimitsConfig { /// QUIC idle timeout. #[serde(default = "d_idle")] pub idle_timeout_ms: u64, + /// Bytes the kernel may hold for the relay's UDP socket in each direction, or 0 for the + /// system's default. The system caps it (`net.core.rmem_max` on Linux, often 212992), and the + /// relay logs a warning when it is given less than this. Each fan-out's acknowledgements arrive + /// together: at 2,000 sessions of 2 Mbit/s on loopback, a buffer of 212992 dropped about a million + /// packets in 20 s, and 4 MiB dropped none at that steady load. + #[serde(default = "d_socket_buffer")] + pub quic_socket_buffer_bytes: usize, /// Bytes a QUIC or WebTransport session may have sent and unacknowledged, or /// buffered, before a write waits. Kept near the path's bandwidth-delay product, so /// what waits waits in the relay's scheduler, where priority and delivery timeouts @@ -332,6 +339,9 @@ fn d_upstream() -> u64 { fn d_fill() -> u64 { 2_000 } +fn d_socket_buffer() -> usize { + 4 << 20 +} fn d_writes() -> usize { 8 } @@ -380,6 +390,7 @@ impl Default for LimitsConfig { goaway_timeout_ms: d_goaway(), websocket_buffer_bytes: d_ws_buffer(), idle_timeout_ms: d_idle(), + quic_socket_buffer_bytes: d_socket_buffer(), quic_send_window_bytes: d_send_window(), quic_stream_window_bytes: d_stream_window(), quic_connection_window_bytes: d_connection_window(), diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs index db66b66d7..9d29c7cdb 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs @@ -441,7 +441,8 @@ pub struct ObjectIn<'a> { pub header: &'a SubgroupHeader, pub subgroup_id: u64, pub first_in_stream: Option, - pub object: &'a SubgroupObject, + /// The object, held once: the cache, each subscriber's queue and each QUIC send buffer share it. + pub object: Arc, pub arrived: Instant, pub object_timeout: u64, pub signed: Option<(u64, &'static str)>, @@ -875,6 +876,7 @@ impl Relay { /// stream is the one the relay takes its subgroup from (§7.3): the cache counts it /// open from its first object, and [`Relay::cache_stream_ended`] closes the count. pub fn fan_out_object(&self, key: &TrackKey, o: ObjectIn<'_>, started: &mut HashSet) -> bool { + let payload = crate::sched::shared_payload(&o.object); let mut st = self.inner.state.lock().unwrap(); let receipts_key = self.receipts_of(key); let is_receipts = self.is_receipts_track(&key.1); @@ -914,7 +916,7 @@ impl Relay { publisher_priority: priority, properties: o.object.properties.clone(), status: o.object.status, - payload: Bytes::from(o.object.payload.clone()), + payload: payload.clone(), }; let size = cache::size_of(&cached); let ins = track.cache.insert(group, oid, cached); @@ -958,7 +960,7 @@ impl Relay { }; sub.sched.push( LaneKey { sub: sub.id, kind: LaneKind::Subgroup { group, subgroup: o.subgroup_id } }, - Item { op: Op::Object { header, object: o.object.clone() }, prio, deadline, meta: Some(meta) }, + Item { op: Op::Object { header, object: o.object.clone(), payload: payload.clone() }, prio, deadline, meta: Some(meta) }, ); } if matches!(o.object.status, joule_moq::draft22::data::ObjectStatus::EndOfGroup | joule_moq::draft22::data::ObjectStatus::EndOfTrack) { @@ -1003,6 +1005,7 @@ impl Relay { /// One object datagram from upstream (§11.2). pub fn fan_out_datagram(&self, key: &TrackKey, upstream: u64, d: &ObjectDatagram, arrived: Instant, signed: Option<(u64, &'static str)>) { use joule_moq::draft22::data::DatagramContent; + let shared = Arc::new(d.clone()); let mut st = self.inner.state.lock().unwrap(); let receipts_key = self.receipts_of(key); let rsessions: HashSet = st.tracks.get(&receipts_key).map(|t| t.downs.iter().map(|d| d.session).collect()).unwrap_or_default(); @@ -1041,11 +1044,9 @@ impl Relay { // For a datagram SUBGROUP_DELIVERY_TIMEOUT acts as OBJECT_DELIVERY_TIMEOUT (§5.2). let t = effective_timeout(object_timeout, effective_timeout(p.object_timeout, p.subgroup_timeout)); let deadline = (t != 0 && !sub.receipts_track).then(|| arrived + Duration::from_millis(t)); - let mut out = d.clone(); - out.track_alias = sub.alias; let prio = Prio { subscriber: p.subscriber_priority, publisher: priority, group_rank: Prio::group_rank(p.order, d.group_id), fill_rank: 1, mode_rank: 0, within: d.object_id }; let meta = ObjMeta { track: sub.render.clone(), payload_len: payload.len() as u64, object_id: d.object_id, receipts_subscribed: rsessions.contains(&sub.session), fetched: false, dvr: false, signed }; - sub.sched.push(LaneKey { sub: sub.id, kind: LaneKind::Datagram { group: d.group_id, object: d.object_id } }, Item { op: Op::Datagram { datagram: out }, prio, deadline, meta: Some(meta) }); + sub.sched.push(LaneKey { sub: sub.id, kind: LaneKind::Datagram { group: d.group_id, object: d.object_id } }, Item { op: Op::Datagram { datagram: shared.clone(), alias: sub.alias }, prio, deadline, meta: Some(meta) }); } self.charge_cache(&mut st, track_id, d.group_id, size); } diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs index 50742faa1..720e638cb 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs @@ -30,6 +30,7 @@ //! Receipts tracks carry no deadline (WAI moq-streaming-format §4.1), so the receipts of //! an object dropped here still reach a subscriber of its receipts track. +use bytes::Bytes; use crate::summary::Tally; use crate::transport::{BoxSend, SharedConnection, TransportError}; use joule_moq::draft22::data::{FetchCodec, FetchEntry, FetchHeader, ObjectDatagram, SubgroupCodec, SubgroupHeader, SubgroupObject}; @@ -106,13 +107,16 @@ pub struct ObjMeta { } pub enum Op { - Object { header: SubgroupHeader, object: SubgroupObject }, + /// `payload` is the object's payload, shared ([`shared_payload`]), so a session's queue and its + /// transport hold no copy of it. + Object { header: SubgroupHeader, object: Arc, payload: Bytes }, /// The subgroup is complete upstream: FIN the stream, then hold it to /// SUBGROUP_DELIVERY_TIMEOUT. Fin { subgroup_timeout: Option }, /// The upstream stream was reset: reset this one. Reset { code: u64 }, - Datagram { datagram: ObjectDatagram }, + /// One datagram of the upstream's, shared, and this subscription's Track Alias for it. + Datagram { datagram: Arc, alias: u64 }, FetchEntry { request_id: u64, order: GroupOrder, entry: FetchEntry }, FetchFin { request_id: u64, order: GroupOrder }, /// End a fetch stream abruptly, opening it first if it is not (a fill that fails, @@ -120,6 +124,18 @@ pub enum Op { FetchReset { request_id: u64, order: GroupOrder, code: u64 }, } +/// An object's payload as a buffer that keeps the object alive: every subscriber's queue and every +/// QUIC send buffer that takes it share the one allocation. +pub fn shared_payload(object: &Arc) -> Bytes { + struct Payload(Arc); + impl AsRef<[u8]> for Payload { + fn as_ref(&self) -> &[u8] { + &self.0.payload + } + } + Bytes::from_owner(Payload(object.clone())) +} + pub struct Item { pub op: Op, pub prio: Prio, @@ -234,7 +250,7 @@ enum Then { struct Job { lane: LaneKey, send: Option, - bytes: Vec, + parts: Vec, then: Then, deadline: Option, priority: i32, @@ -438,9 +454,9 @@ impl Sched { let conn = self.conn.clone(); let done = self.done_tx.clone(); tokio::spawn(async move { - let Job { lane, send, bytes, then, deadline, priority, meta, track } = job; + let Job { lane, send, parts, then, deadline, priority, meta, track } = job; let mut send = send; - let n = bytes.len() as u64; + let n = parts.iter().map(|p| p.len() as u64).sum::(); // Opening the stream (which can wait for the peer's stream credit) and the // write are both bounded by the object's deadline. let work = async { @@ -450,8 +466,8 @@ impl Sched { send = Some(s); } let s = send.as_mut().expect("a stream"); - if !bytes.is_empty() { - s.write_all(&bytes).await?; + if n != 0 { + s.write_parts(parts).await?; } Ok::<(), TransportError>(()) }; @@ -526,7 +542,7 @@ impl Sched { } let priority = prio.transport_priority(); match op { - Op::Object { header, object } => { + Op::Object { header, object, payload } => { if expired(deadline) { if let Some(m) = &meta { self.count_dropped(m); @@ -547,7 +563,7 @@ impl Sched { self.subs.entry(key.sub).or_default().streams_opened += 1; } let lane = self.lanes.get_mut(&key).expect("the lane"); - let enc = lane.subgroup.as_mut().expect("a subgroup codec").encode_object(&object, &mut bytes); + let enc = lane.subgroup.as_mut().expect("a subgroup codec").encode_object_head(&object, &mut bytes); if enc.is_err() { self.settle(key.sub, 1); self.kill(key, StreamResetCode::INTERNAL_ERROR.0); @@ -555,7 +571,8 @@ impl Sched { } lane.busy = true; let send = lane.send.take(); - self.spawn_job(Job { lane: key, send, bytes, then: Then::Nothing, deadline, priority, meta, track: None }); + let parts = if payload.is_empty() { vec![bytes.into()] } else { vec![bytes.into(), payload] }; + self.spawn_job(Job { lane: key, send, parts, then: Then::Nothing, deadline, priority, meta, track: None }); } Op::Fin { subgroup_timeout } => { if !lane.opened { @@ -566,7 +583,7 @@ impl Sched { lane.busy = true; let send = lane.send.take(); let track = lane.track.clone(); - self.spawn_job(Job { lane: key, send, bytes: vec![], then: Then::Finish(subgroup_timeout), deadline: None, priority, meta: None, track }); + self.spawn_job(Job { lane: key, send, parts: vec![], then: Then::Finish(subgroup_timeout), deadline: None, priority, meta: None, track }); } Op::Reset { code } => { if let Some(s) = lane.send.as_mut() { @@ -575,13 +592,17 @@ impl Sched { self.settle(key.sub, 1); self.remove_lane(key); } - Op::Datagram { datagram } => { + Op::Datagram { datagram, alias } => { self.settle(key.sub, 1); if expired(deadline) { if let Some(m) = &meta { self.count_dropped(m); } - } else if let Ok(b) = datagram.encode() { + } else if let Ok(b) = { + let mut out = (*datagram).clone(); + out.track_alias = alias; + out.encode() + } { let n = b.len() as u64; if self.conn.send_datagram(b.into()).is_ok() && let Some(m) = &meta @@ -607,7 +628,7 @@ impl Sched { } lane.busy = true; let send = lane.send.take(); - self.spawn_job(Job { lane: key, send, bytes, then: Then::Nothing, deadline: None, priority, meta, track: None }); + self.spawn_job(Job { lane: key, send, parts: vec![bytes.into()], then: Then::Nothing, deadline: None, priority, meta, track: None }); } Op::FetchFin { request_id, order } | Op::FetchReset { request_id, order, .. } => { let then = match op { @@ -626,7 +647,7 @@ impl Sched { let lane = self.lanes.get_mut(&key).expect("the lane"); lane.busy = true; let send = lane.send.take(); - self.spawn_job(Job { lane: key, send, bytes, then, deadline: None, priority, meta: None, track: None }); + self.spawn_job(Job { lane: key, send, parts: if bytes.is_empty() { vec![] } else { vec![bytes.into()] }, then, deadline: None, priority, meta: None, track: None }); } } } @@ -796,6 +817,17 @@ impl Sched { mod tests { use super::*; + #[test] + fn a_shared_payload_is_the_objects_own_bytes() { + let object = Arc::new(SubgroupObject::normal(3, vec![7u8; 4096])); + let (a, b) = (shared_payload(&object), shared_payload(&object)); + assert_eq!(a.as_ptr(), object.payload.as_ptr(), "no copy of the payload"); + assert_eq!(b.as_ptr(), object.payload.as_ptr()); + assert_eq!(&a[..], &object.payload[..]); + drop(object); + assert_eq!(a[4095], 7, "the buffer keeps the object alive"); + } + #[test] fn the_order_is_section_5_1_2s() { let p = |s, pp, g| Prio { subscriber: s, publisher: pp, group_rank: g, fill_rank: 1, mode_rank: 1, within: 0 }; diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/server.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/server.rs index 6d4a8974c..981db7805 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/server.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/server.rs @@ -63,7 +63,9 @@ impl RelayServer { let crypto = quinn::crypto::rustls::QuicServerConfig::try_from(tls).map_err(|e| e.to_string())?; let mut sc = quinn::ServerConfig::with_crypto(Arc::new(crypto)); sc.transport_config(Arc::new(transport_config(&cfg))); - let ep = quinn::Endpoint::server(sc, addr).map_err(|e| format!("binding {addr}: {e}"))?; + let sock = crate::transport::quic::bind_udp(addr, cfg.limits.quic_socket_buffer_bytes).map_err(|e| format!("binding {addr}: {e}"))?; + let runtime = quinn::default_runtime().ok_or("no async runtime")?; + let ep = quinn::Endpoint::new(quinn::EndpointConfig::default(), Some(sc), sock, runtime).map_err(|e| format!("binding {addr}: {e}"))?; quic_addr = Some(ep.local_addr().map_err(|e| e.to_string())?); quic_task = Some(tokio::spawn(quic_accept(relay.clone(), ep.clone(), cfg.path.clone()))); endpoint = Some(ep); diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs index 35f72a764..ddcf61b68 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs @@ -881,9 +881,10 @@ async fn upstream_subgroup(relay: Relay, ctx: Arc, h: joule_moq::dra } } let is_first = first == Some(o.object_id); + let o = Arc::new(o); let taken = relay.fan_out_object( &key, - ObjectIn { upstream: ctx.id, header: &h, subgroup_id: sid, first_in_stream: first, object: &o, arrived, object_timeout, signed }, + ObjectIn { upstream: ctx.id, header: &h, subgroup_id: sid, first_in_stream: first, object: o.clone(), arrived, object_timeout, signed }, &mut started, ); if is_first { diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/mod.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/mod.rs index ede42aaf2..ce18ff8af 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/mod.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/mod.rs @@ -84,6 +84,15 @@ pub type TResult = Result; pub trait SendStream: Send + Sync { /// Write all of `buf`, waiting for flow control as the transport requires. async fn write_all(&mut self, buf: &[u8]) -> TResult<()>; + /// Write `parts`, in order, as one write. A transport that keeps what it is given takes the + /// buffers themselves, so one object's payload is held once however many sessions are + /// sending it; any other copies them, here. + async fn write_parts(&mut self, parts: Vec) -> TResult<()> { + match parts.as_slice() { + [one] => self.write_all(one).await, + many => self.write_all(&many.concat()).await, + } + } /// End the stream gracefully (a FIN). fn finish(&mut self) -> TResult<()>; /// End the stream abruptly with an application code (RESET_STREAM). @@ -148,3 +157,40 @@ pub trait Connection: Send + Sync { } pub type SharedConnection = Arc; + +#[cfg(test)] +mod tests { + use super::*; + + /// A transport that keeps no buffer it is given: what it writes is what a default write gives it. + #[derive(Default)] + struct Copying(Vec>); + + #[async_trait] + impl SendStream for Copying { + async fn write_all(&mut self, buf: &[u8]) -> TResult<()> { + self.0.push(buf.to_vec()); + Ok(()) + } + fn finish(&mut self) -> TResult<()> { + Ok(()) + } + fn reset(&mut self, _code: u64) {} + fn set_priority(&mut self, _priority: i32) {} + async fn delivered(&mut self) -> TResult> { + Ok(None) + } + } + + /// A bridge to an earlier draft reads what it is given as one write of the wire, so an object's + /// head and its payload reach it together. + #[tokio::test] + async fn parts_reach_a_copying_transport_as_one_write() { + let mut s = Copying::default(); + s.write_parts(vec![Bytes::from_static(b"head"), Bytes::from_static(b"payload")]).await.unwrap(); + assert_eq!(s.0, vec![b"headpayload".to_vec()]); + let mut s = Copying::default(); + s.write_parts(vec![Bytes::from_static(b"one")]).await.unwrap(); + assert_eq!(s.0, vec![b"one".to_vec()]); + } +} diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/quic.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/quic.rs index 231afa4fc..bfe7c0a05 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/quic.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/quic.rs @@ -65,6 +65,9 @@ impl SendStream for QuicSend { async fn write_all(&mut self, buf: &[u8]) -> TResult<()> { self.0.write_all(buf).await.map_err(write_err) } + async fn write_parts(&mut self, mut parts: Vec) -> TResult<()> { + self.0.write_all_chunks(&mut parts).await.map_err(write_err) + } fn finish(&mut self) -> TResult<()> { self.0.finish().map_err(TransportError::other) } @@ -163,6 +166,26 @@ pub fn quic_diagnostics(c: &quinn::Connection) -> String { ) } +/// A UDP socket for an endpoint, its receive and send buffers asked to be `buffer` bytes each, or +/// left at the system's default when it is 0. One socket takes every peer's packets, and they +/// arrive in bursts, each fan-out's acknowledgements together: past its buffer the kernel drops +/// them, and a dropped acknowledgement is a retransmission and a smaller congestion window. The +/// system caps the size (`net.core.rmem_max` on Linux), and what it gave is logged when it is less. +pub fn bind_udp(addr: std::net::SocketAddr, buffer: usize) -> std::io::Result { + let sock = std::net::UdpSocket::bind(addr)?; + if buffer > 0 { + // The kernel's option is an int: a larger size would wrap to something small. + let buffer = buffer.min(i32::MAX as usize); + let state = quinn::udp::UdpSocketState::new((&sock).into())?; + let (r, s) = (state.set_recv_buffer_size((&sock).into(), buffer), state.set_send_buffer_size((&sock).into(), buffer)); + let got = state.recv_buffer_size((&sock).into()).unwrap_or(0); + if r.is_err() || s.is_err() || got < buffer { + tracing::warn!(asked = buffer, got, "the UDP receive buffer is smaller than asked: raise the system's limit (net.core.rmem_max on Linux), or a burst of packets is dropped"); + } + } + Ok(sock) +} + /// A client endpoint that offers `alpns` and trusts `roots`, with an optional /// transport configuration (a subscriber that limits its receive windows, say). pub fn client_endpoint( @@ -196,7 +219,7 @@ pub fn client_endpoint( /// A client endpoint with the caller's TLS configuration (a relay dialing its upstream, /// [`crate::cascade`], trusts a CA or a pinned certificate): TLS 1.3, offering `moqt-22` /// alone, so the session speaks -22 from its first byte. -pub fn client_endpoint_tls(bind: std::net::SocketAddr, mut tls: rustls::ClientConfig, transport: Option) -> TResult { +pub fn client_endpoint_tls(bind: std::net::SocketAddr, mut tls: rustls::ClientConfig, transport: Option, socket_buffer: usize) -> TResult { crate::cert::install_provider(); tls.alpn_protocols = vec![joule_moq::draft22::ALPN.to_vec()]; let quic = quinn::crypto::rustls::QuicClientConfig::try_from(tls).map_err(TransportError::other)?; @@ -204,7 +227,9 @@ pub fn client_endpoint_tls(bind: std::net::SocketAddr, mut tls: rustls::ClientCo if let Some(t) = transport { cc.transport_config(std::sync::Arc::new(t)); } - let mut ep = quinn::Endpoint::client(bind).map_err(TransportError::other)?; + let sock = bind_udp(bind, socket_buffer).map_err(TransportError::other)?; + let runtime = quinn::default_runtime().ok_or_else(|| TransportError::other("no async runtime"))?; + let mut ep = quinn::Endpoint::new(quinn::EndpointConfig::default(), None, sock, runtime).map_err(TransportError::other)?; ep.set_default_client_config(cc); Ok(ep) } @@ -237,3 +262,21 @@ pub async fn connect_any(ep: &quinn::Endpoint, addr: std::net::SocketAddr, serve } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn a_socket_is_given_the_buffer_asked_for_within_the_systems_limit() { + let any: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); + let of = |s: &std::net::UdpSocket| quinn::udp::UdpSocketState::new(s.into()).unwrap().recv_buffer_size(s.into()).unwrap(); + let default = of(&bind_udp(any, 0).unwrap()); + // Well under any system's limit, and Linux reports twice what it is asked. + let small = 32 * 1024; + let asked = of(&bind_udp(any, small).unwrap()); + assert!(asked >= small, "{asked} of {small}"); + // A size past the limit is capped with a warning, never a refusal, and never wraps. + assert!(of(&bind_udp(any, 1 << 40).unwrap()) >= default); + } +} diff --git a/joule-code/joule-code-rs/crates/joule-moq/src/draft22/data.rs b/joule-code/joule-code-rs/crates/joule-moq/src/draft22/data.rs index b8405fe3d..0e8a44121 100644 --- a/joule-code/joule-code-rs/crates/joule-moq/src/draft22/data.rs +++ b/joule-code/joule-code-rs/crates/joule-moq/src/draft22/data.rs @@ -213,6 +213,14 @@ impl SubgroupCodec { } pub fn encode_object(&mut self, o: &SubgroupObject, out: &mut Vec) -> WireResult<()> { + self.encode_object_head(o, out)?; + out.extend_from_slice(&o.payload); + Ok(()) + } + + /// [`Self::encode_object`], but for the payload's own bytes, which follow on the wire: a + /// sender holding the payload in a buffer it shares sends that buffer rather than a copy. + pub fn encode_object_head(&mut self, o: &SubgroupObject, out: &mut Vec) -> WireResult<()> { let unenc = |reason| Err(WireError::Unencodable { what: "subgroup object", reason }); let delta = match self.prev { None => o.object_id, @@ -235,7 +243,6 @@ impl SubgroupCodec { if o.payload.is_empty() { put_varint(out, o.status.code()); } - out.extend_from_slice(&o.payload); self.prev = Some(o.object_id); Ok(()) } @@ -1045,6 +1052,20 @@ pub fn encode_padding_stream(n: usize) -> Vec { #[cfg(test)] mod tests { use super::*; + + #[test] + fn an_objects_head_and_payload_are_its_whole_encoding() { + let header = SubgroupHeader { track_alias: 1, group_id: 2, subgroup_id: SubgroupId::Zero, publisher_priority: None, properties: false, end_of_group: false, first_object: true }; + let (mut whole, mut parts) = (SubgroupCodec::new(header), SubgroupCodec::new(header)); + for o in [SubgroupObject::normal(0, vec![9u8; 300]), SubgroupObject::normal(4, vec![1u8; 2])] { + let (mut a, mut b) = (Vec::new(), Vec::new()); + whole.encode_object(&o, &mut a).unwrap(); + parts.encode_object_head(&o, &mut b).unwrap(); + assert!(b.len() < a.len(), "the head holds no payload"); + b.extend_from_slice(&o.payload); + assert_eq!(a, b); + } + } use crate::draft22::kvp::KeyValuePair; fn hdr(props: bool) -> SubgroupHeader { -- 2.54.0 (Apple Git-157)