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",