From 869fe81c02a6cb9fd8522ac254b6022667f1cca9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 21 Sep 2026 13:32:33 -0700 Subject: [PATCH 1/4] chore(quest): claim uring-identity From de151749c537ef795683c72ab626e895884c8905 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 21 Sep 2026 14:51:24 -0700 Subject: [PATCH 2/4] refactor(uring)!: derive worker and steering identity from the socket `Endpoint::new` took a Handle next to a socket that already named its worker, and `endpoint::Config::shard` was settable independently of the socket's slot in the reuseport group. Both are now derived: `Handle::udp` adopts a lone socket or a completed `moq_sock::shard::Socket` member, the socket carries its worker (weakly) and shard, and the endpoint, its connections, and the WebTransport handshake run on that worker. An endpoint on a dropped worker is refused up front, and a dial or accept on one that outlived its worker fails instead of parking forever. Co-Authored-By: Claude Opus 5 --- quest/dev/release.md | 1 - quest/main/README.md | 1 - quest/main/uring-identity.md | 40 ------ rs/moq-relay/src/uring.rs | 47 +++---- rs/moq-relay/tests/runtime_uring.rs | 2 +- rs/moq-uring/benches/echo_noq.rs | 6 +- rs/moq-uring/benches/session_lite.rs | 6 +- rs/moq-uring/src/lib.rs | 10 +- rs/moq-uring/src/quic/client.rs | 8 +- rs/moq-uring/src/quic/endpoint.rs | 33 ++--- rs/moq-uring/src/quic/noq/connection.rs | 16 ++- rs/moq-uring/src/quic/noq/endpoint.rs | 52 +++++--- rs/moq-uring/src/quic/server.rs | 10 +- rs/moq-uring/src/quic/web.rs | 26 ++-- rs/moq-uring/src/udp.rs | 74 ++++++++--- rs/moq-uring/src/worker.rs | 67 +++++++++- rs/moq-uring/tests/echo.rs | 4 +- rs/moq-uring/tests/endpoint.rs | 44 +++---- rs/moq-uring/tests/identity.rs | 157 ++++++++++++++++++++++++ rs/moq-uring/tests/qlog.rs | 9 +- rs/moq-uring/tests/session.rs | 28 ++--- rs/moq-uring/tests/teardown.rs | 3 +- rs/moq-uring/tests/web.rs | 12 +- rs/moq-uring/tests/workers.rs | 32 ++--- 24 files changed, 439 insertions(+), 249 deletions(-) delete mode 100644 quest/main/uring-identity.md create mode 100644 rs/moq-uring/tests/identity.rs diff --git a/quest/dev/release.md b/quest/dev/release.md index 890c3f4976..b1ff955bb2 100644 --- a/quest/dev/release.md +++ b/quest/dev/release.md @@ -85,7 +85,6 @@ Public API: none beyond the required quests. Wire: none. ## Required - [E2EE API](/quest/main/e2ee-api.md) - expose epoch-scoped ownership and align the implemented profile -- [uring identity](/quest/main/uring-identity.md) - bind sockets, connections, workers, and steering identity together - [Merge dev](/quest/dev/merge-dev.md) - the tree the release is cut from - [Binding audio tests](/quest/next/binding-audio-tests.md) - every binding proves the audio config it exposes - [Decode format](/quest/next/ffi-decode-format.md) - the C-only decode knob reaches every uniffi binding diff --git a/quest/main/README.md b/quest/main/README.md index 365c0cf841..4f85eb17a4 100644 --- a/quest/main/README.md +++ b/quest/main/README.md @@ -89,7 +89,6 @@ do not add another media abstraction or a renderer crate during stabilization. ## Quests - [E2EE API](/quest/main/e2ee-api.md) - epoch-scoped ownership replaces raw crypto, catalog helpers, and process-global claims -- [uring identity](/quest/main/uring-identity.md) - sockets and connections carry their worker and steering identity - [NVENC registration rollback](/quest/main/nvenc-registration.md) - release resources when mapping fails after registration - [GPU conversion and NVENC](/quest/main/video-gpu-encode.md) - convert, resize and encode imported frames without CPU pixel transfers or fallback diff --git a/quest/main/uring-identity.md b/quest/main/uring-identity.md deleted file mode 100644 index 225aa2db1d..0000000000 --- a/quest/main/uring-identity.md +++ /dev/null @@ -1,40 +0,0 @@ -# [M] uring I/O carries its worker and steering identity - -## Goal - -An endpoint cannot drive a socket through the wrong worker or issue connection -IDs for a different reuseport member. WebTransport setup inherits the worker -that already drives its connection. Keep the worker, UDP, and QUIC layers. - -## Plan - -`Handle::udp` already associates a socket with a worker, but `Endpoint::new` -takes a separate Handle and `endpoint::Config::shard` is independently -settable. A socket from worker A combined with handle B splits receive I/O -and driver progress across unrelated loops. A wrong shard steers replies to -the wrong socket. `web::Request::accept` repeats the independently supplied -handle next to a connection that already has an owner. - -Derive worker identity from the owned socket/connection and carry steering -identity from the completed socket-group member. A plain unsharded socket -remains supported. Remove redundant independently supplied identity rather -than documenting combinations callers must remember. Preserve weak lifetime -links where needed so an I/O handle does not keep a stopped worker alive. - -Update the single-connection helpers and in-tree runtime callers with the -same rule. Keep root Config and the existing TxBuf name; those names do not -cause the mismatch. Document exported metrics fields as part of the API pass. - -Linux CI covers a stopped owner, two workers on the same thread, a two-member -steered listener, and WebTransport setup. Invalid combinations are either -unconstructible through the public API or refused before tasks or I/O start. -Normal single-worker and multi-worker traffic still works. Preserve the -published moq-tokio worker surface when migrating its internal plumbing. - -Public API: breaking construction/configuration changes in moq-uring 0.0.1 -and the shared moq-sock member integration. Wire: no format change. - -## Related - -- [Handshake cancellation](/quest/next/uring-handshake-cancel.md) - cleanup during a suspended handshake without another API -- [PR #3811](https://github.com/moq-dev/moq/pull/3811) - landed the single noq backend before this ownership work diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 16eae650e9..75774df346 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -23,12 +23,6 @@ use anyhow::Context as _; use crate::{auth, cluster, shutdown}; -/// One member's bound socket and its slot in the steered group. -struct Member { - shard: moq_sock::shard::Shard, - socket: std::net::UdpSocket, -} - /// A stop signal a worker parks on, wakeable from the shared runtime. #[derive(Default)] struct Stop { @@ -83,7 +77,7 @@ struct Serve { /// worker that quietly died. Nothing is served until [`serve`](Self::serve) /// spawns the threads. pub struct Workers { - members: Vec, + members: Vec, addr: SocketAddr, server: moq_uring::quic::server::Config, /// What the workers serve, fingerprinted once, for `/certificate.sha256`. @@ -199,10 +193,7 @@ impl Workers { .context("failed to complete the reuseport group")?; let mut members = Vec::with_capacity(count as usize); while let Some(member) = group.member().context("failed to clone a reuseport member")? { - members.push(Member { - shard: member.shard(), - socket: member.into_inner(), - }); + members.push(member); } // Whatever the first member bound, which is the requested address unless // it asked for an ephemeral port. @@ -325,7 +316,7 @@ impl Workers { }; for member in std::mem::take(&mut self.members) { - let index = member.shard.index(); + let index = member.shard().index(); let core = cores.get(index as usize % cores.len().max(1)).copied(); let stop = Arc::new(Stop::default()); let (ready_tx, ready_rx) = std::sync::mpsc::channel(); @@ -478,7 +469,7 @@ fn join(threads: Vec<(u16, std::thread::JoinHandle<()>)>) { /// Everything one worker thread is handed at spawn. struct Spawn { - member: Member, + member: moq_sock::shard::Socket, /// The core to pin to, or `None` when pinning is off or unavailable. core: Option, /// This worker's counter set, shared with whoever scrapes `/metrics`. @@ -503,7 +494,7 @@ struct Spawn { /// alone would not notice one, since the parent holds a sender of its own and /// the channel therefore never closes. fn run_worker(spawn: Spawn) { - let index = spawn.member.shard.index(); + let index = spawn.member.shard().index(); let stop = spawn.stop.clone(); let failures = spawn.failures.clone(); @@ -537,7 +528,7 @@ fn serve_worker(spawn: Spawn) -> bool { ready, failures, } = spawn; - let index = member.shard.index(); + let index = member.shard().index(); if let Some(core) = core { if moq_sock::cpu::pin(core) { tracing::debug!(index, core = core.id(), "pinned io_uring QUIC worker"); @@ -550,18 +541,15 @@ fn serve_worker(spawn: Spawn) -> bool { let mut config = moq_uring::Config::default(); config.metrics = metrics; let worker = moq_uring::Worker::new(config).context("io_uring setup failed")?; - let handle = worker.handle(); - let socket = handle - .udp(member.socket, udp) + // The member carries its slot in the steered group, so adopting it is + // what makes the endpoint issue connection ids that steer back here. + let socket = worker + .handle() + .udp(member, udp) .context("failed to adopt the worker socket")?; - let endpoint = moq_uring::quic::Endpoint::new( - &handle, - socket, - moq_uring::quic::endpoint::Config::default() - .with_server(server) - .with_shard(member.shard), - ) - .context("failed to build the QUIC endpoint")?; + let endpoint = + moq_uring::quic::Endpoint::new(socket, moq_uring::quic::endpoint::Config::default().with_server(server)) + .context("failed to build the QUIC endpoint")?; Ok((worker, endpoint)) })(); @@ -638,7 +626,7 @@ async fn serve_connection( let mut alpn = conn.protocol().map(str::to_owned); let (transport, url) = match conn.protocol() { Some("h3") => { - let request = moq_uring::quic::web::Request::accept(&handle, conn) + let request = moq_uring::quic::web::Request::accept(conn) .await .context("WebTransport handshake failed")?; let url = request.url().clone(); @@ -768,9 +756,8 @@ async fn serve_connection( let (session, driver) = request.ok().await?; let driver_handle = handle.clone(); handle.spawn(async move { - if let Err(err) = driver_handle.run(driver).await { - tracing::debug!(%err, "session driver ended"); - } + let err = driver_handle.run(driver).await; + tracing::debug!(%err, "session driver ended"); }); let node_connection = peer_hop.map(|origin| serve.cluster.nodes.connect_inbound(id, origin)); diff --git a/rs/moq-relay/tests/runtime_uring.rs b/rs/moq-relay/tests/runtime_uring.rs index 8e9f67d27a..8a5dab09a8 100644 --- a/rs/moq-relay/tests/runtime_uring.rs +++ b/rs/moq-relay/tests/runtime_uring.rs @@ -377,7 +377,7 @@ async fn uring_workers_write_qlog_traces() { // rather than a connection that only ever exchanged Initials. let url: url::Url = format!("moql://127.0.0.1:{port}/qlog").parse().expect("parse url"); let origin = moq_tokio::origin::spawn(); - let mut broadcast = origin.create_broadcast("test").expect("create broadcast"); + let broadcast = origin.create_broadcast("test").expect("create broadcast"); broadcast.announce(Default::default()).expect("create broadcast"); let track = broadcast.create_track("video", None).expect("create track"); let mut group = track.append_group().expect("append group"); diff --git a/rs/moq-uring/benches/echo_noq.rs b/rs/moq-uring/benches/echo_noq.rs index 650a71664e..ed124886c9 100644 --- a/rs/moq-uring/benches/echo_noq.rs +++ b/rs/moq-uring/benches/echo_noq.rs @@ -52,8 +52,8 @@ mod linux { let socket = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp_config) .expect("socket"); - let endpoint = quic::Endpoint::new(&handle, socket, quic::endpoint::Config::default().with_server(server)) - .expect("endpoint"); + let endpoint = + quic::Endpoint::new(socket, quic::endpoint::Config::default().with_server(server)).expect("endpoint"); let addr = endpoint.local_addr(); // Queue the first iteration while the worker finishes driving the client // handshake. A zero-capacity channel would block this thread before it can @@ -87,7 +87,7 @@ mod linux { let mut session = worker .block_on(async { let conn = endpoint.accept().await.expect("accept"); - quic::web::Request::accept(&handle, conn) + quic::web::Request::accept(conn) .await .expect("handshake") .ok() diff --git a/rs/moq-uring/benches/session_lite.rs b/rs/moq-uring/benches/session_lite.rs index df792a776f..50efb9797c 100644 --- a/rs/moq-uring/benches/session_lite.rs +++ b/rs/moq-uring/benches/session_lite.rs @@ -125,7 +125,7 @@ mod linux { let server_handle = handle.clone(); handle.spawn(async move { - let conn = quic::server::accept(&server_handle, server_sock, &server_config) + let conn = quic::server::accept(server_sock, &server_config) .await .expect("quic accept"); let (session, driver) = moq_net::Server::new() @@ -140,9 +140,7 @@ mod linux { // Establish and subscribe once; iterations measure steady state. let (_session, mut sub) = worker .block_on(async { - let conn = quic::client::connect(&handle, client_sock, &dial) - .await - .expect("quic connect"); + let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); let (session, driver) = moq_net::Client::new() .with_subscriber(sub_origin.clone()) .connect_lite(std::time::Instant::now(), quic::web::Session::raw(conn)) diff --git a/rs/moq-uring/src/lib.rs b/rs/moq-uring/src/lib.rs index 33a6865260..b825b22c76 100644 --- a/rs/moq-uring/src/lib.rs +++ b/rs/moq-uring/src/lib.rs @@ -16,9 +16,13 @@ //! serves many connections on one socket (demuxed by connection id, dials //! included), each a [`quic::Connection`] implementing the transport traits, //! so `moq_net::Client::connect_lite` and `Server::accept_lite` run real -//! moq-lite sessions on the worker. [`Handle::run`] supplies time and schedules driver wakeups; -//! callers run the returned drivers with [`Handle::spawn`]. The stack underneath is enabled by -//! the `noq` feature; a build without it leaves the module out. +//! moq-lite sessions on the worker. The socket is the identity: whatever is +//! built on it runs on the worker that adopted it, and a socket adopted as a +//! member of a steered reuseport group ([`udp::Bound`]) issues connection ids +//! that steer back to it. [`Handle::run`] supplies time and schedules driver +//! wakeups; callers run the returned drivers with [`Handle::spawn`]. The stack +//! underneath is enabled by the `noq` feature; a build without it leaves the +//! module out. //! //! [`metrics::Metrics`] is how the worker's own health leaves its thread: //! relaxed counters for the buffer pools, the batching mechanisms, the ring, diff --git a/rs/moq-uring/src/quic/client.rs b/rs/moq-uring/src/quic/client.rs index fc22e3fffb..29ca017170 100644 --- a/rs/moq-uring/src/quic/client.rs +++ b/rs/moq-uring/src/quic/client.rs @@ -3,7 +3,7 @@ use std::net::SocketAddr; use super::{Connection, Error, Identity}; -use crate::{Handle, udp}; +use crate::udp; /// Where to dial, as whom, and who to trust. #[derive(Clone, Debug)] @@ -52,10 +52,10 @@ impl Config { /// /// Shorthand for a dial-only [`Endpoint`](super::Endpoint) and one /// [`connect`](super::Endpoint::connect) through it. The connection's driver -/// runs as a task on the worker behind `handle`, so the returned +/// runs as a task on the worker that adopted `socket`, so the returned /// [`Connection`] just works: hand it to `moq_net::Client::connect_lite` or /// use the stream API directly. -pub async fn connect(handle: &Handle, socket: udp::Socket, config: &Config) -> Result { - let endpoint = super::Endpoint::new(handle, socket, super::endpoint::Config::default())?; +pub async fn connect(socket: udp::Socket, config: &Config) -> Result { + let endpoint = super::Endpoint::new(socket, super::endpoint::Config::default())?; endpoint.connect(config).await } diff --git a/rs/moq-uring/src/quic/endpoint.rs b/rs/moq-uring/src/quic/endpoint.rs index e2d4aa12ed..bd5a013fe2 100644 --- a/rs/moq-uring/src/quic/endpoint.rs +++ b/rs/moq-uring/src/quic/endpoint.rs @@ -13,12 +13,15 @@ //! a migrating client stays routable; an Initial for an unsupported version //! gets a version negotiation packet back. //! -//! An endpoint whose socket is a member of a steered `SO_REUSEPORT` group -//! (one worker per core on one port) sets [`Config::shard`], and every id it -//! issues then carries the [`moq_sock::shard::cid_prefix`] steering byte, so -//! the kernel keeps delivering a connection's packets to the worker that owns -//! it. Dials through the endpoint carry it too, which is what steers a -//! cluster peer's responses back to the dialing worker. +//! The socket is the endpoint's identity. Its worker is where the demux and +//! every connection driver run, whichever thread built the endpoint. A socket +//! adopted as a member of a steered `SO_REUSEPORT` group (one worker per core +//! on one port; see [`udp::Bound`](crate::udp::Bound)) brings its slot along, +//! and every id the endpoint issues then carries the +//! [`moq_sock::shard::cid_prefix`] steering byte, so the kernel keeps +//! delivering a connection's packets to the worker that owns it. Dials +//! through the endpoint carry it too, which is what steers a cluster peer's +//! responses back to the dialing worker. use super::server; @@ -44,15 +47,6 @@ pub struct Config { /// peer retransmits and lands once the application drains the backlog or /// a handshake fails. pub backlog: usize, - - /// This socket's slot in a steered `SO_REUSEPORT` group, if it is in one. - /// - /// Every connection id the endpoint issues then leads with the slot's - /// [`cid_prefix`](moq_sock::shard::cid_prefix) byte, which is what the - /// group's filter selects on. Set it if and only if the socket was bound - /// into a group with this shard; a lone socket leaves it `None` and keeps - /// the whole id random. - pub shard: Option, } impl Default for Config { @@ -60,7 +54,6 @@ impl Default for Config { Self { server: None, backlog: 1024, - shard: None, } } } @@ -71,16 +64,10 @@ impl Config { self.server = Some(server); self } - - /// Issue connection ids steering to `shard`'s slot of a reuseport group. - pub fn with_shard(mut self, shard: moq_sock::shard::Shard) -> Self { - self.shard = Some(shard); - self - } } /// A fresh [`CID_LEN`]-byte connection id, leading with the steering prefix -/// when the endpoint sits in a reuseport group. +/// when the endpoint's socket is a member of a reuseport group. pub(crate) fn cid(shard: Option) -> [u8; CID_LEN] { let mut cid: [u8; CID_LEN] = rand::random(); if let Some(shard) = shard { diff --git a/rs/moq-uring/src/quic/noq/connection.rs b/rs/moq-uring/src/quic/noq/connection.rs index c8ddffd3ab..123a3daa10 100644 --- a/rs/moq-uring/src/quic/noq/connection.rs +++ b/rs/moq-uring/src/quic/noq/connection.rs @@ -11,7 +11,8 @@ use rustc_hash::FxHashMap; use super::super::{Error, SEGMENT}; use super::endpoint; -use crate::{Handle, udp}; +use crate::udp; +use crate::worker::Owner; /// The state shared by every handle and the driver, single-threaded behind /// `Rc`. @@ -28,6 +29,9 @@ const TRAIN_SEGMENTS: usize = 63; pub(crate) struct Inner { pub(crate) conn: RefCell, pub(crate) state: RefCell, + /// The worker driving this connection, which anything layered on it + /// (the WebTransport handshake, say) runs on too. + pub(crate) owner: Owner, } pub(crate) struct State { @@ -285,6 +289,11 @@ impl Connection { self.shared.close_code(code, reason); } + /// The worker driving this connection. + pub(crate) fn owner(&self) -> &Owner { + &self.shared.owner + } + /// The peer's certificate chain in DER, leaf first, or `None` if it /// presented none. /// @@ -320,7 +329,7 @@ impl Clone for Connection { /// [kicks](Inner::kick) the driver. The caller spawns the future and reclaims /// the connection's bookkeeping once it resolves. pub(crate) fn launch( - handle: &Handle, + owner: &Owner, socket: Rc, endpoint: Weak, key: ConnectionHandle, @@ -329,6 +338,7 @@ pub(crate) fn launch( let shared = Rc::new(Inner { conn: RefCell::new(conn), state: RefCell::new(State::new()), + owner: owner.clone(), }); let mut driver = Driver { @@ -336,7 +346,7 @@ pub(crate) fn launch( socket, endpoint, key, - deadline: crate::Timer::new(handle), + deadline: owner.timer(), scratch: Vec::with_capacity(TRAIN_SEGMENTS * SEGMENT), blocked: false, }; diff --git a/rs/moq-uring/src/quic/noq/endpoint.rs b/rs/moq-uring/src/quic/noq/endpoint.rs index ca820c4079..936e401993 100644 --- a/rs/moq-uring/src/quic/noq/endpoint.rs +++ b/rs/moq-uring/src/quic/noq/endpoint.rs @@ -23,7 +23,9 @@ use rustc_hash::FxHashMap; use super::super::{Error, endpoint::Config}; use super::connection; use crate::quic::Connection; -use crate::{Handle, udp}; +use crate::shared::Shared; +use crate::udp; +use crate::worker::Owner; /// The accept side: connections whose handshake finished and nobody has /// claimed yet. @@ -34,7 +36,8 @@ struct Accepting { /// State shared by the handles, the demux task, and the per-connection /// teardown tasks. pub(crate) struct Inner { - handle: Handle, + /// The socket's worker, where every task of this endpoint runs. + owner: Owner, socket: Rc, local: SocketAddr, /// The routing table, and the server configuration it accepts with. @@ -71,8 +74,17 @@ pub struct Endpoint { } impl Endpoint { - /// Serve `socket` on the worker behind `handle`. - pub fn new(handle: &Handle, socket: udp::Socket, config: Config) -> Result { + /// Serve `socket` on the worker that adopted it. + /// + /// The socket names the worker: the demux task and every connection + /// driver run there, and a socket adopted as a reuseport member steers + /// the ids this endpoint issues to itself. Refused once that worker has + /// been dropped, since nothing would ever drive the endpoint. + pub fn new(socket: udp::Socket, config: Config) -> Result { + let owner = socket.owner(); + let Some(handle) = owner.handle() else { + return Err(Error::Io(Shared::gone_error().to_string())); + }; let local = socket.local_addr().map_err(|err| Error::Io(err.to_string()))?; let server = match &config.server { Some(server) => { @@ -87,10 +99,10 @@ impl Endpoint { let accepting = server.is_some().then(|| Accepting { queue: VecDeque::new() }); // MTU discovery is off (the GSO pool sends fixed SEGMENT datagrams), // so the endpoint has no reason to allow it either. - let endpoint = noq_proto::Endpoint::new(super::endpoint_config(config.shard)?, server, false); + let endpoint = noq_proto::Endpoint::new(super::endpoint_config(socket.shard())?, server, false); let inner = Rc::new(Inner { - handle: handle.clone(), + owner, socket: Rc::new(socket), local, endpoint: RefCell::new(endpoint), @@ -119,11 +131,11 @@ impl Endpoint { /// /// Fails immediately on an endpoint built without a /// [`server`](Config::server) configuration, and with the socket's error - /// once the endpoint has died. + /// once the endpoint has died, its worker included. pub async fn accept(&self) -> Result { kio::wait(|waiter| { - if let Some(err) = &*self.inner.closed.borrow() { - return Poll::Ready(Err(err.clone())); + if let Some(err) = self.inner.closed() { + return Poll::Ready(Err(err)); } let mut accepting = self.inner.accepting.borrow_mut(); let Some(accepting) = accepting.as_mut() else { @@ -141,8 +153,8 @@ impl Endpoint { /// Dial [`Config::peer`](crate::quic::client::Config::peer) through this /// endpoint's socket, driving the handshake to completion. pub async fn connect(&self, config: &crate::quic::client::Config) -> Result { - if let Some(err) = &*self.inner.closed.borrow() { - return Err(err.clone()); + if let Some(err) = self.inner.closed() { + return Err(err); } let client = super::client_config(config)?; let (key, conn) = self @@ -195,6 +207,18 @@ impl std::fmt::Debug for Endpoint { } impl Inner { + /// Why nothing new can start here: the socket's terminal error, or the + /// worker being gone, which no task would ever report since none runs. + fn closed(&self) -> Option { + if let Some(err) = &*self.closed.borrow() { + return Some(err.clone()); + } + self.owner + .handle() + .is_none() + .then(|| Error::Io(Shared::gone_error().to_string())) + } + /// The demux task: receive, route, and stop once nothing needs us. fn poll_run(self: &Rc, waiter: &kio::Waiter) -> Poll<()> { // Handle drops and connection teardowns re-check the exit condition. @@ -320,7 +344,7 @@ impl Inner { // Hand the connection over once (if) its handshake completes. self.pending.set(self.pending.get() + 1); let inner = self.clone(); - self.handle.spawn(async move { + self.owner.spawn(async move { let outcome = connection::establish(shared).await; inner.pending.set(inner.pending.get() - 1); let mut conn = match outcome { @@ -372,11 +396,11 @@ impl Inner { /// Register `conn`, spawn its driver, and arrange its teardown. fn launch(self: &Rc, key: ConnectionHandle, conn: noq_proto::Connection) -> connection::Shared { - let (shared, driver) = connection::launch(&self.handle, self.socket.clone(), Rc::downgrade(self), key, conn); + let (shared, driver) = connection::launch(&self.owner, self.socket.clone(), Rc::downgrade(self), key, conn); self.conns.borrow_mut().insert(key, shared.clone()); let inner = self.clone(); - self.handle.spawn(async move { + self.owner.spawn(async move { driver.await; inner.release(key); }); diff --git a/rs/moq-uring/src/quic/server.rs b/rs/moq-uring/src/quic/server.rs index 267a0f897c..4fd27a0034 100644 --- a/rs/moq-uring/src/quic/server.rs +++ b/rs/moq-uring/src/quic/server.rs @@ -1,7 +1,7 @@ //! Accepting: everything one incoming connection needs. use super::{Connection, Error, Identity}; -use crate::{Handle, udp}; +use crate::udp; /// Whether connecting clients are asked for a certificate, and against what. /// @@ -83,11 +83,7 @@ impl Config { /// and one [`accept`](super::Endpoint::accept) from it: later arrivals on the /// socket keep reaching the accepted connection, but nothing else is ever /// accepted. Keep the endpoint itself for a listener. -pub async fn accept(handle: &Handle, socket: udp::Socket, config: &Config) -> Result { - let endpoint = super::Endpoint::new( - handle, - socket, - super::endpoint::Config::default().with_server(config.clone()), - )?; +pub async fn accept(socket: udp::Socket, config: &Config) -> Result { + let endpoint = super::Endpoint::new(socket, super::endpoint::Config::default().with_server(config.clone()))?; endpoint.accept().await } diff --git a/rs/moq-uring/src/quic/web.rs b/rs/moq-uring/src/quic/web.rs index 7bbe65c474..8f4261e67d 100644 --- a/rs/moq-uring/src/quic/web.rs +++ b/rs/moq-uring/src/quic/web.rs @@ -23,7 +23,6 @@ use bytes::{Buf, Bytes, BytesMut}; use web_transport_proto as proto; use super::{Connection, Error}; -use crate::Handle; /// The frame type WebTransport bidirectional streams lead with. const FRAME_WEBTRANSPORT: u64 = 0x41; @@ -73,7 +72,6 @@ const H3_NO_ERROR: u64 = 0x0100; /// `h3`. Answer with [`respond`](Self::respond) (or [`ok`](Self::ok)) to get /// the [`Session`], or [`reject`](Self::reject) to refuse it. pub struct Request { - handle: Handle, conn: Connection, /// Closes the connection on every way out of here that is not an answer. guard: Guard, @@ -129,16 +127,17 @@ impl Request { /// Run the server side of the HTTP/3 handshake: exchange SETTINGS, then /// take the CONNECT request. /// - /// There is no timeout here; a peer that stalls mid-handshake is bounded - /// by the connection's idle timeout. - pub async fn accept(handle: &Handle, conn: Connection) -> Result { + /// Everything the session later spawns runs on the worker already driving + /// `conn`. There is no timeout here; a peer that stalls mid-handshake is + /// bounded by the connection's idle timeout. + pub async fn accept(conn: Connection) -> Result { // Nothing else will close it. The endpoint keeps a connection, its // routes, and its driver task until the driver sees a terminal state, // and the backlog stopped counting this one when it was accepted, so a // peer that keeps sending would otherwise hold a rejected handshake // open for as long as it liked. let failed = conn.clone(); - match Self::handshake(handle, conn).await { + match Self::handshake(conn).await { Ok(request) => Ok(request), Err(err) => { failed.close_code(H3_GENERAL_PROTOCOL_ERROR, &err.to_string()); @@ -147,7 +146,7 @@ impl Request { } } - async fn handshake(handle: &Handle, mut conn: Connection) -> Result { + async fn handshake(mut conn: Connection) -> Result { // Our control stream: the SETTINGS advertising WebTransport support. let mut control = open_uni(&mut conn).await?; let mut settings = proto::Settings::default(); @@ -242,7 +241,6 @@ impl Request { let request = read_connect(&mut recv).await?; Ok(Self { - handle: handle.clone(), guard: Guard::new(conn.clone()), conn, request, @@ -311,7 +309,7 @@ impl Request { // The guard stays armed across the wait below. Cancelling this future // mid-grace would otherwise skip the deliberate close and leak the // connection, which is the very thing the guard is here to prevent. - let mut deadline = crate::Timer::after(&self.handle, CLOSE_GRACE); + let mut deadline = self.conn.owner().after(CLOSE_GRACE); let send = &mut self.send; kio::wait(|waiter| { let mut cx = Context::from_waker(waiter.waker()); @@ -389,7 +387,6 @@ impl std::fmt::Debug for Request { /// The WebTransport layering shared by a session's clones. struct Web { - handle: Handle, /// The CONNECT stream's id: what every stream header and datagram carries. session_id: u64, /// Precomputed per-kind prefixes carrying the session id. @@ -445,7 +442,6 @@ impl Session { /// Assemble the web flavor and spawn its capsule reader. fn establish(request: Request, protocol: Option) -> Self { let Request { - handle, conn, guard: _, request: _, @@ -475,7 +471,6 @@ impl Session { .collect(); let web = Rc::new(Web { - handle: handle.clone(), session_id, header_uni: header_uni.into(), header_bi: header_bi.into(), @@ -499,7 +494,8 @@ impl Session { // stream; read it so the close code survives the H3 mapping. let capsules = web.clone(); let capsule_conn = conn.clone(); - handle.spawn(async move { read_capsules(capsules, capsule_conn, recv).await }); + conn.owner() + .spawn(async move { read_capsules(capsules, capsule_conn, recv).await }); Self { conn, @@ -735,10 +731,10 @@ impl web_transport_trait::poll::Session for Session { // the task rather than here because flow control can take it in // pieces, and abandoning a partial frame would leave the browser with // neither the code nor the reason. - let mut deadline = crate::Timer::after(&web.handle, CLOSE_GRACE); + let mut deadline = self.conn.owner().after(CLOSE_GRACE); let reason = reason.to_string(); let mut conn = self.conn.clone(); - web.handle.spawn(async move { + self.conn.owner().spawn(async move { let mut offset = 0; kio::wait(|waiter| { let mut cx = Context::from_waker(waiter.waker()); diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index a52ddf70e6..9ca5de176a 100644 --- a/rs/moq-uring/src/udp.rs +++ b/rs/moq-uring/src/udp.rs @@ -37,7 +37,7 @@ use std::io; use std::net::{IpAddr, SocketAddr, SocketAddrV6, UdpSocket}; use std::os::fd::AsRawFd; use std::ptr::NonNull; -use std::rc::{Rc, Weak}; +use std::rc::Rc; use std::sync::atomic::{AtomicU16, Ordering}; use std::task::Poll; @@ -46,6 +46,7 @@ use io_uring::{cqueue, opcode, types}; use crate::Error; use crate::metrics::Counters; use crate::shared::{Cqe, Op, Shared}; +use crate::worker::Owner; /// Space reserved for received control messages: `UDP_GRO` plus the packet's /// `IP_TOS` or `IPV6_TCLASS` (the kernel emits one or the other), two ints. @@ -374,12 +375,42 @@ impl TxSlot { } } +/// A bound socket a worker takes over, with whatever identity it carries. +/// +/// Both variants convert with `From`, so [`crate::Handle::udp`] takes either +/// as is. The member is the only way a socket gets a steering slot: the +/// group that completed it is what proved the slot is real. +pub enum Bound { + /// A socket on its own, bound by the caller. + Lone(UdpSocket), + /// One member of a completed steered `SO_REUSEPORT` group. Every + /// connection id an endpoint on it issues then leads with the member's + /// [`cid_prefix`](moq_sock::shard::cid_prefix), so the group's filter + /// keeps delivering a connection's packets to this socket. + Member(moq_sock::shard::Socket), +} + +impl From for Bound { + fn from(socket: UdpSocket) -> Self { + Self::Lone(socket) + } +} + +impl From for Bound { + fn from(member: moq_sock::shard::Socket) -> Self { + Self::Member(member) + } +} + /// Everything both the [`Socket`] handle and in-flight ops keep alive. pub(crate) struct SockShared { io: UdpSocket, - worker: Weak, + /// The worker, as the I/O built on this socket carries it. + owner: Owner, + /// This socket's slot in a steered reuseport group, if it is in one. + shard: Option, /// The worker's counters, held directly rather than reached through - /// `worker`, so counting a datagram is not a `Weak::upgrade`. + /// `owner`, so counting a datagram is not a `Weak::upgrade`. metrics: std::sync::Arc, config: Config, bgid: u16, @@ -392,10 +423,7 @@ impl SockShared { /// Whether the worker loop that would drive this socket is gone: dropped /// outright, or torn down while handles keep the shared state alive. fn worker_gone(&self) -> bool { - match self.worker.upgrade() { - Some(shared) => shared.stopped.get(), - None => true, - } + self.owner.handle().is_none() } /// A packet released its buffer slice. @@ -408,7 +436,7 @@ impl SockShared { // A receive that died on ENOBUFS can start again now. if rx.armed.is_none() && rx.error.is_none() { drop(rx); - if let Some(shared) = self.worker.upgrade() { + if let Some(shared) = self.owner.upgrade() { arm_recv(&shared, self); } } @@ -456,7 +484,7 @@ impl Drop for SockShared { // Every op referencing our buffers has completed (ops own an `Rc` of // us), so the kernel is done; give the buffer group id back. if self.rx.borrow().ring.is_some() - && let Some(shared) = self.worker.upgrade() + && let Some(shared) = self.owner.upgrade() { let ring = shared.ring.borrow_mut(); let _ = ring.submitter().unregister_buf_ring(self.bgid); @@ -475,11 +503,18 @@ pub struct Socket { impl Socket { /// A test-only observer for whether every kernel operation released this socket. #[cfg(test)] - pub(crate) fn downgrade(&self) -> Weak { + pub(crate) fn downgrade(&self) -> std::rc::Weak { Rc::downgrade(&self.shared) } - pub(crate) fn bind(shared: &Rc, io: UdpSocket, config: Config) -> Result { + pub(crate) fn bind(shared: &Rc, bound: Bound, config: Config) -> Result { + let (io, shard) = match bound { + Bound::Lone(io) => (io, None), + Bound::Member(member) => { + let shard = member.shard(); + (member.into_inner(), Some(shard)) + } + }; let floor = if config.gro { MAX_RECV + RECV_OVERHEAD } else { 2048 }; if config.rx_buffer_len < floor || config.rx_buffers_max == 0 || config.tx_buffers_max == 0 { return Err(io::Error::new( @@ -570,7 +605,8 @@ impl Socket { let sock = Rc::new(SockShared { io, - worker: Rc::downgrade(shared), + owner: Owner::new(shared), + shard, metrics: shared.metrics.clone(), config, bgid, @@ -600,6 +636,16 @@ impl Socket { self.shared.io.local_addr() } + /// The worker driving this socket, which everything built on it runs on. + pub(crate) fn owner(&self) -> Owner { + self.shared.owner.clone() + } + + /// This socket's slot in a steered reuseport group, if it is in one. + pub(crate) fn shard(&self) -> Option { + self.shared.shard + } + /// A received packet, or the socket's terminal error, registering `waiter` /// while neither is available. Queued packets drain before an error /// surfaces. @@ -682,7 +728,7 @@ impl Drop for Socket { fn drop(&mut self) { self.shared.closed.set(true); let rx = self.shared.rx.borrow(); - if let (Some(key), Some(shared)) = (rx.armed, self.shared.worker.upgrade()) { + if let (Some(key), Some(shared)) = (rx.armed, self.shared.owner.upgrade()) { drop(rx); // Fire-and-forget: the cancel's own CQE is consumed by the worker, // and the receive's terminal CQE releases the socket state. @@ -802,7 +848,7 @@ impl TxBuf { ), )); } - let shared = match self.sock.worker.upgrade() { + let shared = match self.sock.owner.upgrade() { Some(shared) if !shared.stopped.get() => shared, _ => return Err(Shared::gone_error()), }; diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index 8dd910e1fb..3a85fb00ec 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -1,7 +1,7 @@ //! The per-thread worker: ring ownership, the drive loop, and parking. use std::cell::RefCell; -use std::rc::Rc; +use std::rc::{Rc, Weak}; use std::sync::atomic::Ordering; use std::task::{Context, Poll}; use std::time::{Duration, Instant}; @@ -522,11 +522,72 @@ impl Handle { /// /// The caller configures and binds the socket (options, addresses); this /// takes over receive and send. `config` picks the batching mechanisms. - pub fn udp(&self, socket: std::net::UdpSocket, config: udp::Config) -> Result { + /// + /// The socket is what names this worker from here on: an + /// [`Endpoint`](crate::quic::Endpoint) built on it runs its tasks here, + /// whichever thread's handle built it. A member of a steered reuseport + /// group ([`moq_sock::shard::Socket`]) brings its slot along, so the + /// connection ids issued through it steer back to this socket. + pub fn udp(&self, socket: impl Into, config: udp::Config) -> Result { if self.shared.stopped.get() { return Err(Shared::gone_error().into()); } - udp::Socket::bind(&self.shared, socket, config) + udp::Socket::bind(&self.shared, socket.into(), config) + } +} + +/// The worker behind a socket, endpoint, or connection, held weakly. +/// +/// I/O carries its owner so it cannot be driven through a different worker, +/// but a handle to that I/O must not keep a dropped worker's ring alive, so +/// this holds no strong reference. Once the worker is gone, spawning is a +/// no-op (like [`Handle::spawn`]) and timers never fire, which is what the +/// tasks that would have consumed them expect. +#[derive(Clone)] +pub(crate) struct Owner { + shared: Weak, + /// Held directly: a timer on a dropped worker still has to exist, since + /// the driver that owns it is torn down by the same drop that would need + /// it. + timers: Rc>, +} + +impl Owner { + pub(crate) fn new(shared: &Rc) -> Self { + Self { + shared: Rc::downgrade(shared), + timers: shared.timers.clone(), + } + } + + /// The worker's core while it is still allocated, torn down or not. + pub fn upgrade(&self) -> Option> { + self.shared.upgrade() + } + + /// A strong handle, or `None` once the worker is dropped or torn down. + pub fn handle(&self) -> Option { + let shared = self.shared.upgrade()?; + (!shared.stopped.get()).then_some(Handle { shared }) + } + + /// Run a `!Send` future on the worker, or drop it if the worker is gone. + pub fn spawn(&self, future: impl Future + 'static) { + if let Some(handle) = self.handle() { + handle.spawn(future); + } + } + + /// A disarmed timer on the worker. + pub fn timer(&self) -> crate::Timer { + crate::Timer::from_heap(self.timers.clone()) + } + + /// A timer that expires after `duration`. + pub fn after(&self, duration: Duration) -> crate::Timer { + let mut timer = self.timer(); + timer.set(Instant::now().checked_add(duration)); + timer } } diff --git a/rs/moq-uring/tests/echo.rs b/rs/moq-uring/tests/echo.rs index 17112c7ad9..3f2197fdad 100644 --- a/rs/moq-uring/tests/echo.rs +++ b/rs/moq-uring/tests/echo.rs @@ -59,7 +59,7 @@ fn echo_noq_peer() { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket"); let endpoint = - quic::Endpoint::new(&handle, socket, quic::endpoint::Config::default().with_server(server)).expect("endpoint"); + quic::Endpoint::new(socket, quic::endpoint::Config::default().with_server(server)).expect("endpoint"); let addr = endpoint.local_addr(); let payload: Vec = (0..PAYLOAD).map(|i| (i * 31 % 251) as u8).collect(); let expected = payload.clone(); @@ -91,7 +91,7 @@ fn echo_noq_peer() { worker .block_on(async move { let conn = endpoint.accept().await.expect("accept"); - let request = quic::web::Request::accept(&handle, conn).await.expect("handshake"); + let request = quic::web::Request::accept(conn).await.expect("handshake"); let mut session = request.ok().await.expect("respond"); let (mut send, mut recv) = std::future::poll_fn(|cx| session.poll_accept_bi(cx)) .await diff --git a/rs/moq-uring/tests/endpoint.rs b/rs/moq-uring/tests/endpoint.rs index 423394b071..155f268c86 100644 --- a/rs/moq-uring/tests/endpoint.rs +++ b/rs/moq-uring/tests/endpoint.rs @@ -60,7 +60,6 @@ fn dial_and_accept_share_one_socket() { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket"); quic::Endpoint::new( - handle, sock, quic::endpoint::Config::default().with_server(server_config(&certs)), ) @@ -102,13 +101,11 @@ fn a_close_reaches_both_ends_with_its_code() { .expect("socket") }; let server = quic::Endpoint::new( - &handle, socket(&handle), quic::endpoint::Config::default().with_server(server_config(&certs)), ) .expect("server endpoint"); - let client = - quic::Endpoint::new(&handle, socket(&handle), quic::endpoint::Config::default()).expect("dial-only endpoint"); + let client = quic::Endpoint::new(socket(&handle), quic::endpoint::Config::default()).expect("dial-only endpoint"); worker .block_on(async move { @@ -142,7 +139,6 @@ fn unsupported_version_is_negotiated_only_by_servers() { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket"); let endpoint = quic::Endpoint::new( - &handle, sock, quic::endpoint::Config::default().with_server(server_config(&certs)), ) @@ -151,8 +147,7 @@ fn unsupported_version_is_negotiated_only_by_servers() { let dial_only_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("dial-only socket"); - let dial_only = - quic::Endpoint::new(&handle, dial_only_sock, quic::endpoint::Config::default()).expect("dial-only endpoint"); + let dial_only = quic::Endpoint::new(dial_only_sock, quic::endpoint::Config::default()).expect("dial-only endpoint"); let dial_only_addr = dial_only.local_addr(); // The raw client runs on its own thread (the worker owns this one) and @@ -284,13 +279,11 @@ fn a_peer_stop_reaches_the_accepted_stream() { .expect("socket") }; let server = quic::Endpoint::new( - &handle, socket(&handle), quic::endpoint::Config::default().with_server(server_config(&certs)), ) .expect("server endpoint"); - let client = - quic::Endpoint::new(&handle, socket(&handle), quic::endpoint::Config::default()).expect("dial-only endpoint"); + let client = quic::Endpoint::new(socket(&handle), quic::endpoint::Config::default()).expect("dial-only endpoint"); worker .block_on(async move { @@ -359,7 +352,7 @@ fn an_unanswered_dial_times_out() { dial.transport.idle_timeout = Duration::from_millis(500); let result = worker - .block_on(async move { quic::client::connect(&handle, client_sock, &dial).await }) + .block_on(async move { quic::client::connect(client_sock, &dial).await }) .expect("worker"); assert!(result.is_err(), "a dial into a black hole must time out"); } @@ -378,7 +371,7 @@ fn the_backlog_bounds_pending_handshakes() { .expect("socket"); let mut config = quic::endpoint::Config::default().with_server(server_config(&certs)); config.backlog = 0; - let endpoint = quic::Endpoint::new(&handle, sock, config).expect("endpoint"); + let endpoint = quic::Endpoint::new(sock, config).expect("endpoint"); let server = endpoint.local_addr(); let accepted = std::rc::Rc::new(std::cell::Cell::new(false)); @@ -398,7 +391,7 @@ fn the_backlog_bounds_pending_handshakes() { dial.transport.idle_timeout = Duration::from_millis(500); let result = worker - .block_on(async move { quic::client::connect(&handle, client_sock, &dial).await }) + .block_on(async move { quic::client::connect(client_sock, &dial).await }) .expect("worker"); assert!(result.is_err(), "a handshake over the backlog must not complete"); assert!(!accepted.get(), "a handshake over the backlog must not be accepted"); @@ -417,14 +410,14 @@ fn the_backlog_bounds_queued_connections() { .expect("socket"); let mut config = quic::endpoint::Config::default().with_server(server_config(&certs)); config.backlog = 1; - let endpoint = quic::Endpoint::new(&handle, sock, config).expect("endpoint"); + let endpoint = quic::Endpoint::new(sock, config).expect("endpoint"); let server = endpoint.local_addr(); let first_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("first client socket"); let first = worker - .block_on(async { quic::client::connect(&handle, first_sock, &dial_config(server)).await }) + .block_on(async { quic::client::connect(first_sock, &dial_config(server)).await }) .expect("worker") .expect("first connection"); @@ -434,7 +427,7 @@ fn the_backlog_bounds_queued_connections() { let mut second_config = dial_config(server); second_config.transport.idle_timeout = Duration::from_millis(500); let second = worker - .block_on(async { quic::client::connect(&handle, second_sock, &second_config).await }) + .block_on(async { quic::client::connect(second_sock, &second_config).await }) .expect("worker"); assert!(second.is_err(), "a completed connection must still occupy the backlog"); @@ -459,7 +452,6 @@ fn connections_share_one_tx_buffer_fairly() { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp_config) .expect("socket"); let endpoint = quic::Endpoint::new( - &handle, sock, quic::endpoint::Config::default().with_server(server_config(&certs)), ) @@ -471,7 +463,7 @@ fn connections_share_one_tx_buffer_fairly() { let first_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("first client socket"); - let first_client = quic::client::connect(&handle, first_sock, &dial_config(server)) + let first_client = quic::client::connect(first_sock, &dial_config(server)) .await .expect("first connection"); let mut first_server = endpoint.accept().await.expect("first accepted connection"); @@ -542,7 +534,7 @@ fn connections_share_one_tx_buffer_fairly() { let mut second_config = dial_config(server); second_config.transport.idle_timeout = Duration::from_millis(500); let started = Instant::now(); - let second = quic::client::connect(&handle, second_sock, &second_config).await; + let second = quic::client::connect(second_sock, &second_config).await; let elapsed = started.elapsed(); running.set(false); let second = second.expect("the second connection must not starve"); @@ -565,7 +557,7 @@ fn accept_needs_a_server_config() { let sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket"); - let endpoint = quic::Endpoint::new(&handle, sock, quic::endpoint::Config::default()).expect("endpoint"); + let endpoint = quic::Endpoint::new(sock, quic::endpoint::Config::default()).expect("endpoint"); let result = worker.block_on(async move { endpoint.accept().await }).expect("worker"); assert!(matches!(result, Err(quic::Error::NotServer)), "got {result:?}"); @@ -594,12 +586,7 @@ fn an_identity_is_read_once() { let sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket"); - quic::Endpoint::new( - handle, - sock, - quic::endpoint::Config::default().with_server(config.clone()), - ) - .expect("endpoint") + quic::Endpoint::new(sock, quic::endpoint::Config::default().with_server(config.clone())).expect("endpoint") }; // Two endpoints, as two workers would build them, then a real handshake @@ -614,11 +601,8 @@ fn an_identity_is_read_once() { let sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("client socket"); - let dial = handle.clone(); handle.spawn(async move { - quic::client::connect(&dial, sock, &dial_config(server)) - .await - .expect("dial"); + quic::client::connect(sock, &dial_config(server)).await.expect("dial"); }); second.accept().await.expect("accepted connection"); }) diff --git a/rs/moq-uring/tests/identity.rs b/rs/moq-uring/tests/identity.rs new file mode 100644 index 0000000000..5a08c4966a --- /dev/null +++ b/rs/moq-uring/tests/identity.rs @@ -0,0 +1,157 @@ +//! A socket names its worker, and everything built on it follows: an +//! endpoint runs where its socket was adopted, whichever handle is in scope, +//! and a socket whose worker is gone refuses to start anything. +//! +//! Kernel-gated: skips loudly below the Linux 6.12 floor (GitHub-hosted CI), +//! and runs everywhere else. + +#![cfg(all(target_os = "linux", feature = "noq"))] + +#[path = "support.rs"] +mod support; + +use std::net::UdpSocket; +use std::task::{Context, Poll, Waker}; +use std::time::Duration; + +use moq_uring::{Config, Error, Worker, quic, udp}; + +fn worker() -> Option { + match Worker::new(Config::default()) { + Ok(worker) => Some(worker), + Err(Error::Unsupported(reason)) => { + eprintln!("skipping io_uring identity test: {reason}"); + None + } + Err(err) => panic!("worker setup failed: {err}"), + } +} + +const ALPN: &str = "moq-uring-identity"; + +fn server_config(certs: &support::Certs) -> quic::server::Config { + let mut config = quic::server::Config::new(quic::Identity::open(&certs.cert, &certs.key).expect("identity")); + config.alpn = vec![ALPN.to_string()]; + config +} + +fn dial_config(peer: std::net::SocketAddr) -> quic::client::Config { + let mut config = quic::client::Config::new(peer, "localhost"); + config.alpn = vec![ALPN.to_string()]; + config.verify = false; + config +} + +fn socket(handle: &moq_uring::Handle) -> udp::Socket { + handle + .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) + .expect("socket") +} + +/// Poll `future` once, the way a caller that never drives a worker would. +fn poll_once(future: F) -> Poll { + let mut future = std::pin::pin!(future); + future.as_mut().poll(&mut Context::from_waker(Waker::noop())) +} + +/// Building an endpoint on a socket whose worker has been dropped is refused +/// up front: nothing would ever run its demux, so a dial or accept through it +/// could only hang. +#[test] +fn an_endpoint_refuses_a_stopped_worker() { + let Some(worker) = worker() else { return }; + let sock = socket(&worker.handle()); + drop(worker); + + let err = quic::Endpoint::new(sock, quic::endpoint::Config::default()).expect_err("endpoint on a stopped worker"); + assert!( + matches!(err, quic::Error::Io(_)), + "refused as a dead socket, got {err:?}" + ); +} + +/// An endpoint that outlives its worker fails what it is asked for instead +/// of parking it forever, whether a dial or an accept. +#[test] +fn a_stopped_worker_fails_dials_and_accepts() { + let Some(worker) = worker() else { return }; + let certs = support::certs().expect("certificates"); + let sock = socket(&worker.handle()); + let endpoint = quic::Endpoint::new( + sock, + quic::endpoint::Config::default().with_server(server_config(&certs)), + ) + .expect("endpoint"); + let peer = endpoint.local_addr(); + drop(worker); + + match poll_once(endpoint.connect(&dial_config(peer))) { + Poll::Ready(Err(quic::Error::Io(_))) => {} + other => panic!("a dial on a stopped worker should fail at once, got {other:?}"), + } + match poll_once(endpoint.accept()) { + Poll::Ready(Err(quic::Error::Io(_))) => {} + other => panic!("an accept on a stopped worker should fail at once, got {other:?}"), + } +} + +/// Two workers on one thread: an endpoint on the second worker's socket makes +/// no progress while only the first is driven, however its futures are +/// polled, and completes once its own worker runs. +#[test] +fn an_endpoint_runs_on_the_worker_that_adopted_its_socket() { + let Some(mut bystander) = worker() else { return }; + let mut owner = worker().expect("a second worker on the same thread"); + let certs = support::certs().expect("certificates"); + + let sock = socket(&owner.handle()); + let endpoint = quic::Endpoint::new( + sock, + quic::endpoint::Config::default().with_server(server_config(&certs)), + ) + .expect("endpoint"); + let addr = endpoint.local_addr(); + + // A client on its own thread and worker, dialing the endpoint. Its + // Initial reaches the socket immediately; only the owner can read it. + let client = std::thread::spawn(move || { + let mut worker = Worker::new(Config::default()).expect("client worker"); + let sock = socket(&worker.handle()); + worker + .block_on(async move { + let mut conn = quic::client::connect(sock, &dial_config(addr)).await.expect("dial"); + web_transport_trait::poll::Session::close(&mut conn, 0, "done"); + }) + .expect("client loop"); + }); + + // Driving the bystander polls the accept from its loop, but the demux + // that would feed it is a task on the owner, which is not running. + let bystander_handle = bystander.handle(); + let stalled = bystander + .block_on(async { + let mut deadline = moq_uring::Timer::after(&bystander_handle, Duration::from_millis(500)); + let mut accept = std::pin::pin!(endpoint.accept()); + kio::wait(|waiter| { + let mut cx = Context::from_waker(waiter.waker()); + if accept.as_mut().poll(&mut cx).is_ready() { + return Poll::Ready(false); + } + deadline.poll(waiter).map(|()| true) + }) + .await + }) + .expect("bystander loop"); + assert!(stalled, "the bystander worker served an endpoint it does not own"); + + let accepted = owner + .block_on(async { endpoint.accept().await.expect("accept on the owner") }) + .expect("owner loop"); + assert_eq!( + web_transport_trait::poll::Session::protocol(&accepted), + Some(ALPN), + "negotiated ALPN" + ); + drop(accepted); + client.join().expect("client thread"); +} diff --git a/rs/moq-uring/tests/qlog.rs b/rs/moq-uring/tests/qlog.rs index e465842ef3..1ee6910dfb 100644 --- a/rs/moq-uring/tests/qlog.rs +++ b/rs/moq-uring/tests/qlog.rs @@ -70,11 +70,8 @@ fn a_connection_writes_a_trace() { dial.verify = false; dial.transport.qlog = Some(sink.clone()); - let accepting = handle.clone(); handle.spawn(async move { - let conn = quic::server::accept(&accepting, server_sock, &server) - .await - .expect("quic accept"); + let conn = quic::server::accept(server_sock, &server).await.expect("quic accept"); // Hold the connection open until the worker is dropped, so the trace // covers a live connection rather than a torn-down one. std::future::pending::<()>().await; @@ -83,9 +80,7 @@ fn a_connection_writes_a_trace() { worker .block_on(async move { - let conn = quic::client::connect(&handle, client_sock, &dial) - .await - .expect("quic connect"); + let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); assert_eq!( web_transport_trait::poll::Session::protocol(&conn), Some(ALPN), diff --git a/rs/moq-uring/tests/session.rs b/rs/moq-uring/tests/session.rs index 3e8c8af321..50fc783670 100644 --- a/rs/moq-uring/tests/session.rs +++ b/rs/moq-uring/tests/session.rs @@ -80,7 +80,7 @@ fn lite_session_over_the_worker() { // The publisher session runs as a worker task for the life of the test. let server_handle = handle.clone(); handle.spawn(async move { - let conn = quic::server::accept(&server_handle, server_sock, &server_config) + let conn = quic::server::accept(server_sock, &server_config) .await .expect("quic accept"); let (session, driver) = moq_net::Server::new() @@ -96,9 +96,7 @@ fn lite_session_over_the_worker() { let sub = sub_origin.clone(); let payload = worker .block_on(async move { - let conn = quic::client::connect(&handle, client_sock, &dial) - .await - .expect("quic connect"); + let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); assert_eq!( web_transport_trait::poll::Session::protocol(&conn), Some(ALPN), @@ -191,7 +189,6 @@ fn two_lite_sessions_share_the_server_socket() { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("server socket"); let endpoint = quic::Endpoint::new( - &handle, server_sock, quic::endpoint::Config::default().with_server(server_config), ) @@ -227,9 +224,7 @@ fn two_lite_sessions_share_the_server_socket() { let client_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("client socket"); - let conn = quic::client::connect(&handle, client_sock, &dial) - .await - .expect("quic connect"); + let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); let (session, driver) = moq_net::Client::new() .with_subscriber(sub.clone()) .connect_lite(std::time::Instant::now(), quic::web::Session::raw(conn)) @@ -298,18 +293,15 @@ fn configured_roots_verify_the_server() { dial.system_roots = false; dial.roots = vec![certs.cert.clone()]; - let server_handle = handle.clone(); handle.spawn(async move { - quic::server::accept(&server_handle, server_sock, &server_config) + quic::server::accept(server_sock, &server_config) .await .expect("quic accept"); }); worker .block_on(async move { - let conn = quic::client::connect(&handle, client_sock, &dial) - .await - .expect("quic connect"); + let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); assert_eq!( web_transport_trait::poll::Session::protocol(&conn), Some(ALPN), @@ -350,7 +342,6 @@ fn required_client_auth_refuses_an_anonymous_client() { .expect("client socket"); let endpoint = quic::Endpoint::new( - &handle, server_sock, quic::endpoint::Config::default().with_server(server_config), ) @@ -372,7 +363,7 @@ fn required_client_auth_refuses_an_anonymous_client() { worker .block_on(async move { - match quic::client::connect(&handle, client_sock, &dial).await { + match quic::client::connect(client_sock, &dial).await { // The dial itself was refused; done. Err(_) => {} // Established locally, but the server's refusal closes it. @@ -417,18 +408,15 @@ fn a_root_bundle_is_loaded_whole() { dial.system_roots = false; dial.roots = vec![bundle.roots.clone()]; - let server_handle = handle.clone(); handle.spawn(async move { - quic::server::accept(&server_handle, server_sock, &server_config) + quic::server::accept(server_sock, &server_config) .await .expect("quic accept"); }); worker .block_on(async move { - let conn = quic::client::connect(&handle, client_sock, &dial) - .await - .expect("quic connect"); + let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); assert_eq!( web_transport_trait::poll::Session::protocol(&conn), Some(ALPN), diff --git a/rs/moq-uring/tests/teardown.rs b/rs/moq-uring/tests/teardown.rs index 0946794b0f..aa761df561 100644 --- a/rs/moq-uring/tests/teardown.rs +++ b/rs/moq-uring/tests/teardown.rs @@ -109,7 +109,6 @@ fn a_published_close_has_already_left_the_client() { let handle = worker.handle(); let sock = handle.udp(server_sock, udp::Config::default()).expect("server socket"); let endpoint = quic::Endpoint::new( - &handle, sock, quic::endpoint::Config::default().with_server(server_config(&certs)), ) @@ -129,7 +128,7 @@ fn a_published_close_has_already_left_the_client() { udp::Config::default(), ) .expect("client socket"); - let endpoint = quic::Endpoint::new(&handle, sock, quic::endpoint::Config::default()).expect("client endpoint"); + let endpoint = quic::Endpoint::new(sock, quic::endpoint::Config::default()).expect("client endpoint"); client_worker .block_on(async move { diff --git a/rs/moq-uring/tests/web.rs b/rs/moq-uring/tests/web.rs index a4624039a5..6b9e357ae1 100644 --- a/rs/moq-uring/tests/web.rs +++ b/rs/moq-uring/tests/web.rs @@ -41,7 +41,7 @@ fn h3_endpoint(handle: &moq_uring::Handle, certs: &support::Certs) -> quic::Endp let sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket"); - quic::Endpoint::new(handle, sock, quic::endpoint::Config::default().with_server(server)).expect("endpoint") + quic::Endpoint::new(sock, quic::endpoint::Config::default().with_server(server)).expect("endpoint") } /// The tokio-side client, in its own runtime on its own thread. @@ -149,7 +149,7 @@ fn webtransport_echo_end_to_end() { let conn = endpoint.accept().await.expect("accept"); assert_eq!(conn.protocol(), Some("h3"), "negotiated ALPN"); - let request = quic::web::Request::accept(&handle, conn).await.expect("handshake"); + let request = quic::web::Request::accept(conn).await.expect("handshake"); assert_eq!(request.url().path(), "/echo"); assert_eq!(request.url().query(), Some("token=abc")); assert_eq!(request.protocols(), [PROTO.to_string()]); @@ -273,7 +273,7 @@ fn lite_session_over_webtransport() { worker .block_on(async move { let conn = endpoint.accept().await.expect("accept"); - let request = quic::web::Request::accept(&handle, conn).await.expect("handshake"); + let request = quic::web::Request::accept(conn).await.expect("handshake"); // The WebTransport equivalent of ALPN: pick the moq version. let protocol = request.protocols().iter().find(|p| *p == PROTO).cloned(); let mut response = quic::web::Response::default(); @@ -368,7 +368,7 @@ fn an_abandoned_handshake_closes_the_connection() { .block_on(async move { let conn = endpoint.accept().await.expect("accept"); let mut watch = conn.clone(); - let request = quic::web::Request::accept(&handle, conn).await.expect("handshake"); + let request = quic::web::Request::accept(conn).await.expect("handshake"); let err = request .respond(quic::web::Response::default().with_protocol("never-offered")) .await @@ -407,7 +407,7 @@ fn a_rejection_reaches_the_peer() { worker .block_on(async move { let conn = endpoint.accept().await.expect("accept"); - let request = quic::web::Request::accept(&handle, conn).await.expect("handshake"); + let request = quic::web::Request::accept(conn).await.expect("handshake"); within( &handle, "the rejection to be delivered", @@ -478,7 +478,7 @@ fn a_dropped_stream_carries_a_webtransport_code() { worker .block_on(async move { let conn = endpoint.accept().await.expect("accept"); - let request = quic::web::Request::accept(&handle, conn).await.expect("handshake"); + let request = quic::web::Request::accept(conn).await.expect("handshake"); let mut session = request .respond(quic::web::Response::default().with_protocol(PROTO)) .await diff --git a/rs/moq-uring/tests/workers.rs b/rs/moq-uring/tests/workers.rs index 87fe1daa3c..7cd4ae95ce 100644 --- a/rs/moq-uring/tests/workers.rs +++ b/rs/moq-uring/tests/workers.rs @@ -1,10 +1,11 @@ //! A steered thread-per-core group, end to end: two workers on two threads, -//! each with its own socket in one `SO_REUSEPORT` group and an endpoint -//! issuing steering-prefixed connection ids, serving clients that dial the -//! shared port. Whichever worker the Initial hashes to owns the connection, -//! and the prefix keeps every later packet (handshake continuation included) -//! on that worker; a wrong prefix stalls the handshake, so every dial -//! completing is what proves the steering. +//! each adopting its own member of one `SO_REUSEPORT` group and serving an +//! endpoint on it, for clients that dial the shared port. The member carries +//! the slot, so the endpoint issues steering-prefixed connection ids without +//! being told which. Whichever worker the Initial hashes to owns the +//! connection, and the prefix keeps every later packet (handshake +//! continuation included) on that worker; a wrong prefix stalls the +//! handshake, so every dial completing is what proves the steering. //! //! Kernel-gated: skips loudly below the Linux 6.12 floor (GitHub-hosted CI), //! and runs everywhere else. @@ -87,7 +88,7 @@ fn a_steered_group_serves_a_shared_port() { let mut group = group.complete(claims).expect("complete group"); let mut members = Vec::new(); while let Some(member) = group.member().expect("clone retained socket") { - members.push((member.shard(), member.into_inner())); + members.push(member); } let addr = group.addr(); @@ -99,24 +100,23 @@ fn a_steered_group_serves_a_shared_port() { let threads: Vec<_> = members .into_iter() .zip(&stops) - .map(|((shard, socket), stop)| { + .map(|(member, stop)| { let accepted = accepted.clone(); let stop = stop.clone(); let cert = certs.cert.clone(); let key = certs.key.clone(); std::thread::spawn(move || { + let shard = member.shard(); let mut worker = Worker::new(Config::default()).expect("worker"); let handle = worker.handle(); - let socket = handle.udp(socket, udp::Config::default()).expect("socket"); + // The member is adopted whole: its slot is what the endpoint + // steers with, and there is no other way to hand one over. + let socket = handle.udp(member, udp::Config::default()).expect("socket"); let mut server = quic::server::Config::new(quic::Identity::open(cert, key).expect("identity")); server.alpn = vec![ALPN.to_string()]; - let endpoint = quic::Endpoint::new( - &handle, - socket, - quic::endpoint::Config::default().with_server(server).with_shard(shard), - ) - .expect("endpoint"); + let endpoint = quic::Endpoint::new(socket, quic::endpoint::Config::default().with_server(server)) + .expect("endpoint"); handle.spawn(async move { // Accepted connections are dropped once counted; the @@ -144,7 +144,7 @@ fn a_steered_group_serves_a_shared_port() { let socket = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("client socket"); - let mut conn = quic::client::connect(&handle, socket, &dial).await.expect("connect"); + let mut conn = quic::client::connect(socket, &dial).await.expect("connect"); assert_eq!( web_transport_trait::poll::Session::protocol(&conn), Some(ALPN), From befe376d7ddf4e441587b78944852e51ea02d8d6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 21 Sep 2026 15:49:43 -0700 Subject: [PATCH 3/4] test(uring): drive the identity client until the server closes The client's handshake completes before the server's, so a client that stopped after connecting never sent its last flight and the owner's accept waited on a handshake that had timed out. Co-Authored-By: Claude Opus 5 --- rs/moq-uring/tests/identity.rs | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/rs/moq-uring/tests/identity.rs b/rs/moq-uring/tests/identity.rs index 5a08c4966a..7df4aedb87 100644 --- a/rs/moq-uring/tests/identity.rs +++ b/rs/moq-uring/tests/identity.rs @@ -114,15 +114,17 @@ fn an_endpoint_runs_on_the_worker_that_adopted_its_socket() { // A client on its own thread and worker, dialing the endpoint. Its // Initial reaches the socket immediately; only the owner can read it. + // The client is driven until the server closes on it: its side of the + // handshake completes before the server's, so it cannot stop earlier. let client = std::thread::spawn(move || { let mut worker = Worker::new(Config::default()).expect("client worker"); let sock = socket(&worker.handle()); worker .block_on(async move { let mut conn = quic::client::connect(sock, &dial_config(addr)).await.expect("dial"); - web_transport_trait::poll::Session::close(&mut conn, 0, "done"); + std::future::poll_fn(|cx| web_transport_trait::poll::Session::poll_closed(&mut conn, cx)).await }) - .expect("client loop"); + .expect("client loop") }); // Driving the bystander polls the accept from its loop, but the demux @@ -144,14 +146,21 @@ fn an_endpoint_runs_on_the_worker_that_adopted_its_socket() { .expect("bystander loop"); assert!(stalled, "the bystander worker served an endpoint it does not own"); - let accepted = owner - .block_on(async { endpoint.accept().await.expect("accept on the owner") }) + owner + .block_on(async { + let mut accepted = endpoint.accept().await.expect("accept on the owner"); + assert_eq!( + web_transport_trait::poll::Session::protocol(&accepted), + Some(ALPN), + "negotiated ALPN" + ); + web_transport_trait::poll::Session::close(&mut accepted, 0, "done"); + std::future::poll_fn(|cx| web_transport_trait::poll::Session::poll_closed(&mut accepted, cx)).await; + }) .expect("owner loop"); - assert_eq!( - web_transport_trait::poll::Session::protocol(&accepted), - Some(ALPN), - "negotiated ALPN" - ); - drop(accepted); - client.join().expect("client thread"); + + match client.join().expect("client thread") { + quic::Error::App { code: 0, .. } => {} + other => panic!("the client saw {other:?} instead of the server's close"), + } } From c1de2ce919d4881cde8ca6f91ba93326a23b2d77 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 21 Sep 2026 16:11:34 -0700 Subject: [PATCH 4/4] docs(uring): describe socket identity in the README Co-Authored-By: Claude Opus 5 --- rs/moq-uring/README.md | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/rs/moq-uring/README.md b/rs/moq-uring/README.md index 5987df7206..fe6737f7db 100644 --- a/rs/moq-uring/README.md +++ b/rs/moq-uring/README.md @@ -48,10 +48,14 @@ the UDP sockets bound through it. thread that spawned the worker, or read the worker's own with `Handle::metrics`. `moq-relay` publishes them at `/metrics` on its internal listener. -- **Steering**: an endpoint whose socket sits in a `moq-sock` steered - `SO_REUSEPORT` group sets `endpoint::Config::shard`, and every issued - connection id leads with the group's steering byte, so the kernel keeps a - connection (and a cluster dial's responses) on the worker that owns it. +- **Identity**: the socket names its worker. `Handle::udp` adopts a lone + `UdpSocket` or a member of a completed `moq-sock` steered `SO_REUSEPORT` + group (`udp::Bound`), and a `quic::Endpoint` built on it runs its demux and + every connection driver on that worker, whichever handle built it. A member + brings its slot along, so every issued connection id leads with the group's + steering byte and the kernel keeps a connection (and a cluster dial's + responses) on the worker that owns it. An endpoint on a dropped worker is + refused. Requires **Linux 6.12**; `Worker::new` refuses older kernels with a legible error rather than degrading (note that default container seccomp policies