From dcfb9068a117dbd6b41676742aaf3cf089384c22 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 2 Nov 2026 13:37:18 -0500 Subject: [PATCH] joule-moq-relay: a fan-out holds up no other track, and each core has its own socket A big room's fan-out held the relay's one lock for as long as it took to reach every subscriber, so every other room waited. A track's subscribers are now copy on write: a fan-out takes a snapshot under the lock and delivers outside it, so the time under the lock no longer grows with the audience. With a snapshot, a fan-out could still push to a subscription that has just left, which would put a stream after its cancel or past its PUBLISH_DONE count. So a subscription is marked gone under its own lock, and a fan-out holds that lock across each push: a push comes before the mark, or sees it and isn't made. One UDP socket and one endpoint driver carried every peer's packets. The listener now binds one socket per core on its one port, each its own endpoint. Steering by address, which SO_REUSEPORT does by default, would end a session whenever a viewer's address changed: its packets would reach a socket holding no such connection. So a small kernel filter steers by the first byte of the connection ID, and each socket's endpoint issues IDs that come back to it. Co-Authored-By: Claude Opus 5.5 --- joule-code/CHANGELOG.md | 6 + joule-code/joule-code-rs/Cargo.lock | 2 + .../crates/joule-moq-relay/Cargo.toml | 4 + .../crates/joule-moq-relay/README.md | 6 +- .../joule-moq-relay/examples/fanout_load.rs | 18 +- .../crates/joule-moq-relay/relay.example.toml | 3 +- .../crates/joule-moq-relay/src/config.rs | 6 + .../crates/joule-moq-relay/src/relay.rs | 109 ++++++++---- .../crates/joule-moq-relay/src/server.rs | 41 +++-- .../crates/joule-moq-relay/src/session.rs | 22 ++- .../joule-moq-relay/src/transport/quic.rs | 157 +++++++++++++++++- .../joule-moq-relay/tests/quic_sockets.rs | 76 +++++++++ .../crates/joule-moq-remote/Cargo.lock | 2 + .../crates/joule-moq-room/Cargo.lock | 2 + wai/wai-fleet/Cargo.lock | 2 + 15 files changed, 397 insertions(+), 59 deletions(-) create mode 100644 joule-code/joule-code-rs/crates/joule-moq-relay/tests/quic_sockets.rs diff --git a/joule-code/CHANGELOG.md b/joule-code/CHANGELOG.md index ac3081698..62779d6e2 100644 --- a/joule-code/CHANGELOG.md +++ b/joule-code/CHANGELOG.md @@ -15,6 +15,12 @@ All notable changes to the Joule Code standard are recorded here. The format fol (`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. +- **A fan-out delivers outside the relay's lock**, from a copy-on-write snapshot of its track's + subscribers, so a large audience holds up no other track. +- **`limits.quic_sockets`** (0, one per core up to 16; Linux, elsewhere one): UDP sockets on the + listener's one port, each its own endpoint. The kernel steers each packet by the first byte of its + connection ID, and each socket's endpoint issues IDs that come back to it, so a viewer whose + address changes keeps its connection, which steering by address would lose. - **`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. diff --git a/joule-code/joule-code-rs/Cargo.lock b/joule-code/joule-code-rs/Cargo.lock index 209baf56a..6edd99600 100644 --- a/joule-code/joule-code-rs/Cargo.lock +++ b/joule-code/joule-code-rs/Cargo.lock @@ -2109,6 +2109,7 @@ dependencies = [ "futures", "hex", "joule-moq", + "libc", "quinn", "rand_core 0.6.4", "ring", @@ -2116,6 +2117,7 @@ dependencies = [ "rustls-native-certs", "serde", "serde_json", + "socket2", "thiserror", "tokio", "tokio-rustls", 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 adb8e5c22..9e279bc3b 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 @@ -20,6 +20,10 @@ path = "src/main.rs" # objects are sealed (dvr.rs), and the relay never opens one. joule-moq = { path = "../joule-moq", features = ["secure-objects"] } quinn = "0.11" +# The per-core UDP sockets (transport::quic::bind_udp_group): SO_REUSEPORT before the bind, and the +# filter that steers a packet to the socket holding its connection. +socket2 = { version = "0.6", features = ["all"] } +libc = "0.2" rustls = { version = "0.23", default-features = false, features = ["ring", "std"] } ring = "0.17" # WebTransport over HTTP/3 (MIT OR Apache-2.0). Its `quinn` feature lets one quinn 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 7c3a26b82..72b0b2460 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 @@ -331,7 +331,11 @@ not verify. `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 +arrived, how late, and the packets the kernel dropped at each end. A fan-out delivers from a +snapshot of its track's subscribers, outside the relay's lock, so a large audience holds up no +other track. On Linux the QUIC listener binds `limits.quic_sockets` UDP sockets on its one port +(one per core up to 16 by default), each its own endpoint, and the kernel steers each packet by its +connection ID, so a viewer whose address changes keeps its connection. 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 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 20f3a6077..cb96a6a23 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] [--per-endpoint E] +//! [--keyframe-every K --keyframe-bytes B] [--per-endpoint E] [--relay-sockets N] //! ``` //! //! It prints one JSON object: the relay's resident memory before the subscribers, with them idle @@ -30,6 +30,8 @@ use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::time::timeout; +/// Bytes each client socket asks of the kernel. +const CLIENT_SOCKET_BUFFER: usize = 4 << 20; /// Latency buckets of 50 µs, up to 10 s; the last holds everything later. const BUCKET_US: u64 = 50; const BUCKETS: usize = 200_000; @@ -41,11 +43,13 @@ fn arg(args: &[String], flag: &str, default: T) -> T { fn relay_main(args: &[String]) { let threads: usize = arg(args, "--relay-threads", 8); + let sockets: usize = arg(args, "--relay-sockets", 0); 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; + c.limits.quic_sockets = sockets; 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")); @@ -105,7 +109,7 @@ fn spawn_relay(args: &[String]) -> (Child, std::net::SocketAddr, Vec) { let mut child = Command::new("taskset") .args(["-c", &cpus]) .arg(&exe) - .args(["relay", "--relay-threads", &threads.to_string()]) + .args(["relay", "--relay-threads", &threads.to_string(), "--relay-sockets", &arg::(args, "--relay-sockets", 0).to_string()]) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::null()) @@ -175,7 +179,15 @@ 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(|_| { + let ep = quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint"); + // A viewer has a socket of its own, and these share one: a small buffer here would + // drop what no real viewer's would, and measure this program, not the relay. + ep.rebind(quic::bind_udp(bind, CLIENT_SOCKET_BUFFER).expect("a socket")).expect("rebinding"); + ep + }) + .collect(); let connecting = Instant::now(); let gate = Arc::new(tokio::sync::Semaphore::new(128)); let mut joins = Vec::with_capacity(subscribers); 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 5d6ff354e..a6034ab4b 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,7 +47,8 @@ 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_socket_buffer_bytes = 4194304 # each UDP socket's buffers; the system caps them (net.core.rmem_max on Linux) +quic_sockets = 0 # UDP sockets on the one port: 0 is one per core up to 16 (Linux; elsewhere one) 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/config.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs index c2070e86c..60b5f7642 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 @@ -252,6 +252,11 @@ pub struct LimitsConfig { /// packets in 20 s, and 4 MiB dropped none at that steady load. #[serde(default = "d_socket_buffer")] pub quic_socket_buffer_bytes: usize, + /// UDP sockets the QUIC listener binds on its one port, each its own endpoint: 0 is one per core + /// up to 16. The kernel steers each packet to its socket by its connection ID, so a viewer whose + /// address changes keeps its connection. Linux only; elsewhere, one. + #[serde(default)] + pub quic_sockets: 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 @@ -391,6 +396,7 @@ impl Default for LimitsConfig { websocket_buffer_bytes: d_ws_buffer(), idle_timeout_ms: d_idle(), quic_socket_buffer_bytes: d_socket_buffer(), + quic_sockets: 0, 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 9d29c7cdb..0ea367c6b 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 @@ -152,14 +152,24 @@ pub struct SubParams { /// OBJECT_DELIVERY_TIMEOUT and SUBGROUP_DELIVERY_TIMEOUT the subscriber set; 0 none. pub object_timeout: u64, pub subgroup_timeout: u64, + /// The subscription left its track. A fan-out works from a snapshot of the track's subscribers, + /// outside the relay's lock, so one may still be delivering to it: it holds this lock across + /// each push, so a push either comes before this is set, and so before the session's cancel or + /// its PUBLISH_DONE count, or sees it and is not made. + pub gone: bool, } impl SubParams { pub fn passes(&self, loc: Location) -> bool { - self.forward && loc >= self.start && self.end.is_none_or(|e| loc <= e) + !self.gone && self.forward && loc >= self.start && self.end.is_none_or(|e| loc <= e) } } +/// A subscription leaves its track: nothing is pushed to it after this returns. +pub fn mark_gone(sub: &DownSub) { + sub.params.lock().unwrap().gone = true; +} + /// A downstream subscription. pub struct DownSub { pub id: u64, @@ -208,7 +218,9 @@ pub struct Track { pub props: Properties, pub largest: Option, pub upstreams: Vec, - pub downs: Vec>, + /// Its downstream subscriptions, written copy on write: a fan-out takes a snapshot under the + /// relay's lock and delivers outside it, so a large audience holds up no other track. + pub downs: Arc>>, pub cache: TrackCache, pub pub_object_timeout: u64, pub pub_subgroup_timeout: u64, @@ -230,7 +242,7 @@ impl Track { props: Properties::new(), largest: None, upstreams: Vec::new(), - downs: Vec::new(), + downs: Arc::default(), cache: TrackCache::default(), pub_object_timeout: 0, pub_subgroup_timeout: 0, @@ -880,11 +892,7 @@ impl Relay { let mut st = self.inner.state.lock().unwrap(); let receipts_key = self.receipts_of(key); let is_receipts = self.is_receipts_track(&key.1); - let rsessions: HashSet = if is_receipts { - HashSet::new() - } else { - st.tracks.get(&receipts_key).map(|t| t.downs.iter().map(|d| d.session).collect()).unwrap_or_default() - }; + let receipt_downs = if is_receipts { None } else { st.tracks.get(&receipts_key).map(|t| t.downs.clone()) }; let session_tally = st.sessions.get(&o.upstream).map(|c| c.tally.clone()); let group = o.header.group_id; // The group's age (age-assurance §5.5): a subscriber below it is forwarded nothing of it. @@ -926,8 +934,19 @@ impl Relay { return true; } let track_id = track.id; - for sub in &track.downs { - let p = sub.params.lock().unwrap().clone(); + let downs = track.downs.clone(); + if matches!(o.object.status, joule_moq::draft22::data::ObjectStatus::EndOfGroup | joule_moq::draft22::data::ObjectStatus::EndOfTrack) { + self.dvr_seal(track, group); + } + if ins == Insert::New { + self.charge_cache(&mut st, track_id, group, size); + } + // What every subscriber costs, spent outside the lock: a large audience's fan-out holds up + // no other track. + drop(st); + let rsessions: HashSet = receipt_downs.iter().flat_map(|d| d.iter()).map(|d| d.session).collect(); + for sub in downs.iter() { + let p = sub.params.lock().unwrap(); if !p.passes(loc) { continue; } @@ -963,12 +982,6 @@ impl Relay { 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) { - self.dvr_seal(track, group); - } - if ins == Insert::New { - self.charge_cache(&mut st, track_id, group, size); - } true } @@ -985,11 +998,17 @@ impl Relay { /// An upstream subgroup stream ended: with a FIN, or reset with `reset`. pub fn end_subgroup(&self, key: &TrackKey, group: u64, subgroup: u64, subgroup_timeout_pub: u64, started: &HashSet, reset: Option) { - let st = self.inner.state.lock().unwrap(); - let Some(track) = st.tracks.get(key) else { return }; - for sub in track.downs.iter().filter(|d| started.contains(&d.id)) { + let (downs, default_priority) = { + let st = self.inner.state.lock().unwrap(); + let Some(track) = st.tracks.get(key) else { return }; + (track.downs.clone(), track.default_priority) + }; + for sub in downs.iter().filter(|d| started.contains(&d.id)) { let lane = LaneKey { sub: sub.id, kind: LaneKind::Subgroup { group, subgroup } }; - let p = sub.params.lock().unwrap().clone(); + let p = sub.params.lock().unwrap(); + if p.gone { + continue; + } let op = match reset { Some(code) => Op::Reset { code }, None => { @@ -997,7 +1016,7 @@ impl Relay { Op::Fin { subgroup_timeout: (t != 0).then(|| Duration::from_millis(t)) } } }; - let prio = Prio { subscriber: p.subscriber_priority, publisher: track.default_priority, group_rank: Prio::group_rank(p.order, group), fill_rank: 1, mode_rank: 1, within: subgroup }; + let prio = Prio { subscriber: p.subscriber_priority, publisher: default_priority, group_rank: Prio::group_rank(p.order, group), fill_rank: 1, mode_rank: 1, within: subgroup }; sub.sched.push(lane, Item { op, prio, deadline: None, meta: None }); } } @@ -1008,7 +1027,7 @@ impl Relay { 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(); + let receipt_downs = st.tracks.get(&receipts_key).map(|t| t.downs.clone()); let age = st.tracks.get(key).map_or(0, |t| self.group_age(key, &t.digest, d.group_id)); let Some(track) = st.tracks.get_mut(key) else { return }; let loc = Location::new(d.group_id, d.object_id); @@ -1032,8 +1051,12 @@ impl Relay { } let track_id = track.id; let object_timeout = effective_timeout(track.pub_object_timeout, track.pub_subgroup_timeout); - for sub in &track.downs { - let p = sub.params.lock().unwrap().clone(); + let downs = track.downs.clone(); + self.charge_cache(&mut st, track_id, d.group_id, size); + drop(st); + let rsessions: HashSet = receipt_downs.iter().flat_map(|d| d.iter()).map(|d| d.session).collect(); + for sub in downs.iter() { + let p = sub.params.lock().unwrap(); if !p.passes(loc) { continue; } @@ -1048,7 +1071,6 @@ impl Relay { 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: shared.clone(), alias: sub.alias }, prio, deadline, meta: Some(meta) }); } - self.charge_cache(&mut st, track_id, d.group_id, size); } /// An upstream of `key` is gone. When it was the last, every downstream @@ -1095,7 +1117,7 @@ impl Relay { let mut st = self.inner.state.lock().unwrap(); let mut gone_aliases = Vec::new(); let ends = match st.tracks.get_mut(key) { - None => Vec::new(), + None => Arc::default(), Some(track) => { track.upstreams.retain(|u| { if u.session == session { @@ -1113,9 +1135,11 @@ impl Relay { // PUBLISH_DONE's wait, and then its session's end) finds nothing. if !gone_aliases.is_empty() && track.upstreams.is_empty() { track.ended = Some(status); - std::mem::take(&mut track.downs) + let ends = std::mem::take(&mut track.downs); + ends.iter().for_each(|d| mark_gone(d)); + ends } else { - Vec::new() + Arc::default() } } }; @@ -1125,8 +1149,8 @@ impl Relay { } ends }; - for sub in ends { - end_down(sub, status); + for sub in ends.iter() { + end_down(sub.clone(), status); } self.inner.change.notify_waiters(); self.inner.ns_changed.notify_waiters(); @@ -1141,7 +1165,8 @@ impl Relay { None => Vec::new(), Some(track) => { let before = track.downs.len(); - track.downs.retain(|d| d.id != sub.id); + Arc::make_mut(&mut track.downs).retain(|d| d.id != sub.id); + mark_gone(sub); if track.downs.len() != before { sub.tally.lock().unwrap().unsubscribed(&sub.render, now_unix_us()); } @@ -1296,3 +1321,25 @@ pub fn signed_of(v: &crate::door::Verdict) -> Option<(u64, &'static str)> { _ => None, } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_subscription_that_left_passes_nothing() { + let mut p = SubParams { + subscriber_priority: 128, + order: GroupOrder::Ascending, + forward: true, + start: Location::new(0, 0), + end: None, + object_timeout: 0, + subgroup_timeout: 0, + gone: false, + }; + assert!(p.passes(Location::new(1, 0))); + p.gone = true; + assert!(!p.passes(Location::new(1, 0)), "a fan-out's snapshot may still hold it"); + } +} 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 981db7805..30b66da27 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 @@ -22,8 +22,9 @@ pub struct RelayServer { pub identity: RelayIdentity, pub quic_addr: Option, pub websocket_addr: Option, - endpoint: Option, - quic_task: Option>, + /// One to each UDP socket, each with its own driver and accept loop. + endpoints: Vec, + quic_tasks: Vec>, websocket_task: Option>, } @@ -54,8 +55,7 @@ impl RelayServer { _ => return Err("tls: cert_path and key_path go together".into()), }; let relay = Relay::new(cfg.clone())?; - let mut endpoint = None; - let mut quic_task = None; + let (mut endpoints, mut quic_tasks) = (Vec::new(), Vec::new()); let mut websocket_task = None; let mut quic_addr = None; if let Some(addr) = cfg.quic_listen { @@ -63,12 +63,20 @@ 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 sock = crate::transport::quic::bind_udp(addr, cfg.limits.quic_socket_buffer_bytes).map_err(|e| format!("binding {addr}: {e}"))?; + use crate::transport::quic::{bind_udp_group, socket_count, ShardedCid}; + let count = socket_count(cfg.limits.quic_sockets); + let socks = bind_udp_group(addr, count, 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); + for (i, sock) in socks.into_iter().enumerate() { + let mut ec = quinn::EndpointConfig::default(); + if count > 1 { + ec.cid_generator(move || Box::new(ShardedCid::new(i, count))); + } + let ep = quinn::Endpoint::new(ec, Some(sc.clone()), sock, runtime.clone()).map_err(|e| format!("binding {addr}: {e}"))?; + quic_addr = Some(ep.local_addr().map_err(|e| e.to_string())?); + quic_tasks.push(tokio::spawn(quic_accept(relay.clone(), ep.clone(), cfg.path.clone()))); + endpoints.push(ep); + } } let mut websocket_addr = None; if let Some(addr) = cfg.websocket_listen { @@ -81,17 +89,22 @@ impl RelayServer { for l in relay.links() { tokio::spawn(crate::cascade::run_link(relay.clone(), l.clone())); } - Ok(RelayServer { relay, identity, quic_addr, websocket_addr, endpoint, quic_task, websocket_task }) + Ok(RelayServer { relay, identity, quic_addr, websocket_addr, endpoints, quic_tasks, websocket_task }) } pub fn relay(&self) -> &Relay { &self.relay } + /// The QUIC connections each of the listener's UDP sockets holds. + pub fn quic_connections_per_socket(&self) -> Vec { + self.endpoints.iter().map(|e| e.open_connections()).collect() + } + /// Stop accepting, GOAWAY every session, wait up to the configured GOAWAY timeout, /// close what is left with GOAWAY_TIMEOUT. Returns how many sessions that closed. pub async fn shutdown(self) -> usize { - if let Some(ep) = &self.endpoint { + for ep in &self.endpoints { ep.set_server_config(None); } // The WebSocket listener stops taking upgrades; the sessions it made go on. @@ -100,11 +113,13 @@ impl RelayServer { } let t = Duration::from_millis(self.relay.config().limits.goaway_timeout_ms); let closed = self.relay.shutdown(t).await; - if let Some(ep) = &self.endpoint { + for ep in &self.endpoints { ep.close(quinn::VarInt::from_u32(0), b"the relay stopped"); + } + for ep in &self.endpoints { let _ = timeout(Duration::from_secs(5), ep.wait_idle()).await; } - if let Some(t) = &self.quic_task { + for t in &self.quic_tasks { t.abort(); } closed 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 ddcf61b68..1d54b4dc6 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 @@ -1329,6 +1329,7 @@ async fn subscribe(relay: Relay, ctx: Arc, r: TrackRequest, tx: ReqT end, object_timeout: r.params.object_delivery_timeout().unwrap_or(0), subgroup_timeout: varint_param(&r.params, param::SUBGROUP_DELIVERY_TIMEOUT).unwrap_or(0), + gone: false, }; let id = relay.next_id(); let sub = Arc::new(DownSub { @@ -1346,7 +1347,7 @@ async fn subscribe(relay: Relay, ctx: Arc, r: TrackRequest, tx: ReqT tx: tx.clone(), ended: watch::channel(false).0, }); - track.downs.push(sub.clone()); + Arc::make_mut(&mut track.downs).push(sub.clone()); let mut params = Parameters::new(); if let Some(l) = track.largest { params.push(Parameter::LargestObject(l)); @@ -2558,14 +2559,17 @@ async fn cleanup(relay: &Relay, ctx: &Arc, why: TransportError) { st.sessions.remove(&ctx.id); let mut downs = Vec::new(); for t in st.tracks.values_mut() { - t.downs.retain(|d| { - if d.session == ctx.id { - downs.push(d.clone()); - false - } else { - true - } - }); + if t.downs.iter().any(|d| d.session == ctx.id) { + Arc::make_mut(&mut t.downs).retain(|d| { + if d.session == ctx.id { + crate::relay::mark_gone(d); + downs.push(d.clone()); + false + } else { + true + } + }); + } } let nss: Vec = st.namespaces.iter().filter(|n| n.session == ctx.id).map(|n| n.id).collect(); st.ns_subs.retain(|s| s.session != ctx.id); 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 bfe7c0a05..9880881a0 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 @@ -173,6 +173,11 @@ pub fn quic_diagnostics(c: &quinn::Connection) -> String { /// 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)?; + size_buffers(&sock, buffer)?; + Ok(sock) +} + +fn size_buffers(sock: &std::net::UdpSocket, buffer: usize) -> std::io::Result<()> { 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); @@ -183,7 +188,122 @@ pub fn bind_udp(addr: std::net::SocketAddr, buffer: usize) -> std::io::Result usize { + if !cfg!(target_os = "linux") { + return 1; + } + match configured { + 0 => std::thread::available_parallelism().map_or(1, |n| n.get()).min(16), + n => n.min(256), + } +} + +/// `count` UDP sockets on one address, each to be its own endpoint with its own driver, so no one +/// socket and task carries every peer's packets. The kernel steers each packet by the first byte of +/// its destination connection ID ([`ShardedCid`]), and not by its address: a viewer whose address +/// changes keeps its connection, which steering by address would lose. One socket when `count` is 1. +pub fn bind_udp_group(addr: std::net::SocketAddr, count: usize, buffer: usize) -> std::io::Result> { + if count <= 1 { + return Ok(vec![bind_udp(addr, buffer)?]); + } + let mut addr = addr; + let mut socks = Vec::with_capacity(count); + for _ in 0..count { + let s = socket2::Socket::new(socket2::Domain::for_address(addr), socket2::Type::DGRAM, Some(socket2::Protocol::UDP))?; + s.set_reuse_port(true)?; + s.bind(&addr.into())?; + let s: std::net::UdpSocket = s.into(); + // The first socket fixes the port of a listener bound to port 0. + addr = s.local_addr()?; + size_buffers(&s, buffer)?; + socks.push(s); + } + steer_by_connection_id(&socks[0], count as u32)?; + Ok(socks) +} + +/// Attach the filter that picks a packet's socket, by the order the group's sockets were bound: the +/// first byte of its destination connection ID, modulo the count. A long header's starts at byte 6 +/// (flags, four of version, its length) and a short header's at byte 1. A client's first Initial +/// names an ID the client chose, which picks the socket; that socket's endpoint then issues only +/// IDs that come back to it ([`ShardedCid`]). +#[cfg(target_os = "linux")] +fn steer_by_connection_id(sock: &std::net::UdpSocket, count: u32) -> std::io::Result<()> { + use std::os::fd::AsRawFd; + const LD_B_ABS: u16 = 0x30; + const JMP_JSET: u16 = 0x45; + const JMP_JA: u16 = 0x05; + const ALU_MOD: u16 = 0x94; + const RET_A: u16 = 0x16; + let op = |code, jt, jf, k| libc::sock_filter { code, jt, jf, k }; + let mut program = [ + op(LD_B_ABS, 0, 0, 0), + op(JMP_JSET, 0, 2, 0x80), + op(LD_B_ABS, 0, 0, 6), + op(JMP_JA, 0, 0, 1), + op(LD_B_ABS, 0, 0, 1), + op(ALU_MOD, 0, 0, count), + op(RET_A, 0, 0, 0), + ]; + let prog = libc::sock_fprog { len: program.len() as u16, filter: program.as_mut_ptr() }; + // SAFETY: `prog` points at `program`, both alive across the call, with its true length, and the + // kernel copies the program before it returns. + let r = unsafe { libc::setsockopt(sock.as_raw_fd(), libc::SOL_SOCKET, libc::SO_ATTACH_REUSEPORT_CBPF, (&raw const prog).cast(), std::mem::size_of::() as libc::socklen_t) }; + if r != 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) +} + +#[cfg(not(target_os = "linux"))] +fn steer_by_connection_id(_sock: &std::net::UdpSocket, _count: u32) -> std::io::Result<()> { + Err(std::io::Error::new(std::io::ErrorKind::Unsupported, "more than one socket needs Linux")) +} + +/// The length of a connection ID this listener issues. +const CID_LEN: usize = 8; + +/// The connection IDs of one socket of a group: random, but for a first byte that the kernel's +/// filter ([`steer_by_connection_id`]) maps back to this socket. +pub struct ShardedCid { + shard: u8, + shards: u16, +} + +impl ShardedCid { + /// The IDs of socket `shard` of `shards`, at most 256. + pub fn new(shard: usize, shards: usize) -> ShardedCid { + assert!(shard < shards && shards <= 256, "socket {shard} of {shards}"); + ShardedCid { shard: shard as u8, shards: shards as u16 } + } + + /// `first` with its value modulo the count made this socket's. The largest it gives is + /// (⌊256 / shards⌋ − 1) × shards + shards − 1, which a byte holds. + fn first_byte(&self, first: u8) -> u8 { + let q = first as u16 % (256 / self.shards); + (q * self.shards + self.shard as u16) as u8 + } +} + +impl quinn::ConnectionIdGenerator for ShardedCid { + fn generate_cid(&mut self) -> quinn::ConnectionId { + use rand_core_06::RngCore; + let mut b = [0u8; CID_LEN]; + rand_core_06::OsRng.fill_bytes(&mut b); + b[0] = self.first_byte(b[0]); + quinn::ConnectionId::new(&b) + } + fn cid_len(&self) -> usize { + CID_LEN + } + fn cid_lifetime(&self) -> Option { + None + } } /// A client endpoint that offers `alpns` and trusts `roots`, with an optional @@ -267,6 +387,41 @@ pub async fn connect_any(ep: &quinn::Endpoint, addr: std::net::SocketAddr, serve mod tests { use super::*; + #[test] + fn a_sockets_connection_ids_come_back_to_it() { + use quinn::ConnectionIdGenerator; + for shards in [2usize, 3, 4, 7, 8, 16, 32, 128, 255, 256] { + for shard in [0, 1, shards / 2, shards - 1] { + let mut g = ShardedCid::new(shard, shards); + assert!((0..=255u8).all(|b| g.first_byte(b) as usize % shards == shard), "socket {shard} of {shards}"); + let cid = g.generate_cid(); + assert_eq!(cid.len(), 8); + assert_eq!(cid[0] as usize % shards, shard); + } + } + } + + #[test] + fn a_socket_count_of_zero_is_one_per_core_within_a_bound() { + let auto = socket_count(0); + assert!((1..=16).contains(&auto)); + if cfg!(target_os = "linux") { + assert_eq!(socket_count(5), 5); + } else { + assert_eq!(socket_count(5), 1, "only Linux can steer by connection ID"); + } + } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn a_group_of_sockets_shares_one_port() { + let any: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); + let socks = bind_udp_group(any, 4, 0).unwrap(); + assert_eq!(socks.len(), 4); + let port = socks[0].local_addr().unwrap().port(); + assert!(port != 0 && socks.iter().all(|s| s.local_addr().unwrap().port() == port)); + } + #[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(); diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/quic_sockets.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/quic_sockets.rs new file mode 100644 index 000000000..9d09e59fd --- /dev/null +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/quic_sockets.rs @@ -0,0 +1,76 @@ +//! The QUIC listener's UDP sockets: the sessions spread across them, and a viewer whose address +//! changes keeps its connection, because the kernel steers a packet by its connection ID and not by +//! its address. On a system with one socket, this holds trivially. + +mod common; + +use common::*; +use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject}; +use joule_moq::draft22::{Parameters, UniEvent}; +use joule_moq_relay::client::ClientOptions; +use std::time::Duration; +use tokio::time::timeout; + +const SOCKETS: usize = 4; +const VIEWERS: usize = 32; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_viewer_whose_address_changes_keeps_its_connection() { + timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time"); +} + +async fn next_payload(c: &joule_moq_relay::client::Client) -> Vec { + let mut r = timeout(STEP, c.next_data(STEP)).await.expect("in time").expect("a stream"); + loop { + match timeout(STEP, r.next()).await.expect("in time").expect("a read") { + Some(UniEvent::SubgroupObject(o)) => return o.payload, + Some(_) => continue, + None => panic!("a stream with no object"), + } + } +} + +async fn run() { + let server = relay(|c| c.limits.quic_sockets = SOCKETS).await; + let (publisher, _pub_ep) = connect(&server, Via::Quic, ClientOptions::default()).await; + let (space, track) = (ns(&["moved"]), name("video")); + let _published = publisher.publish(space.clone(), track.clone(), 1, Parameters::new(), Default::default()).await.expect("PUBLISH"); + + let mut viewers = Vec::with_capacity(VIEWERS); + for _ in 0..VIEWERS { + let (c, ep) = connect(&server, Via::Quic, ClientOptions::default()).await; + let held = c.subscribe(space.clone(), track.clone(), Parameters::new()).await.expect("SUBSCRIBE"); + viewers.push((c, ep.expect("a QUIC endpoint"), held)); + } + + let per_socket = server.quic_connections_per_socket(); + assert_eq!(per_socket.iter().sum::(), VIEWERS + 1, "{per_socket:?}"); + if cfg!(target_os = "linux") { + assert_eq!(per_socket.len(), SOCKETS); + assert!(per_socket.iter().filter(|n| **n > 0).count() > 1, "the sessions spread across the sockets: {per_socket:?}"); + } + + let send = |g: u64| { + let h = SubgroupHeader { track_alias: 1, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true }; + (h, [SubgroupObject::normal(0, g.to_le_bytes().to_vec())]) + }; + let (h, o) = send(0); + timeout(STEP, publisher.send_subgroup(h, &o)).await.expect("in time").expect("a send"); + for (c, _, _) in &viewers { + assert_eq!(next_payload(c).await, 0u64.to_le_bytes()); + } + + // Every viewer moves to a new address. Steered by address, most of their packets would reach + // sockets that hold no such connection, and their sessions would end. + for (_, ep, _) in &viewers { + ep.rebind(std::net::UdpSocket::bind(loopback()).expect("a new socket")).expect("rebinding"); + } + for g in 1..=3 { + let (h, o) = send(g); + timeout(STEP, publisher.send_subgroup(h, &o)).await.expect("in time").expect("a send"); + for (c, _, _) in &viewers { + assert_eq!(next_payload(c).await, g.to_le_bytes(), "group {g}, after the move"); + } + } + assert_eq!(server.quic_connections_per_socket().iter().sum::(), VIEWERS + 1, "no session ended"); +} diff --git a/joule-code/joule-code-rs/crates/joule-moq-remote/Cargo.lock b/joule-code/joule-code-rs/crates/joule-moq-remote/Cargo.lock index 768afa9cf..51322d2c8 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-remote/Cargo.lock +++ b/joule-code/joule-code-rs/crates/joule-moq-remote/Cargo.lock @@ -1036,6 +1036,7 @@ dependencies = [ "futures", "hex", "joule-moq", + "libc", "quinn", "rand_core 0.6.4", "ring", @@ -1043,6 +1044,7 @@ dependencies = [ "rustls-native-certs", "serde", "serde_json", + "socket2", "thiserror", "tokio", "tokio-rustls", diff --git a/joule-code/joule-code-rs/crates/joule-moq-room/Cargo.lock b/joule-code/joule-code-rs/crates/joule-moq-room/Cargo.lock index e6ce383d2..e45f0c830 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-room/Cargo.lock +++ b/joule-code/joule-code-rs/crates/joule-moq-room/Cargo.lock @@ -1515,6 +1515,7 @@ dependencies = [ "futures", "hex", "joule-moq", + "libc", "quinn", "rand_core 0.6.4", "ring", @@ -1522,6 +1523,7 @@ dependencies = [ "rustls-native-certs", "serde", "serde_json", + "socket2", "thiserror", "tokio", "tokio-rustls", diff --git a/wai/wai-fleet/Cargo.lock b/wai/wai-fleet/Cargo.lock index d19a8d305..bb63fc42f 100644 --- a/wai/wai-fleet/Cargo.lock +++ b/wai/wai-fleet/Cargo.lock @@ -1157,6 +1157,7 @@ dependencies = [ "futures", "hex", "joule-moq", + "libc", "quinn", "rand_core 0.6.4", "ring", @@ -1164,6 +1165,7 @@ dependencies = [ "rustls-native-certs", "serde", "serde_json", + "socket2", "thiserror", "tokio", "tokio-rustls", -- 2.54.0 (Apple Git-157)