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.

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

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

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

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<T: std::str::FromStr>(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<u8>) {
     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::<usize>(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<quinn::Endpoint> = (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<quinn::Endpoint> = (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);

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

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(),

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<Location>,
     pub upstreams: Vec<Upstream>,
-    pub downs: Vec<Arc<DownSub>>,
+    /// 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<Vec<Arc<DownSub>>>,
     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<u64> = 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<u64> = 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<u64>, reset: Option<u64>) {
-        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<u64> = 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<u64> = 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");
+    }
+}

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<SocketAddr>,
     pub websocket_addr: Option<SocketAddr>,
-    endpoint: Option<quinn::Endpoint>,
-    quic_task: Option<JoinHandle<()>>,
+    /// One to each UDP socket, each with its own driver and accept loop.
+    endpoints: Vec<quinn::Endpoint>,
+    quic_tasks: Vec<JoinHandle<()>>,
     websocket_task: Option<JoinHandle<()>>,
 }
 
@@ -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<usize> {
+        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

joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs

@@ -1329,6 +1329,7 @@ async fn subscribe(relay: Relay, ctx: Arc<SessionCtx>, 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<SessionCtx>, 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<SessionCtx>, 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<u64> = st.namespaces.iter().filter(|n| n.session == ctx.id).map(|n| n.id).collect();
         st.ns_subs.retain(|s| s.session != ctx.id);

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<std::net::UdpSocket> {
     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<st
             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)
+    Ok(())
+}
+
+/// How many UDP sockets a listener binds when it is asked for `configured`, 0 meaning one per core
+/// up to 16. More than one only on Linux, which can steer a packet by its connection ID.
+pub fn socket_count(configured: usize) -> 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<Vec<std::net::UdpSocket>> {
+    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::<libc::sock_fprog>() 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<std::time::Duration> {
+        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();

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<u8> {
+    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::<usize>(), 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::<usize>(), VIEWERS + 1, "no session ended");
+}

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

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

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