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 new file mode 100644 index 000000000..20f3a6077 --- /dev/null +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs @@ -0,0 +1,316 @@ +//! 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] [--per-endpoint E] +//! ``` +//! +//! 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; + +/// 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 +} + +/// 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 +} + +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 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"); + 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 drops_start = udp_drops(addr.port()); + 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.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(); + 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.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]); + 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},\"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, + 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], + drops_end.0.saturating_sub(drops_start.0), + drops_end.1.saturating_sub(drops_start.1), + by[0], + by[1], + by[2], + ); + 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); + } +} 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 {