diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 6712767a9c..a54936068f 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -24,7 +24,7 @@ above ([hang](/lib/rs/hang)); relays and CDNs implement only this. - **Routes** record the relay hops and a cost, which is what the relay [cluster](/bin/relay/cluster) routes on. A hop of 0 marks the chain anonymous: `Route::is_anonymous()` is true, and that route ranks below every fully identified one. - **Stats** counters per broadcast and session, drained by [`moq-stats`](https://docs.rs/moq-stats). -It runs over anything implementing `web_transport_trait::Session`: quinn, +It runs over anything implementing `web_transport_trait::poll::Session`: quinn, quiche, noq, the browser, iroh, or qmux over TCP, Unix sockets, and WebSockets. [`moq-tokio`](https://docs.rs/moq-tokio) wires those up. @@ -37,6 +37,29 @@ See the [Rust quick start](/lib/rs/#quick-start) and [`@moq/net`](/lib/js/net). The [path pattern](/concept/moq-lite#path-patterns) grammar lives on the concept page. +## Driving sessions + +`Client::connect(timers, transport)`, `Server::accept(timers, transport)`, and +`server::Handshake::ok()` return `(Session, Driver)`. The lite-only entry points +return the same pair. `moq-net` never spawns tasks: poll or spawn the driver to +run the protocol, process close requests, and update session statistics. + +```rust +let (session, driver) = client + .connect(moq_tokio::runtime::Runtime::new(), transport) + .await?; +tokio::spawn(driver); +``` + +Dropping the last session handle requests closure when the driver next runs. +Dropping the driver cancels the session. Keep both alive while using the +connection. `moq-tokio` and `moq-wasm` spawn the drivers for their callers. + +`Timers` supplies a clock and re-armable timers independently of the transport +and executor. `runtime::Test` (feature `test-runtime`) provides virtual time; +tests advance its clock and poll their drivers explicitly. The lite-only path +also supports native `!Send` transports when driven on their owning thread. + ## Patterns `Pattern` describes a set of paths; `Patterns` is a union reduced by diff --git a/rs/moq-e2ee/tests/support/harness.rs b/rs/moq-e2ee/tests/support/harness.rs index 11b303bc7b..85ead4fd0c 100644 --- a/rs/moq-e2ee/tests/support/harness.rs +++ b/rs/moq-e2ee/tests/support/harness.rs @@ -2,44 +2,28 @@ //! [`moq_net::Session`]s over the in-memory mock transport. //! //! The harness runs the full MoQ handshake (Client::connect + Server::accept) -//! over a [`MockSession`] pair and spawns both protocol drivers, giving tests +//! over a [`super::mock::MockSession`] pair and spawns both protocol drivers, giving tests //! two live sessions ready for pub/sub without any real QUIC or network I/O. #![allow(dead_code)] -use std::{marker::PhantomData, pin::Pin, task::Poll}; +use std::{pin::Pin, task::Poll}; use moq_net::{Client, Server, Session, Version, origin}; -use super::mock::{MockSession, create_mock_session_pair}; +use super::mock::create_mock_session_pair; -/// A tokio-backed [`moq_net::runtime::Runtime`] for these tests: machines are -/// spawned onto tokio and timers are tokio sleeps, so `tokio::time::pause` keeps -/// working. A copy of the crate's own `runtime::tokio_test` module, which is -/// `cfg(test)` and therefore invisible to integration tests. -pub struct TokioRuntime(PhantomData); +/// Tokio's clock and timers for these tests. +#[derive(Clone, Default)] +pub struct TokioRuntime; -impl TokioRuntime { +impl TokioRuntime { pub fn new() -> Self { - Self(PhantomData) + Self } } -impl Clone for TokioRuntime { - fn clone(&self) -> Self { - Self(PhantomData) - } -} - -impl Default for TokioRuntime { - fn default() -> Self { - Self::new() - } -} - -// Unbounded: timers don't involve the transport, so origin drivers can borrow -// a transportless handle (`TokioRuntime::<()>::new()`). -impl moq_net::runtime::Timers for TokioRuntime { +impl moq_net::runtime::Timers for TokioRuntime { type Timer = TokioTimer; fn timer(&self) -> Self::Timer { @@ -51,15 +35,6 @@ impl moq_net::runtime::Timers for TokioRuntime { } } -// Boxable: tokio work-steals, so the machine must be `Send`. -impl moq_net::runtime::Runtime for TokioRuntime { - type Transport = S; - - fn spawn(&self, machine: moq_net::runtime::Machine) { - tokio::spawn(machine); - } -} - /// The [`moq_net::runtime::Timer`] handed out by [`TokioRuntime`]. pub struct TokioTimer { at: Option, @@ -151,21 +126,25 @@ pub async fn connect_mock(opts: MockConnectOptions) -> MockPair { server = server.with_subscriber(subscribe); } - // Run both handshakes concurrently; the runtime spawns each side's machine + // Run both handshakes concurrently and spawn each side's driver // the moment its handshake resolves: on draft-17+ the server's accept blocks // on the client's SETUP, which only reaches the wire once the client's // machine is polled (and vice versa for the server's own SETUP). let client_fut = async { - client + let (session, driver) = client .connect(TokioRuntime::new(), client_transport) .await - .expect("client handshake failed") + .expect("client handshake failed"); + tokio::spawn(driver); + session }; let server_fut = async { - server + let (session, driver) = server .accept(TokioRuntime::new(), server_transport) .await - .expect("server handshake failed") + .expect("server handshake failed"); + tokio::spawn(driver); + session }; let (client_session, server_session) = tokio::join!(client_fut, server_fut); diff --git a/rs/moq-e2ee/tests/transport.rs b/rs/moq-e2ee/tests/transport.rs index 9051e6be25..6b9d4a6fc7 100644 --- a/rs/moq-e2ee/tests/transport.rs +++ b/rs/moq-e2ee/tests/transport.rs @@ -12,7 +12,7 @@ const TEST_TIMEOUT: Duration = Duration::from_secs(10); fn produce_origin(hop: u64) -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); - tokio::spawn(driver.run(TokioRuntime::<()>::new())); + tokio::spawn(driver.run(TokioRuntime::new())); producer } diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index a07fc03c0d..d4a59138dc 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -206,7 +206,7 @@ impl MoqOriginProducer { pub(crate) fn spawn(config: moq_net::origin::Config) -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(config); #[cfg(not(target_arch = "wasm32"))] - crate::ffi::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + crate::ffi::spawn(driver.run(moq_tokio::runtime::Runtime::new())); #[cfg(target_arch = "wasm32")] crate::ffi::spawn(driver.run(crate::runtime::Runtime)); producer diff --git a/rs/moq-ffi/src/runtime.rs b/rs/moq-ffi/src/runtime.rs index ee8964181e..58196d7379 100644 --- a/rs/moq-ffi/src/runtime.rs +++ b/rs/moq-ffi/src/runtime.rs @@ -1,9 +1,8 @@ -//! The browser runtime: microtask spawning and `wasmtimer`-backed timers. +//! Browser-backed timers for MoQ drivers. use std::{pin::Pin, task::Poll}; -/// The [`moq_net::Runtime`] for the browser: machines run on the microtask -/// queue via `web_async::spawn`, timers are `setTimeout`-backed sleeps. +/// Supplies the browser clock and timers to MoQ drivers. #[derive(Clone, Copy, Debug, Default)] pub(crate) struct Runtime; @@ -15,18 +14,6 @@ impl moq_net::Timers for Runtime { } } -impl moq_net::Runtime for Runtime { - type Transport = web_transport_wasm::Session; - - fn spawn(&self, machine: moq_net::runtime::Machine) { - web_async::spawn(async move { - // The session surfaces the result through `closed()`; nothing to do - // with it here. - let _ = machine.await; - }); - } -} - /// A `web_async::time::Sleep` driven through the [`moq_net::runtime::Timer`] /// contract. pub(crate) struct Timer { diff --git a/rs/moq-ffi/src/session.rs b/rs/moq-ffi/src/session.rs index 750b3871c2..839446f722 100644 --- a/rs/moq-ffi/src/session.rs +++ b/rs/moq-ffi/src/session.rs @@ -291,15 +291,19 @@ impl Client { } .map_err(|err| MoqError::Connect(format!("{err}")))?; - // The runtime spawns the protocol machine on the microtask queue. The machine + // Run the driver on the microtask queue. The driver // holds no session clone, so dropping the last handle still closes the // transport and ends that task. - let session = moq_net::Client::new() + let (session, driver) = moq_net::Client::new() .with_publisher(&publish) .with_subscriber(subscribe.clone()) .connect(crate::runtime::Runtime, transport) .await?; + crate::ffi::spawn(async move { + let _ = driver.await; + }); + Ok(Arc::new(MoqSession::accepted(session, publish, subscribe))) } } diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index a72dba6744..629c1b75fa 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -360,7 +360,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-hls/src/export/upstream.rs b/rs/moq-hls/src/export/upstream.rs index 63ca786921..767b1cb51d 100644 --- a/rs/moq-hls/src/export/upstream.rs +++ b/rs/moq-hls/src/export/upstream.rs @@ -40,7 +40,7 @@ mod tests { #[tokio::test] async fn self_references_keep_the_catalog_broadcast_after_replacement() { let (origin, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); - let driver = tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + let driver = tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); let old = origin.create_broadcast("a/live").unwrap(); let source = moq_mux::Source::new(origin.consume(), "a/live"); let upstream = Upstream { diff --git a/rs/moq-hls/src/server/mod.rs b/rs/moq-hls/src/server/mod.rs index 3065102ee0..1879cbacf2 100644 --- a/rs/moq-hls/src/server/mod.rs +++ b/rs/moq-hls/src/server/mod.rs @@ -146,7 +146,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-mux/src/source.rs b/rs/moq-mux/src/source.rs index 6eb61e5d10..f475bf2e60 100644 --- a/rs/moq-mux/src/source.rs +++ b/rs/moq-mux/src/source.rs @@ -288,7 +288,7 @@ impl BroadcastConfig for hang::catalog::BinaryConfig { pub(crate) fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index d8e48f5adf..1e7dcc1eea 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -6,9 +6,7 @@ use crate::{ ietf, lite, setup, stats, }; -// The transport methods are called on the projected `R::Transport`, which needs the -// trait itself in scope (a plain type parameter would not). -use web_transport_trait::{MaybeSend, MaybeSync, poll::Session as _}; +use web_transport_trait::{MaybeSend, MaybeSync}; /// A MoQ client session builder. #[derive(Default, Clone)] @@ -153,10 +151,16 @@ impl Client { } /// Start a lite session on an already-negotiated version: build our SETUP, - /// wire the origins, and hand the machine to the runtime. - fn start_lite(&self, runtime: R, session: R::Transport, version: lite::Version) -> Result + /// wire the origins, and return the session and its driver. + fn start_lite( + &self, + runtime: R, + session: S, + version: lite::Version, + ) -> Result<(Session, crate::Driver), Error> where - R: crate::runtime::Runtime + 'static, + S: crate::transport::poll::Session, + R: crate::runtime::Timers + 'static, { let (publish, subscribe) = self.origins(); @@ -190,12 +194,12 @@ impl Client { peer_setup: None, })?; - Ok(Session::spawn( + Ok(Session::new( runtime, session, version.into(), start.recv_bandwidth, - crate::runtime::Protocol::Lite(Box::new(start.driver)), + crate::driver::Protocol::Lite(Box::new(start.driver)), start.goaway, )) } @@ -209,9 +213,10 @@ impl Client { /// [`Boxable`](crate::transport::poll::Boxable) transport. An ietf ALPN, an /// unknown one, or the legacy no-ALPN SETUP negotiation is refused with /// [`Error::Version`]. - pub async fn connect_lite(&self, runtime: R, session: R::Transport) -> Result + pub async fn connect_lite(&self, runtime: R, session: S) -> Result<(Session, crate::Driver), Error> where - R: crate::runtime::Runtime + 'static, + S: crate::transport::poll::Session, + R: crate::runtime::Timers + 'static, { let version = match session.protocol() { Some(ALPN_LITE_06_WIP) => lite::Version::Lite06Wip, @@ -224,15 +229,14 @@ impl Client { self.start_lite(runtime, session, version) } - /// Perform the MoQ handshake, returning the [`Session`]. + /// Perform the MoQ handshake, returning the [`Session`] and its [`Driver`](crate::Driver). /// - /// The session's protocol machine is handed to `runtime` - /// ([`Runtime::spawn`](crate::runtime::Runtime::spawn)), so there is nothing - /// else to drive: the runtime runs the session for as long as it lives. - pub async fn connect(&self, runtime: R, mut session: R::Transport) -> Result + /// Poll or spawn the returned driver to run the session. `runtime` supplies + /// only its clock and timers. + pub async fn connect(&self, runtime: R, mut session: S) -> Result<(Session, crate::Driver), Error> where - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, - R::Transport: crate::transport::poll::Boxable, + S: crate::transport::poll::Boxable, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { let (publish, subscribe) = self.origins(); @@ -270,12 +274,12 @@ impl Client { })?; tracing::debug!(version = ?v, "connected"); - return Ok(Session::spawn( + return Ok(Session::new( runtime, session, v, None, - crate::runtime::Protocol::Ietf(protocol), + crate::driver::Protocol::Ietf(protocol), goaway, )); } @@ -376,7 +380,7 @@ impl Client { ( start.recv_bandwidth, - crate::runtime::Protocol::Lite(Box::new(start.driver)), + crate::driver::Protocol::Lite(Box::new(start.driver)), start.goaway, ) } @@ -409,11 +413,11 @@ impl Client { peer_setup_stream: None, peer_declared: Some(peer_declared), })?; - (None, crate::runtime::Protocol::Ietf(protocol), goaway) + (None, crate::driver::Protocol::Ietf(protocol), goaway) } }; - Ok(Session::spawn(runtime, session, version, recv_bw, protocol, goaway)) + Ok(Session::new(runtime, session, version, recv_bw, protocol, goaway)) } } @@ -675,12 +679,12 @@ mod tests { .into(), ); - // `connect` returns as soon as the handshake completes; the protocol machine - // was handed to the runtime, which spawned it onto tokio. - let _session = client + // Start the returned driver after the handshake completes. + let (_session, driver) = client .connect(crate::runtime::tokio_test::Tokio::new(), fake.clone()) .await .unwrap(); + tokio::spawn(driver); // Verify the client setup was encoded using Draft14 framing (ALPN_LITE fallback path). let mut setup_bytes = Bytes::from(fake.control_writes()); @@ -726,7 +730,7 @@ mod tests { // Paused time auto-advances while every task is idle, so a `connect` that waits // on the silent peer trips this rather than hanging the suite. - tokio::time::timeout( + let (_session, _driver) = tokio::time::timeout( std::time::Duration::from_secs(30), client.connect(crate::runtime::tokio_test::Tokio::new(), transport), ) @@ -745,31 +749,25 @@ mod tests { run_alpn_lite_fallback_case(None).await; } - // This fake reports no send-rate estimate, so it never reaches a timer in the - // bandwidth loop. `connect` hands the machine to the runtime we pass, so with - // the deterministic Test runtime nothing runs on an ambient executor at all. - // - // The machine must hold no Session clone (the #2286 leak), so the transport - // still closes when the caller drops their last session handle. The close is - // relayed through the machine (the Session holds no transport), so it lands - // on the next tick. + // No executor is running: only explicitly polling the driver may process + // a session close, and the driver must not retain a session handle. #[test] - fn machine_is_runtime_polled_and_holds_no_session() { + fn driver_is_caller_polled_and_holds_no_session() { let fake = FakeSession::new(Some(ALPN_LITE_04), Vec::new()); let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - let runtime = crate::runtime::Test::::new(); - let session = futures::executor::block_on(client.connect(runtime.clone(), fake.clone())).unwrap(); + let runtime = crate::runtime::Test::new(); + let (session, mut driver) = futures::executor::block_on(client.connect(runtime.clone(), fake.clone())).unwrap(); assert_eq!(session.version(), Version::Lite(lite::Version::Lite04)); - // The machine sits inside the Test runtime, stepped manually: nothing was - // spawned onto an ambient runtime, and it is still pending. - assert_eq!(runtime.tick(), 1); + // Construction leaves the driver idle until the caller polls it. + assert!(driver.poll(&kio::Waiter::noop()).is_pending()); // The caller drops their only session clone; the machine observes the // last handle going away and closes the transport. drop(session); - runtime.tick(); + assert!(fake.state.close_events.lock().unwrap().is_empty()); + let _ = driver.poll(&kio::Waiter::noop()); assert_eq!( fake.state.close_events.lock().unwrap()[0].0, SessionError::Cancel.to_code() @@ -784,17 +782,18 @@ mod tests { let fake = FakeSession::new(Some(ALPN_LITE_04), Vec::new()); let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - let runtime = crate::runtime::Test::::new(); - let session = futures::executor::block_on(client.connect(runtime.clone(), fake.clone())).unwrap(); + let runtime = crate::runtime::Test::new(); + let (session, mut driver) = futures::executor::block_on(client.connect(runtime.clone(), fake.clone())).unwrap(); let clone = session.clone(); // One clone dropping does nothing while another is alive. drop(session); - runtime.tick(); + assert!(fake.state.close_events.lock().unwrap().is_empty()); + let _ = driver.poll(&kio::Waiter::noop()); assert!(fake.state.close_events.lock().unwrap().is_empty()); clone.abort(Error::Cancel); - runtime.tick(); + let _ = driver.poll(&kio::Waiter::noop()); assert_eq!( fake.state.close_events.lock().unwrap()[0].0, SessionError::Cancel.to_code() @@ -802,28 +801,28 @@ mod tests { // And the machine publishes the transport's terminal error, which is // what `closed()` reports. - runtime.tick(); + let _ = driver.poll(&kio::Waiter::noop()); futures::executor::block_on(clone.closed()); // The final drop requests no second close: the handle-side close is once. let closes = fake.state.close_events.lock().unwrap().len(); drop(clone); - runtime.tick(); + let _ = driver.poll(&kio::Waiter::noop()); assert_eq!(fake.state.close_events.lock().unwrap().len(), closes); } - // A runtime that drops the machine instead of running it tears the session + // Dropping the driver instead of running it tears the session // down: the machine was the only transport holder, and `closed()` resolves // rather than parking forever on a machine nobody polls. #[test] - fn dropped_machine_resolves_closed() { + fn dropped_driver_resolves_closed() { let fake = FakeSession::new(Some(ALPN_LITE_04), Vec::new()); let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - let runtime = crate::runtime::Test::::new(); - let session = futures::executor::block_on(client.connect(runtime.clone(), fake.clone())).unwrap(); + let runtime = crate::runtime::Test::new(); + let (session, driver) = futures::executor::block_on(client.connect(runtime.clone(), fake.clone())).unwrap(); - runtime.shutdown(); + drop(driver); assert!(matches!(futures::executor::block_on(session.closed()), Error::Cancel)); } @@ -971,7 +970,7 @@ mod tests { } // The point of the lite-only entry: a `!Send` transport yields a `!Send` - // machine held by a local runtime, while the severed Session handle stays + // driver polled by its caller, while the severed Session handle stays // Send + Sync. Compiling is most of the assertion; the rest checks the // machine still relays the close. #[test] @@ -983,15 +982,15 @@ mod tests { }; let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - let runtime = crate::runtime::Test::::new(); - let session = futures::executor::block_on(client.connect_lite(runtime.clone(), local)).unwrap(); - assert_eq!(runtime.tick(), 1); + let runtime = crate::runtime::Test::new(); + let (session, mut driver) = futures::executor::block_on(client.connect_lite(runtime.clone(), local)).unwrap(); + assert!(driver.poll(&kio::Waiter::noop()).is_pending()); fn assert_send_sync(_: &T) {} assert_send_sync(&session); session.abort(Error::Cancel); - runtime.tick(); + let _ = driver.poll(&kio::Waiter::noop()); assert_eq!( fake.state.close_events.lock().unwrap()[0].0, SessionError::Cancel.to_code() @@ -999,7 +998,7 @@ mod tests { } // The server-side twin: a `!Send` transport accepts a lite session whose - // machine a local runtime drives. + // driver the caller polls directly. #[test] fn accept_lite_over_a_send_less_transport() { let fake = FakeSession::new(Some(ALPN_LITE_04), Vec::new()); @@ -1009,13 +1008,14 @@ mod tests { }; let server = crate::Server::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - let runtime = crate::runtime::Test::::new(); - let session = futures::executor::block_on(server.accept_lite(runtime.clone(), local)).unwrap(); + let runtime = crate::runtime::Test::new(); + let (session, mut driver) = futures::executor::block_on(server.accept_lite(runtime.clone(), local)).unwrap(); assert_eq!(session.version(), Version::Lite(lite::Version::Lite04)); - assert_eq!(runtime.tick(), 1); + assert!(driver.poll(&kio::Waiter::noop()).is_pending()); drop(session); - runtime.tick(); + assert!(fake.state.close_events.lock().unwrap().is_empty()); + let _ = driver.poll(&kio::Waiter::noop()); assert_eq!( fake.state.close_events.lock().unwrap()[0].0, SessionError::Cancel.to_code() @@ -1032,7 +1032,7 @@ mod tests { _local: std::rc::Rc::new(()), }; let client = Client::new(); - let runtime = crate::runtime::Test::::new(); + let runtime = crate::runtime::Test::new(); let result = futures::executor::block_on(client.connect_lite(runtime, local)); assert!(matches!(result, Err(Error::Version))); } @@ -1046,10 +1046,11 @@ mod tests { fake.set_send_rate(Some(1_000_000)); let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - let session = client + let (session, driver) = client .connect(crate::runtime::tokio_test::Tokio::new(), fake.clone()) .await .unwrap(); + tokio::spawn(driver); // The construction-time snapshot, before the machine sampled anything. assert_eq!( @@ -1078,10 +1079,11 @@ mod tests { fake.set_bytes_sent(Some(0)); let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - let session = client + let (session, driver) = client .connect(crate::runtime::tokio_test::Tokio::new(), fake.clone()) .await .unwrap(); + tokio::spawn(driver); assert!( session.send_bandwidth().is_none(), "no send-rate estimate, so nothing samples on its own" @@ -1108,10 +1110,11 @@ mod tests { fake.set_send_rate(Some(1_000_000)); let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - let session = client + let (session, driver) = client .connect(crate::runtime::tokio_test::Tokio::new(), fake.clone()) .await .unwrap(); + tokio::spawn(driver); let mut bandwidth = session.send_bandwidth().expect("backend reports an estimate"); assert_eq!( diff --git a/rs/moq-net/src/driver.rs b/rs/moq-net/src/driver.rs new file mode 100644 index 0000000000..ae4a4d248a --- /dev/null +++ b/rs/moq-net/src/driver.rs @@ -0,0 +1,134 @@ +//! Drive a session on the caller's executor. + +use std::task::Poll; + +use crate::{Error, runtime::Timers}; + +/// The future driving a [`crate::Session`]'s protocol state. +/// +/// Returned by [`crate::Client::connect`] and [`crate::Server::accept`]. Poll it, +/// either as a [`Future`] or by stepping [`poll`](Self::poll) from a +/// [`kio`]-style poll function. It resolves when the session ends, and keeps +/// returning that same result if polled again. +/// +/// It holds no session handle: dropping the last [`crate::Session`] requests +/// closure on the next poll. Dropping the driver cancels the session, and +/// [`crate::Session::closed`] resolves with [`Error::Cancel`]. Its `Send`-ness +/// follows its transport and timer provider. +#[must_use = "the session makes no progress unless its driver is polled"] +pub struct Driver { + state: State, + // Retains the waiter across `Future` polls so its kio registrations stay + // live. Kept out of `State` so the borrow `hold` hands back doesn't + // collide with the `&mut` that polling the state needs. + park: kio::Park, +} + +/// The protocol half of a machine, one variant per negotiated wire protocol. +/// +/// The lite driver is a named machine, so the machine's `Send`-ness follows +/// the transport (a pinned `!Send` transport yields a `!Send` machine that +/// stays on its thread). The ietf driver is still a boxed future; the box +/// demands `Send` on native, which is why the ietf path requires a +/// [`Boxable`](crate::transport::poll::Boxable) transport until it too becomes +/// a named machine. +pub(crate) enum Protocol { + /// Boxed for size only: a concrete box, so `Send` stays inferred. + Lite(Box>), + Ietf(crate::util::MaybeSendBox<'static, Result<(), Error>>), +} + +/// Everything the machine polls, split from the park so the two borrow +/// disjointly. +pub(crate) struct State { + pub(crate) protocol: Protocol, + // The session supervisor, polled alongside the protocol: it executes the + // handles' close requests, publishes the transport's terminal error, and + // samples stats. It finishes once the transport reports closed, and the + // machine is not done until it has: the protocol's terminal transport close + // is what `Session::closed` observes, so resolving before it is published + // would leave waiters parked on a machine nobody polls again. `None` once + // finished, since a completed machine must not be polled again. + pub(crate) supervisor: Option>, + // Cached so a poll after completion doesn't re-poll a finished protocol. + pub(crate) result: Option>, +} + +impl Driver { + pub(crate) fn new(state: State) -> Self { + Self { + state, + park: kio::Park::default(), + } + } + + /// Drive the protocol one step, registering `waiter` for the next wakeup. + /// + /// The `poll_*` counterpart of `.await`ing the machine, for runtimes + /// composing it into their own [`kio`]-style poll loops. + pub fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { + self.state.poll(waiter) + } +} + +impl Protocol { + fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { + match self { + Self::Lite(driver) => driver.poll(waiter), + Self::Ietf(driver) => waiter.poll_future(driver.as_mut()), + } + } +} + +impl State { + fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { + if let Some(supervisor) = &mut self.supervisor + && supervisor.poll(waiter).is_ready() + { + self.supervisor = None; + } + + if self.result.is_none() + && let Poll::Ready(result) = self.protocol.poll(waiter) + { + self.result = Some(result); + // The protocol's last act was closing the transport, which wakes the + // supervisor's close watch; poll it now instead of waiting a turn. + if let Some(supervisor) = &mut self.supervisor + && supervisor.poll(waiter).is_ready() + { + self.supervisor = None; + } + } + + match (&self.result, &self.supervisor) { + (Some(result), None) => Poll::Ready(result.clone()), + _ => Poll::Pending, + } + } +} + +// Every part of the machine is a poll machine driven through `&mut` (the one +// pinned piece, the boxed ietf future, is `Unpin` itself), so nothing relies +// on address stability and the machine may move freely between polls. +impl Unpin for Driver {} + +impl Future for Driver { + type Output = Result<(), Error>; + + fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll { + let this = &mut *self; + // Disjoint field borrows: `hold` borrows the park for as long as the + // waiter lives, while the state is polled through its own `&mut`. + let waiter = this.park.hold(cx); + this.state.poll(waiter) + } +} + +impl std::fmt::Debug for Driver { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Driver") + .field("done", &self.state.result.is_some()) + .finish() + } +} diff --git a/rs/moq-net/src/goaway.rs b/rs/moq-net/src/goaway.rs index cd1773042c..9ab2f651c0 100644 --- a/rs/moq-net/src/goaway.rs +++ b/rs/moq-net/src/goaway.rs @@ -270,14 +270,14 @@ impl Protocol { /// The draft makes the deadline the sender's promise, not the peer's, so the /// driver arms this after the message hits the wire rather than making the /// caller hold a handle and await it. -pub(crate) struct Enforce { +pub(crate) struct Enforce { session: S, /// The armed deadline, or `None` when the GOAWAY carried no timeout (ready /// immediately: there is nothing to enforce). deadline: Option<(crate::runtime::Deadline, Duration)>, } -impl Enforce { +impl Enforce { pub fn new(runtime: &R, session: S, timeout: Option) -> Self { Self { session, @@ -307,7 +307,7 @@ impl Enforce( +pub(crate) async fn enforce( runtime: &R, session: &mut S, timeout: Option, diff --git a/rs/moq-net/src/ietf/control.rs b/rs/moq-net/src/ietf/control.rs index ebca2556bd..30ab511c5e 100644 --- a/rs/moq-net/src/ietf/control.rs +++ b/rs/moq-net/src/ietf/control.rs @@ -32,7 +32,7 @@ impl Control { /// /// `runtime` arms the give-up timer, so the wait cannot outlive the caller's /// clock. - pub async fn next_request_id(&self, runtime: &R) -> Result { + pub async fn next_request_id(&self, runtime: &R) -> Result { let mut timeout = crate::runtime::Deadline::after(runtime, std::time::Duration::from_secs(10)); kio::wait(|waiter| { diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 96e59fa932..76444c071c 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -220,7 +220,7 @@ enum NamespaceEvent { } #[derive(Clone)] -pub(super) struct Publisher { +pub(super) struct Publisher { // Arms the advertise, retry, and linger timers. runtime: R, session: S, @@ -276,7 +276,7 @@ impl Drop for Join { impl Publisher where S: crate::transport::poll::Boxable, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { pub fn new( @@ -2526,7 +2526,7 @@ mod serve_tests { use crate::lite::test_transport::{Log, ScriptedSession, SinkSession}; use crate::model::ProduceTest; - type TestRuntime = crate::runtime::tokio_test::Tokio; + type TestRuntime = crate::runtime::tokio_test::Tokio; fn occurrences(log: &Log, needle: &[u8]) -> usize { let writes = log.writes.lock().unwrap(); @@ -3373,7 +3373,7 @@ mod tests { /// The tokio-backed test runtime. Its transport parameter is phantom, so one /// type serves every fake session in this module. - type TestRuntime = crate::runtime::tokio_test::Tokio; + type TestRuntime = crate::runtime::tokio_test::Tokio; async fn settle() { tokio::time::sleep(Duration::from_millis(1)).await; diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index eb784069cf..1b880dff54 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -15,7 +15,7 @@ use super::{ }; /// Everything one moq-transport session needs to start. -pub struct Config { +pub struct Config { /// The runtime that arms the session's timers. pub runtime: R, @@ -71,7 +71,7 @@ pub fn start( ) -> Result<(MaybeSendBox<'static, Result<(), Error>>, crate::goaway::Handle), Error> where S: crate::transport::poll::Boxable, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { let Config { @@ -514,7 +514,7 @@ fn peer_from_params(params: &ietf::Parameters, version: Version) -> Result( +async fn run_setup( runtime: R, mut session: S, version: Version, @@ -600,7 +600,7 @@ async fn run_unis( ) -> Result<(), Error> where S: crate::transport::poll::Boxable, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { let outer_version = crate::Version::Ietf(version); @@ -711,7 +711,7 @@ async fn run_uni_group( ) -> Result<(), Error> where S: crate::transport::poll::Boxable, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { let kind: u64 = stream.decode_peek().await?; @@ -742,7 +742,7 @@ async fn run_dispatch( ) -> Result<(), Error> where S: crate::transport::poll::Boxable, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { // PUBLISH_NAMESPACE decodes differently once the MoQ Cluster extension is @@ -886,7 +886,7 @@ mod tests { /// The tokio-backed test runtime. Its transport parameter is phantom, so one /// type serves every fake session in this module. - type TestRuntime = crate::runtime::tokio_test::Tokio; + type TestRuntime = crate::runtime::tokio_test::Tokio; fn occurrences(log: &crate::lite::test_transport::Log, needle: &[u8]) -> usize { let writes = log.writes.lock().unwrap(); diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index b8985e7c17..f310a91b51 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -382,7 +382,7 @@ struct Advertised { } #[derive(Clone)] -pub(super) struct Subscriber { +pub(super) struct Subscriber { // Arms the track-alias and request-id timeouts. runtime: R, session: S, @@ -421,7 +421,7 @@ pub(super) struct Subscriber( +async fn resolve_track_alias( runtime: &R, aliases: kio::Consumer, alias: u64, @@ -450,7 +450,7 @@ async fn resolve_track_alias( impl Subscriber where S: crate::transport::poll::Boxable, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { #[allow(clippy::too_many_arguments)] @@ -2131,7 +2131,7 @@ where impl Subscriber where S: crate::transport::poll::Boxable, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { /// The group producer this subgroup stream writes into, and the Object ID it starts at. @@ -2281,7 +2281,7 @@ impl GroupIngest { impl Subscriber where S: crate::transport::poll::Boxable, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { /// Read a fill fetch stream: the head of the group a subscription joins part way through. @@ -2765,7 +2765,7 @@ mod tests { /// The tokio-backed test runtime. Its transport parameter is phantom, so one /// type serves every fake session in this module. - type TestRuntime = crate::runtime::tokio_test::Tokio; + type TestRuntime = crate::runtime::tokio_test::Tokio; #[tokio::test(start_paused = true)] async fn track_alias_waits_for_control_message() { @@ -5354,7 +5354,7 @@ mod stitch_tests { util::{TaskSet, Tasks}, }; - type TestRuntime = crate::runtime::tokio_test::Tokio; + type TestRuntime = crate::runtime::tokio_test::Tokio; const VERSION: Version = Version::Draft20; const ALIAS: u64 = 7; @@ -5989,7 +5989,7 @@ mod joining_fetch_tests { util::{TaskSet, Tasks}, }; - type TestRuntime = crate::runtime::tokio_test::Tokio; + type TestRuntime = crate::runtime::tokio_test::Tokio; const JOINING_DRAFTS: [Version; 6] = [ Version::Draft14, diff --git a/rs/moq-net/src/lib.rs b/rs/moq-net/src/lib.rs index d8a232bc67..65afd8ffae 100644 --- a/rs/moq-net/src/lib.rs +++ b/rs/moq-net/src/lib.rs @@ -47,15 +47,15 @@ //! last producer signals consumers that no more updates are coming. //! //! ## Async -//! This library is async-first, but it never spawns onto a global executor and -//! never reaches for ambient state. [`Client::connect`] and [`Server::accept`] -//! take a [`Runtime`], which supplies the two things a session cannot do alone: -//! run its protocol [`runtime::Machine`] to completion and arm its timers. The -//! machine holds no session handle, so the transport still closes when the last -//! [`Session`] clone drops (or on [`Session::abort`]), which in turn finishes -//! the machine. Each transport ships its runtime (`moq-tokio`, `moq-wasm`, a -//! thread-per-core io_uring runtime), and `runtime::Test` is the -//! deterministic one for tests; see the [`runtime`] module. +//! This library never spawns tasks. [`Client::connect`] and [`Server::accept`] +//! take [`Timers`] and return a `(Session, Driver)` pair. Poll or spawn the +//! [`Driver`] to run the session. It holds no session handle, so dropping the +//! last [`Session`] clone requests transport closure and lets the driver finish. +//! Dropping the driver cancels the session instead. +//! +//! `moq-tokio` and `moq-wasm` supply timers and spawn drivers for their callers. +//! Direct callers can use any executor, including a thread-local executor for +//! `!Send` transports. `runtime::Test` (feature `test-runtime`) supplies virtual time for tests. //! //! Origins follow the caller-driven pattern: [`origin::Producer::new`] returns a //! `(Producer, Driver)` pair, [`origin::Driver::run`] takes the [`Timers`] the @@ -68,7 +68,7 @@ //! (plain [`std::task::Waker`] plumbing) and `futures`, so any executor can poll //! them, and the `poll_xxx` counterparts can be stepped synchronously with a //! [`kio::Waiter`]. Purely model-layer methods (tracks, groups, frames, -//! origins) never arm a timer and need no [`Runtime`] at all; they read the +//! origins) never arm a timer and need no [`Timers`] at all; they read the //! crate's ambient clock for passive stamps (arrival times, cache ticks). #![warn(missing_docs)] @@ -79,6 +79,7 @@ mod client; mod coding; +mod driver; mod error; pub mod goaway; // Not part of the public API: compiled only for the crate's own tests and for the @@ -103,6 +104,7 @@ pub mod transport; pub use client::*; pub use coding::{BoundsExceeded, DecodeError, EncodeError, VarInt}; +pub use driver::Driver; pub use error::*; /// The session direction a client advertises in its SETUP (moq-lite-05+). pub use lite::Role; @@ -110,7 +112,7 @@ pub use model::*; pub use path::{ AsPath, InvalidPattern, Path, PathOwned, PathPrefixes, PathRelative, PathRelativeOwned, Pattern, Patterns, }; -pub use runtime::{Runtime, Timers}; +pub use runtime::Timers; pub use server::Server; pub use session::Session; pub use version::*; diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 61a597efaa..c91eed1373 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -20,7 +20,7 @@ use crate::{ use super::Version; -pub(super) struct PublisherConfig { +pub(super) struct PublisherConfig { /// The runtime that arms the publisher's timers. pub runtime: R, pub session: S, @@ -128,7 +128,7 @@ impl Shared { /// The publisher half: accepts control streams and drives each as a child state /// machine. Resolves only on a transport error; children never end the session. -pub(super) struct Publisher { +pub(super) struct Publisher { shared: Arc>, // Cloned into each control-stream child that arms timers (PROBE, announce linger). runtime: R, @@ -138,7 +138,7 @@ pub(super) struct Publisher>, } -impl Publisher { +impl Publisher { pub fn new(config: PublisherConfig) -> Self { // Identity stamped onto outbound announce hops. Derived from the // origin we're consuming so it matches the local relay identity @@ -167,7 +167,7 @@ impl Publisher Publisher where S: crate::transport::poll::Session, - R: crate::runtime::Runtime, + R: crate::runtime::Timers, { pub fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { let _ = self.children.poll(waiter); @@ -194,7 +194,7 @@ where } #[cfg(test)] -impl Publisher { +impl Publisher { /// Test shim: drive one announce-interest stream like the old `run_announce`. async fn run_announce( stream: &mut Stream, @@ -210,7 +210,7 @@ impl Publisher { +struct Control { shared: Arc>, // Handed to the children that arm timers (PROBE, announce linger). runtime: R, @@ -220,7 +220,7 @@ struct Control { // A state machine's enum is its storage: one transient instance per stream, so the // big variant is the working state, not padding held in bulk. #[allow(clippy::large_enum_variant)] -enum ControlState { +enum ControlState { /// Reading the stream's type. Start { stream: Stream, @@ -237,7 +237,7 @@ enum ControlState kio::Task for Control { +impl kio::Task for Control { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { if let Err(err) = ready!(self.poll_serve(waiter)) { tracing::warn!(%err, "control stream error"); @@ -246,7 +246,7 @@ impl kio::Task f } } -impl Control { +impl Control { fn poll_serve(&mut self, waiter: &kio::Waiter) -> Poll> { loop { match &mut self.state { @@ -315,7 +315,7 @@ impl Control { +struct ProbeServe { shared: Arc>, runtime: R, stream: Option>, @@ -323,7 +323,7 @@ struct ProbeServe, } -impl ProbeServe { +impl ProbeServe { const PROBE_INTERVAL: Duration = Duration::from_millis(100); const PROBE_MAX_AGE: Duration = Duration::from_secs(10); const PROBE_MAX_DELTA: f64 = 0.25; @@ -1641,7 +1641,7 @@ mod announce_test { use std::sync::Mutex; /// The tokio-backed test runtime, matching the fake transport. - type TestRuntime = crate::runtime::tokio_test::Tokio; + type TestRuntime = crate::runtime::tokio_test::Tokio; type TestPublisher = Publisher; const VERSION: Version = Version::Lite06Wip; @@ -3014,7 +3014,7 @@ mod tests { use crate::model::ProduceTest; /// The tokio-backed test runtime, matching the fake transport. - type TestRuntime = crate::runtime::tokio_test::Tokio; + type TestRuntime = crate::runtime::tokio_test::Tokio; /// A peer that declares no origin in its SETUP is split-horizoned by the identity /// the caller assigned it, on the data plane and not just the announce filter. diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 260800ee52..d4b91fec39 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -11,7 +11,7 @@ use super::{ DataType, PeerSetup, Publisher, PublisherConfig, Setup, Subscriber, SubscriberConfig, SubscriberDriver, Version, }; -pub(crate) struct SessionStart { +pub(crate) struct SessionStart { pub recv_bandwidth: Option, /// The session's protocol machine, named so its `Send`-ness stays inferred /// from the transport instead of being fixed by a box. @@ -50,7 +50,7 @@ pub async fn accept_setup( } /// Everything one moq-lite session needs to start. -pub struct Config { +pub struct Config { /// The runtime that arms the session's timers. pub runtime: R, @@ -94,7 +94,7 @@ pub struct Config(config: Config) -> Result, Error> where S: crate::transport::poll::Session, - R: crate::runtime::Runtime, + R: crate::runtime::Timers, { let Config { runtime, @@ -208,7 +208,7 @@ where /// The lite session driver: one poll function racing every protocol arm, in /// place of a task set of boxed futures. -pub(crate) struct Driver { +pub(crate) struct Driver { /// Advertising our capabilities, or `None` once sent (or on a version with no /// Setup Stream). setup: Option>, @@ -227,7 +227,7 @@ pub(crate) struct Driver Driver where S: crate::transport::poll::Session, - R: crate::runtime::Runtime, + R: crate::runtime::Timers, { pub(crate) fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { let res = std::task::ready!(self.poll_protocol(waiter)); @@ -371,7 +371,7 @@ impl SendSetup { /// Runs on every version, including those with no GOAWAY message: the deadline is /// the sender's own timer, so a caller draining a lite-03 peer still gets the /// session closed on schedule; the peer just never learns why. -struct SendGoaway { +struct SendGoaway { version: Version, runtime: R, goaway: crate::goaway::Protocol, @@ -382,7 +382,7 @@ struct SendGoaway, } -enum SendGoawayState { +enum SendGoawayState { /// Parked on the send trigger, racing the transport close so a parked trigger /// never blocks the driver. The trigger fires at most once. Waiting, @@ -401,7 +401,7 @@ enum SendGoawayState), } -impl SendGoaway { +impl SendGoaway { fn new(runtime: R, session: S, goaway: crate::goaway::Protocol, version: Version) -> Self { Self { version, diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index a6e2ff551e..14f5ee37ec 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -4104,7 +4104,7 @@ impl ProduceTest for Config { fn produce(self) -> Producer { let (producer, driver) = Producer::new(self); if tokio::runtime::Handle::try_current().is_ok() { - web_async::spawn(driver.run(crate::runtime::tokio_test::Tokio::<()>::new())); + web_async::spawn(driver.run(crate::runtime::tokio_test::Tokio::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. @@ -5314,7 +5314,7 @@ mod tests { async fn driver_resolves_with_live_consumers() { let (producer, driver) = Producer::new(Config::new(origin(1))); let consumer = producer.consume(); - let run = driver.run(crate::runtime::tokio_test::Tokio::<()>::new()); + let run = driver.run(crate::runtime::tokio_test::Tokio::new()); drop(producer); tokio::time::timeout(Duration::from_secs(5), run) .await diff --git a/rs/moq-net/src/runtime.rs b/rs/moq-net/src/runtime.rs index a96e5505d0..f7621e5167 100644 --- a/rs/moq-net/src/runtime.rs +++ b/rs/moq-net/src/runtime.rs @@ -1,25 +1,15 @@ -//! Executor integration: who runs session machines and arms their timers. +//! Timer integration for session and origin drivers. //! -//! This crate never spawns onto a global executor and never reaches for ambient -//! state (no thread-locals, no globals). Instead, [`crate::Client::connect`] and -//! [`crate::Server::accept`] take a [`Runtime`], which supplies the two things a -//! session cannot do alone: run its protocol [`Machine`] to completion, and wake -//! a [`Timer`] when a deadline passes. +//! [`crate::Client::connect`] and [`crate::Server::accept`] take [`Timers`] and +//! return a [`crate::Driver`] for the caller to poll or spawn. This crate never +//! spawns tasks. Timer providers supply a clock and re-armable registrations, +//! without choosing a transport or executor. //! -//! Each transport ships its runtime: `moq-tokio` spawns machines onto tokio, -//! `moq-wasm` onto the browser microtask queue, and a thread-per-core runtime -//! keeps them on its own thread. `Test` (behind the `test-runtime` feature) is -//! the deterministic one for tests: nothing runs and no time passes unless the -//! test says so. -//! -//! There is deliberately no `Send` bound anywhere here. A runtime whose -//! [`Runtime::Transport`] is `Send` gets a `Send` [`Machine`] and may move it -//! across threads (tokio does); a pinned `!Send` transport stays on its thread. +//! `Test` (behind `test-runtime`) provides a virtual clock that advances only +//! when the test requests it. use std::task::Poll; -use crate::Error; - /// The instant type used for deadlines and annotations. /// /// [`std::time::Instant`] on native. The browser has no monotonic std clock, so @@ -52,10 +42,8 @@ pub trait Timer { /// The timer half of a runtime: mint [`Timer`]s and read the clock they follow. /// -/// Split from [`Runtime`] so work that only arms deadlines (the origin driver's -/// linger and hold-down machinery) can borrow a runtime's timers without caring -/// which transport it drives. Cloned into whatever arms timers, so clones must -/// be cheap (a ZST or a reference count). +/// Shared by session and origin drivers, independently of their transport or +/// executor. Clones must be cheap (a ZST or a reference count). pub trait Timers: Clone { /// The timer registration this runtime hands out. type Timer: Timer; @@ -76,27 +64,6 @@ pub trait Timers: Clone { } } -/// The executor a session runs on: [`Timers`] plus a transport and a spawn. -/// -/// Passed to [`crate::Client::connect`] / [`crate::Server::accept`], which hand -/// their protocol [`Machine`] to [`spawn`](Self::spawn). -pub trait Runtime: Timers { - /// The transport this runtime can drive. Pinning the transport per runtime - /// is what lets each implementation know the concrete `Machine` type it - /// spawns, including whether it is `Send`. - type Transport: crate::transport::poll::Session; - - /// Take ownership of a session's protocol machine and poll it to completion. - /// - /// The machine holds no [`crate::Session`] clone, so running it never keeps - /// the session alive; once the last session handle drops, the machine closes - /// the transport and finishes on its own. The machine is also the only thing - /// holding the transport, so dropping it instead tears the whole session - /// down: the transport closes and [`crate::Session::closed`] resolves with - /// [`crate::Error::Cancel`]. - fn spawn(&self, machine: Machine); -} - /// A boxed [`Timer`], minted by [`AnyTimers`]. pub(crate) struct BoxTimer(MaybeSendTimer); @@ -284,140 +251,16 @@ impl Deadline { } } -impl std::fmt::Debug for Deadline { +impl std::fmt::Debug for Deadline { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Deadline").field("at", &self.at).finish() } } -/// The future driving a [`crate::Session`]'s protocol state. -/// -/// Created by [`crate::Client::connect`] / [`crate::Server::accept`] and handed -/// straight to [`Runtime::spawn`]; the only thing to do with one is poll it, -/// either as a [`Future`] or by stepping [`poll`](Self::poll) from a -/// [`kio`]-style poll function. It resolves when the session ends, and keeps -/// returning that same result if polled again. -pub struct Machine { - state: MachineState, - // Retains the waiter across `Future` polls so its kio registrations stay - // live. Kept out of `MachineState` so the borrow `hold` hands back doesn't - // collide with the `&mut` that polling the state needs. - park: kio::Park, -} - -/// The protocol half of a machine, one variant per negotiated wire protocol. -/// -/// The lite driver is a named machine, so the machine's `Send`-ness follows -/// the transport (a pinned `!Send` transport yields a `!Send` machine that -/// stays on its thread). The ietf driver is still a boxed future; the box -/// demands `Send` on native, which is why the ietf path requires a -/// [`Boxable`](crate::transport::poll::Boxable) transport until it too becomes -/// a named machine. -pub(crate) enum Protocol { - /// Boxed for size only: a concrete box, so `Send` stays inferred. - Lite(Box>), - Ietf(crate::util::MaybeSendBox<'static, Result<(), Error>>), -} - -/// Everything the machine polls, split from the park so the two borrow -/// disjointly. -pub(crate) struct MachineState { - pub(crate) protocol: Protocol, - // The session supervisor, polled alongside the protocol: it executes the - // handles' close requests, publishes the transport's terminal error, and - // samples stats. It finishes once the transport reports closed, and the - // machine is not done until it has: the protocol's terminal transport close - // is what `Session::closed` observes, so resolving before it is published - // would leave waiters parked on a machine nobody polls again. `None` once - // finished, since a completed machine must not be polled again. - pub(crate) supervisor: Option>, - // Cached so a poll after completion doesn't re-poll a finished protocol. - pub(crate) result: Option>, -} - -impl Machine { - pub(crate) fn new(state: MachineState) -> Self { - Self { - state, - park: kio::Park::default(), - } - } - - /// Drive the protocol one step, registering `waiter` for the next wakeup. - /// - /// The `poll_*` counterpart of `.await`ing the machine, for runtimes - /// composing it into their own [`kio`]-style poll loops. - pub fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { - self.state.poll(waiter) - } -} - -impl Protocol { - fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { - match self { - Self::Lite(driver) => driver.poll(waiter), - Self::Ietf(driver) => waiter.poll_future(driver.as_mut()), - } - } -} - -impl MachineState { - fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { - if let Some(supervisor) = &mut self.supervisor - && supervisor.poll(waiter).is_ready() - { - self.supervisor = None; - } - - if self.result.is_none() - && let Poll::Ready(result) = self.protocol.poll(waiter) - { - self.result = Some(result); - // The protocol's last act was closing the transport, which wakes the - // supervisor's close watch; poll it now instead of waiting a turn. - if let Some(supervisor) = &mut self.supervisor - && supervisor.poll(waiter).is_ready() - { - self.supervisor = None; - } - } - - match (&self.result, &self.supervisor) { - (Some(result), None) => Poll::Ready(result.clone()), - _ => Poll::Pending, - } - } -} - -// Every part of the machine is a poll machine driven through `&mut` (the one -// pinned piece, the boxed ietf future, is `Unpin` itself), so nothing relies -// on address stability and the machine may move freely between polls. -impl Unpin for Machine {} - -impl Future for Machine { - type Output = Result<(), Error>; - - fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll { - let this = &mut *self; - // Disjoint field borrows: `hold` borrows the park for as long as the - // waiter lives, while the state is polled through its own `&mut`. - let waiter = this.park.hold(cx); - this.state.poll(waiter) - } -} - -impl std::fmt::Debug for Machine { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("Machine") - .field("done", &self.state.result.is_some()) - .finish() - } -} - #[cfg(feature = "test-runtime")] mod test; #[cfg(feature = "test-runtime")] -pub use test::{Never, Test}; +pub use test::Test; /// A tokio-backed runtime for this crate's own unit tests, so the existing /// `tokio::time::pause`/`advance` tests keep their semantics: `now` reads @@ -427,28 +270,20 @@ pub use test::{Never, Test}; /// in `tests/support`). #[cfg(all(test, not(target_family = "wasm")))] pub(crate) mod tokio_test { - use std::{marker::PhantomData, pin::Pin, task::Poll}; + use std::{pin::Pin, task::Poll}; - use super::{Instant, Machine, Runtime, Timer}; + use super::{Instant, Timer}; - // The default `S` makes a bare `Tokio::new()` a transportless timers handle - // for origin drivers, mirroring `moq_tokio::runtime::Runtime`. - pub(crate) struct Tokio(PhantomData); + #[derive(Clone, Default)] + pub(crate) struct Tokio; - impl Tokio { + impl Tokio { pub fn new() -> Self { - Self(PhantomData) - } - } - - impl Clone for Tokio { - fn clone(&self) -> Self { - Self(PhantomData) + Self } } - // Unbounded: timers don't involve the transport. - impl super::Timers for Tokio { + impl super::Timers for Tokio { type Timer = TokioTimer; fn timer(&self) -> Self::Timer { @@ -460,15 +295,6 @@ pub(crate) mod tokio_test { } } - // Boxable: tokio work-steals, so the machine must be `Send`. - impl Runtime for Tokio { - type Transport = S; - - fn spawn(&self, machine: Machine) { - tokio::spawn(machine); - } - } - pub(crate) struct TokioTimer { at: Option, // Allocated on the first poll after arming, then re-armed in place via diff --git a/rs/moq-net/src/runtime/test.rs b/rs/moq-net/src/runtime/test.rs index 63145ac2cb..d010c82eeb 100644 --- a/rs/moq-net/src/runtime/test.rs +++ b/rs/moq-net/src/runtime/test.rs @@ -1,38 +1,25 @@ -//! The deterministic test runtime: virtual time, manual machine stepping. +//! A virtual clock and timer registrations for deterministic tests. use std::{ collections::HashMap, - marker::PhantomData, sync::{Arc, Mutex}, task::Poll, }; -use super::{Instant, Machine, Runtime, Timer, Timers}; +use super::{Instant, Timer, Timers}; -/// A deterministic [`Runtime`] for tests: no thread runs machines and no time -/// passes unless the test says so. +/// A virtual clock for deterministic timer tests. /// -/// - [`tick`](Self::tick) polls every spawned machine once. -/// - [`advance`](Self::advance) moves the virtual clock, firing elapsed timers. -/// - [`advance_to_timer`](Self::advance_to_timer) jumps straight to the -/// earliest armed timer. -/// -/// [`Runtime::now`] reads the virtual clock, so deadlines armed relative to -/// "now" stay coherent with the advances a test performs. Clones share one -/// clock and machine queue. -/// -/// The transport parameter defaults to [`Never`] for tests that only exercise -/// timers. -pub struct Test { - shared: Arc>>, - _transport: PhantomData, +/// Clones share the clock. Advance time explicitly, then poll session or origin +/// drivers yourself to observe elapsed deadlines. +pub struct Test { + shared: Arc>, } -struct Shared { +struct Shared { now: Instant, timers: HashMap, next_id: u64, - machines: Vec>>, } struct Entry { @@ -40,7 +27,7 @@ struct Entry { waiters: kio::WaiterList, } -impl Test { +impl Test { /// A fresh runtime whose virtual clock starts at the real current instant. pub fn new() -> Self { Self { @@ -48,9 +35,7 @@ impl Test { now: Instant::now(), timers: HashMap::new(), next_id: 0, - machines: Vec::new(), })), - _transport: PhantomData, } } @@ -97,38 +82,10 @@ impl Test { None => false, } } - - /// Drop every spawned machine without polling it, like a runtime shutting - /// down mid-session. - /// - /// Machines hold runtime clones (for timers), so dropping every external - /// handle alone never drops them; this is the explicit teardown. - pub fn shutdown(&self) { - let machines = std::mem::take(&mut self.shared.lock().unwrap().machines); - drop(machines); - } - - /// Poll every spawned machine once, dropping the finished ones. - /// - /// Returns how many machines remain. Polling is unconditional (no waker - /// bookkeeping): a test advances state, ticks, and asserts. - pub fn tick(&self) -> usize { - // Take the machines out so their polls can reach the timers without - // deadlocking on the shared lock. - let mut machines = std::mem::take(&mut self.shared.lock().unwrap().machines); - let waiter = kio::Waiter::noop(); - machines.retain_mut(|machine| machine.poll(&waiter).is_pending()); - - let mut shared = self.shared.lock().unwrap(); - // A machine spawned by a machine mid-tick landed in the queue already; - // keep both. - shared.machines.extend(machines); - shared.machines.len() - } } -impl Timers for Test { - type Timer = TestTimer; +impl Timers for Test { + type Timer = TestTimer; fn timer(&self) -> Self::Timer { let id = { @@ -155,48 +112,38 @@ impl Timers for Test { } } -impl Runtime for Test { - type Transport = S; - - fn spawn(&self, machine: Machine) { - self.shared.lock().unwrap().machines.push(machine); - } -} - -impl Clone for Test { +impl Clone for Test { fn clone(&self) -> Self { Self { shared: self.shared.clone(), - _transport: PhantomData, } } } -impl Default for Test { +impl Default for Test { fn default() -> Self { Self::new() } } -impl std::fmt::Debug for Test { +impl std::fmt::Debug for Test { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let shared = self.shared.lock().unwrap(); f.debug_struct("Test") .field("now", &shared.now) .field("timers", &shared.timers.len()) - .field("machines", &shared.machines.len()) .finish() } } /// The [`Timer`] handed out by [`Test`]: elapsed exactly when its armed instant /// is at or before the shared virtual clock. -pub struct TestTimer { - shared: Arc>>, +pub struct TestTimer { + shared: Arc>, id: u64, } -impl Timer for TestTimer { +impl Timer for TestTimer { fn set(&mut self, at: Option) { let mut shared = self.shared.lock().unwrap(); if let Some(entry) = shared.timers.get_mut(&self.id) { @@ -220,128 +167,12 @@ impl Timer for TestTimer { } } -impl Drop for TestTimer { +impl Drop for TestTimer { fn drop(&mut self) { self.shared.lock().unwrap().timers.remove(&self.id); } } -/// An uninhabited transport, for [`Test`] runtimes that never open a session. -#[derive(Debug, Clone, Copy)] -pub enum Never {} - -impl std::fmt::Display for Never { - fn fmt(&self, _: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match *self {} - } -} - -impl std::error::Error for Never {} - -impl web_transport_trait::Error for Never { - fn session_error(&self) -> Option<(u32, String)> { - match *self {} - } -} - -impl web_transport_trait::poll::Session for Never { - type SendStream = Never; - type RecvStream = Never; - type Error = Never; - - fn poll_accept_uni(&mut self, _: &mut std::task::Context<'_>) -> Poll> { - match *self {} - } - - fn poll_accept_bi( - &mut self, - _: &mut std::task::Context<'_>, - ) -> Poll, Self::Error>> { - match *self {} - } - - fn poll_open_uni(&mut self, _: &mut std::task::Context<'_>) -> Poll> { - match *self {} - } - - fn poll_open_bi( - &mut self, - _: &mut std::task::Context<'_>, - ) -> Poll, Self::Error>> { - match *self {} - } - - fn poll_send_datagram(&mut self, _: &mut std::task::Context<'_>, _: &[u8]) -> Poll> { - match *self {} - } - - fn poll_recv_datagram(&mut self, _: &mut std::task::Context<'_>) -> Poll> { - match *self {} - } - - fn max_datagram_size(&self) -> usize { - match *self {} - } - - fn protocol(&self) -> Option<&str> { - match *self {} - } - - fn close(&mut self, _: u32, _: &str) { - match *self {} - } - - fn poll_closed(&mut self, _: &mut std::task::Context<'_>) -> Poll { - match *self {} - } - - fn stats(&self) -> impl web_transport_trait::Stats { - // Uninhabited, so this is never called; a concrete type keeps the - // opaque return type nameable. - web_transport_trait::StatsUnavailable - } -} - -impl web_transport_trait::poll::SendStream for Never { - type Error = Never; - - fn poll_write(&mut self, _: &mut std::task::Context<'_>, _: &[u8]) -> Poll> { - match *self {} - } - - fn set_priority(&mut self, _: u8) { - match *self {} - } - - fn finish(&mut self) -> Result<(), Self::Error> { - match *self {} - } - - fn reset(&mut self, _: u32) { - match *self {} - } - - fn poll_closed(&mut self, _: &mut std::task::Context<'_>) -> Poll> { - match *self {} - } -} - -impl web_transport_trait::poll::RecvStream for Never { - type Error = Never; - - fn poll_read(&mut self, _: &mut std::task::Context<'_>, _: &mut [u8]) -> Poll, Self::Error>> { - match *self {} - } - - fn stop(&mut self, _: u32) { - match *self {} - } - - fn poll_closed(&mut self, _: &mut std::task::Context<'_>) -> Poll> { - match *self {} - } -} - #[cfg(all(test, not(loom)))] mod tests { use std::time::Duration; @@ -371,7 +202,7 @@ mod tests { #[test] fn rearm_after_advance_lands_in_the_future() { - // The reason Runtime::now exists: a deadline armed relative to "now" + // The reason Timers::now exists: a deadline armed relative to "now" // after a big advance must not sit in the virtual past. let rt: Test = Test::new(); rt.advance(Duration::from_secs(3600)); diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 230a02ce7d..3cf2aadc3c 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -9,9 +9,7 @@ use crate::{ ietf, lite, setup, stats, }; -// The transport methods are called on the projected `R::Transport`, which needs the -// trait itself in scope (a plain type parameter would not). -use web_transport_trait::{MaybeSend, MaybeSync, poll::Session as _}; +use web_transport_trait::{MaybeSend, MaybeSync}; /// A MoQ server session builder. #[derive(Default, Clone)] @@ -79,17 +77,18 @@ impl Server { } /// Start a lite session on an accepted transport: wire the origins, answer - /// with our SETUP, and hand the machine to the runtime. - fn start_lite( + /// with our SETUP, and return the session and its driver. + fn start_lite( &self, runtime: R, - session: R::Transport, + session: S, version: lite::Version, client_setup: Option, peer_hop: Option, - ) -> Result + ) -> Result<(Session, crate::Driver), Error> where - R: crate::runtime::Runtime + 'static, + S: crate::transport::poll::Session, + R: crate::runtime::Timers + 'static, { let (publish, subscribe) = self.stat_tagged_origins(); @@ -121,12 +120,12 @@ impl Server { peer_setup: client_setup, })?; - Ok(Session::spawn( + Ok(Session::new( runtime, session, version.into(), start.recv_bandwidth, - crate::runtime::Protocol::Lite(Box::new(start.driver)), + crate::driver::Protocol::Lite(Box::new(start.driver)), start.goaway, )) } @@ -139,9 +138,10 @@ impl Server { /// with [`Error::Version`]). Completes the handshake immediately; a caller /// gating on the advertised path uses /// [`accept_request_lite`](Self::accept_request_lite) instead. - pub async fn accept_lite(&self, runtime: R, session: R::Transport) -> Result + pub async fn accept_lite(&self, runtime: R, session: S) -> Result<(Session, crate::Driver), Error> where - R: crate::runtime::Runtime + 'static, + S: crate::transport::poll::Session, + R: crate::runtime::Timers + 'static, { self.accept_request_lite(runtime, session).await?.ok().await } @@ -151,13 +151,10 @@ impl Server { /// which is what drops the thread-affinity bounds: a pinned `!Send` /// transport can gate on the advertised path too. Anything but a moq-lite /// ALPN is refused with [`Error::Version`]. - pub async fn accept_request_lite( - &self, - runtime: R, - mut session: R::Transport, - ) -> Result, Error> + pub async fn accept_request_lite(&self, runtime: R, mut session: S) -> Result, Error> where - R: crate::runtime::Runtime + 'static, + S: crate::transport::poll::Session, + R: crate::runtime::Timers + 'static, { let (path, role, origin, handshake) = match session.protocol() { Some(alpn @ (ALPN_LITE_05 | ALPN_LITE_06_WIP)) => { @@ -226,21 +223,20 @@ impl Server { }) } - /// Perform the MoQ handshake as a server, returning the [`Session`]. + /// Perform the MoQ handshake as a server, returning the [`Session`] and its [`Driver`](crate::Driver). /// - /// The session's protocol machine is handed to `runtime` - /// ([`Runtime::spawn`](crate::runtime::Runtime::spawn)), so there is nothing - /// else to drive. + /// Poll or spawn the returned driver to run the session. `runtime` supplies + /// only its clock and timers. /// /// Convenience wrapper over [`accept_request`](Self::accept_request) that /// completes the handshake immediately. Use `accept_request` when you need to /// inspect the client's advertised path before deciding what to serve. - pub async fn accept(&self, runtime: R, session: R::Transport) -> Result + pub async fn accept(&self, runtime: R, session: S) -> Result<(Session, crate::Driver), Error> where - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, - R::Transport: crate::transport::poll::Boxable, - ::SendStream: MaybeSync, - ::RecvStream: MaybeSync, + S: crate::transport::poll::Boxable, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, + S::SendStream: MaybeSync, + S::RecvStream: MaybeSync, R::Timer: MaybeSend, { self.accept_request(runtime, session).await?.ok().await @@ -256,16 +252,12 @@ impl Server { /// /// The path is surfaced for moq-lite-05 and every moq-transport draft we speak; /// it's empty on versions with no in-band request path (e.g. lite 01-04). - pub async fn accept_request( - &self, - runtime: R, - mut session: R::Transport, - ) -> Result, Error> + pub async fn accept_request(&self, runtime: R, mut session: S) -> Result, Error> where - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, - R::Transport: crate::transport::poll::Boxable, - ::SendStream: MaybeSync, - ::RecvStream: MaybeSync, + S: crate::transport::poll::Boxable, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, + S::SendStream: MaybeSync, + S::RecvStream: MaybeSync, R::Timer: MaybeSend, { let (encoding, supported) = match session.protocol() { @@ -373,17 +365,17 @@ impl Server { /// Read a draft-17/18 client's SETUP (with its request path) off its uni stream, /// then pause. `ok()` starts the session and hands the stream back for GOAWAY. - async fn accept_ietf_modern( + async fn accept_ietf_modern( &self, runtime: R, - mut session: R::Transport, + mut session: S, version: ietf::Version, - ) -> Result, Error> + ) -> Result, Error> where - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, - R::Transport: crate::transport::poll::Boxable, - ::SendStream: MaybeSync, - ::RecvStream: MaybeSync, + S: crate::transport::poll::Boxable, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, + S::SendStream: MaybeSync, + S::RecvStream: MaybeSync, R::Timer: MaybeSend, { let peer_setup = ietf::accept_setup(&mut session, version).await?; @@ -414,7 +406,7 @@ impl Server { /// the origins to serve, then call [`ok`](Self::ok) to complete the handshake, or /// [`close`](Self::close) to reject it. Modeled on the WebTransport `Request` in /// moq-tokio. -pub struct Handshake { +pub struct Handshake { path: Option, role: Option, origin: Option, @@ -427,16 +419,16 @@ pub struct Handshake { +struct RequestInner { server: Server, - /// Receives the session's machine once `ok()` completes the handshake. + /// Supplies the clock and timers for the accepted session. runtime: R, handshake: PausedHandshake, } /// The handshake state captured at the pause point. Every variant defers its /// session start to [`Handshake::ok`] so origins set on the handshake still apply. -enum PausedHandshake { +enum PausedHandshake { /// moq-lite 03/04: no Setup Stream. LiteBare { session: S, version: lite::Version }, /// moq-lite 05+: the client's Setup Stream has been read. `ok()` starts the @@ -452,23 +444,20 @@ enum PausedHandshake>), + Boxed(Box>), } +type Accept = crate::util::MaybeSendBox<'static, Result<(Session, crate::Driver), Error>>; + /// A paused non-lite handshake. See [`PausedHandshake::Boxed`] for why this is a /// trait object. /// /// `MaybeSync` is not decoration: a caller holding a [`Handshake`] across an /// await behind `&self` (moq-relay authenticates that way) needs /// `&Handshake: Send`, which is `Handshake: Sync`, which is this. -trait Paused: MaybeSend + MaybeSync { +trait Paused: MaybeSend + MaybeSync { /// Complete the handshake with the final server config. - fn ok( - self: Box, - server: Server, - runtime: R, - peer_hop: Option, - ) -> crate::util::MaybeSendBox<'static, Result>; + fn ok(self: Box, server: Server, runtime: R, peer_hop: Option) -> Accept; /// Reject the handshake, closing the transport with `err`'s wire code. fn close(self: Box, err: Error); @@ -483,20 +472,15 @@ struct PausedIetfModern { peer_setup: ietf::PeerSetup, } -impl Paused for PausedIetfModern +impl Paused for PausedIetfModern where - S: crate::transport::poll::Session + crate::transport::poll::Boxable, + S: crate::transport::poll::Boxable, S::SendStream: MaybeSync, S::RecvStream: MaybeSync, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { - fn ok( - self: Box, - server: Server, - runtime: R, - peer_hop: Option, - ) -> crate::util::MaybeSendBox<'static, Result> { + fn ok(self: Box, server: Server, runtime: R, peer_hop: Option) -> Accept { use crate::util::MaybeBoxedExt as _; async move { let Self { @@ -525,12 +509,12 @@ where peer_declared: Some(peer_setup.declared), })?; tracing::debug!(?version, "connected"); - Ok(Session::spawn( + Ok(Session::new( runtime, session, version.into(), None, - crate::runtime::Protocol::Ietf(protocol), + crate::driver::Protocol::Ietf(protocol), goaway, )) } @@ -555,20 +539,15 @@ struct PausedLegacy { peer_declared: ietf::peer::Peer, } -impl Paused for PausedLegacy +impl Paused for PausedLegacy where - S: crate::transport::poll::Session + crate::transport::poll::Boxable, + S: crate::transport::poll::Boxable, S::SendStream: MaybeSync, S::RecvStream: MaybeSync, - R: crate::runtime::Runtime + MaybeSend + MaybeSync + 'static, + R: crate::runtime::Timers + MaybeSend + MaybeSync + 'static, R::Timer: MaybeSend, { - fn ok( - self: Box, - server: Server, - runtime: R, - peer_hop: Option, - ) -> crate::util::MaybeSendBox<'static, Result> { + fn ok(self: Box, server: Server, runtime: R, peer_hop: Option) -> Accept { use crate::util::MaybeBoxedExt as _; async move { let Self { @@ -615,7 +594,7 @@ where })?; ( start.recv_bandwidth, - crate::runtime::Protocol::Lite(Box::new(start.driver)), + crate::driver::Protocol::Lite(Box::new(start.driver)), start.goaway, ) } @@ -637,11 +616,11 @@ where peer_setup_stream: None, peer_declared: Some(peer_declared), })?; - (None, crate::runtime::Protocol::Ietf(protocol), goaway) + (None, crate::driver::Protocol::Ietf(protocol), goaway) } }; - Ok(Session::spawn(runtime, session, version, recv_bw, protocol, goaway)) + Ok(Session::new(runtime, session, version, recv_bw, protocol, goaway)) } .maybe_boxed() } @@ -655,7 +634,7 @@ where impl Handshake where S: crate::transport::poll::Session, - R: crate::runtime::Runtime + 'static, + R: crate::runtime::Timers + 'static, { /// The request path the client advertised in its SETUP. /// @@ -732,11 +711,10 @@ where self.inner.as_mut().expect("request already responded") } - /// Accept the session, returning the [`Session`]. + /// Accept the session, returning the [`Session`] and its [`Driver`](crate::Driver). /// - /// The session's protocol machine is handed to the runtime given to - /// [`Server::accept_request`], so there is nothing else to drive. - pub async fn ok(mut self) -> Result { + /// Poll or spawn the returned driver to run the session. + pub async fn ok(mut self) -> Result<(Session, crate::Driver), Error> { let peer_hop = Some(self.assigned_hop); let RequestInner { server, @@ -764,7 +742,7 @@ where } } -impl RequestInner { +impl RequestInner { fn close(self, err: Error) { let mut session = match self.handshake { PausedHandshake::LiteBare { session, .. } => session, @@ -775,7 +753,7 @@ impl RequestInne } } -impl Drop for Handshake { +impl Drop for Handshake { // A dropped request would otherwise leave the client hanging until its idle // timeout: it already sent SETUP and is waiting on a response. Reject loudly. fn drop(&mut self) { @@ -1141,7 +1119,8 @@ mod tests { .announce("local-route", crate::origin::Route::default().with_hops(local_hops)) .unwrap(); - let session = request.ok().await.unwrap(); + let (session, driver) = request.ok().await.unwrap(); + tokio::spawn(driver); for _ in 0..100 { if occurrences(&log, b"local-route") > 0 { diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index 286afce81d..b7bb52b7a6 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -6,16 +6,16 @@ use web_transport_trait::Stats as _; use crate::{Error, SessionError, Version, bandwidth, goaway}; -/// A close requested by a session handle, executed by the machine. +/// A close requested by a session handle, executed by the driver. #[derive(Clone)] struct Close { code: u32, reason: String, } -/// The stats cell shared between the machine's sampler and the handles. +/// The stats cell shared between the driver's sampler and the handles. struct StatsState { - /// The latest sample the machine took (or the construction-time snapshot). + /// The latest sample the driver took (or the construction-time snapshot). sample: Stats, /// A handle read the stats since the last sample: keep sampling. demanded: bool, @@ -62,25 +62,24 @@ pub struct Stats { /// A MoQ transport session, wrapping a WebTransport connection. /// -/// Returned by [`crate::Client::connect`] and [`crate::Server::accept`], which hand -/// the session's protocol [`runtime::Machine`](crate::runtime::Machine) to the -/// [`Runtime`](crate::runtime::Runtime) they were given: that runtime is the only -/// thing driving the session. +/// Returned with a [`Driver`](crate::Driver) by [`crate::Client::connect`] and +/// [`crate::Server::accept`]. The caller must poll or spawn that driver to run +/// the session. /// /// Like every handle in this library, the lifecycle is reference counted: clones /// share the connection, the transport closes when the last clone drops, and /// [`abort`](Self::abort) closes it explicitly with an error. The handle and the -/// machine are severed in both directions: the machine holds no `Session` clone, -/// so the runtime running it never keeps the session alive, and the `Session` +/// driver are severed in both directions: the driver holds no `Session` clone, +/// so running it never keeps the session alive, and the `Session` /// holds no transport, so the handle is `Send + Sync` whatever transport the -/// runtime drives. Everything transport-shaped (the close, the close reason, -/// the stats sample) is relayed through the machine. +/// driver uses. Everything transport-shaped (the close, the close reason, +/// the stats sample) is relayed through the driver. #[derive(Clone)] pub struct Session { - /// Handle side to machine: `Some` once [`abort`](Self::abort) ran; the + /// Handle side to driver: `Some` once [`abort`](Self::abort) ran; the /// channel closing (the last handle dropping) is the implicit Cancel. close: kio::Producer>, - /// Machine to handle side: the transport's terminal error. + /// Driver to handle side: the transport's terminal error. closed: kio::Consumer>, stats: kio::Shared, version: Version, @@ -112,7 +111,7 @@ impl Session { /// Returns a snapshot of the current connection statistics. /// /// Cheap and non-blocking: this reads the latest sample the session's - /// machine took, and schedules a refresh, so periodic polling observes + /// driver took, and schedules a refresh, so periodic polling observes /// fresh counters (100ms cadence). See [`Stats`] for which /// metrics each backend reports. pub fn stats(&self) -> Stats { @@ -132,7 +131,7 @@ impl Session { /// Close the transport with an explicit error, instead of waiting for the last /// clone to drop. Idempotent: the first close wins. /// - /// The close is executed by the session's machine, so it reaches the wire + /// The close is executed by the session's driver, so it reaches the wire /// once the runtime polls it (immediately on a live runtime). pub fn abort(&self, err: Error) { if let Ok(mut close) = self.close.write() @@ -151,7 +150,7 @@ impl Session { /// rejection arrives as `Error::Session(SessionError::Unauthorized)`); every peer code is /// preserved as [`Error::Session`], and a close carrying no application code surfaces as /// [`Error::Transport`]. See [`Error::from_transport`]. If the runtime drops - /// the machine instead of running it to completion, this resolves with + /// the driver instead of running it to completion, this resolves with /// [`Error::Cancel`]. pub async fn closed(&self) -> Error { match self @@ -163,7 +162,7 @@ impl Session { .await { Ok(err) => err, - // The machine was dropped before it could observe the close. + // The driver was dropped before it could observe the close. Err(kio::Closed) => Error::Cancel, } } @@ -200,16 +199,17 @@ impl Session { } impl Session { - pub(super) fn new( + pub(super) fn new( runtime: R, - session: R::Transport, + session: S, version: Version, recv_bandwidth: Option, - protocol: crate::runtime::Protocol, + protocol: crate::driver::Protocol, goaway: goaway::Handle, - ) -> (Self, crate::runtime::Machine) + ) -> (Self, crate::Driver) where - R: crate::runtime::Runtime + 'static, + S: crate::transport::poll::Session, + R: crate::runtime::Timers + 'static, { let sample = snapshot(&session); @@ -249,34 +249,17 @@ impl Session { recv_bandwidth, goaway: Arc::new(goaway), }; - let machine = crate::runtime::Machine::new(crate::runtime::MachineState { + let driver = crate::Driver::new(crate::driver::State { protocol, supervisor: Some(supervisor), result: None, }); - (session, machine) - } - - /// Build the session, hand its machine to the runtime, and return the handle. - pub(super) fn spawn( - runtime: R, - session: R::Transport, - version: Version, - recv_bandwidth: Option, - protocol: crate::runtime::Protocol, - goaway: goaway::Handle, - ) -> Self - where - R: crate::runtime::Runtime + 'static, - { - let (session, machine) = Self::new(runtime.clone(), session, version, recv_bandwidth, protocol, goaway); - runtime.spawn(machine); - session + (session, driver) } } -/// The machine's transport-facing half of a [`Session`]: it executes the +/// The driver's transport-facing half of a [`Session`]: it executes the /// handles' close requests, publishes the transport's terminal error, and /// samples the connection stats (including the send-bandwidth estimate) while /// anyone is consuming them. diff --git a/rs/moq-net/tests/datagram.rs b/rs/moq-net/tests/datagram.rs index 4c7bec9f2c..2a5681df09 100644 --- a/rs/moq-net/tests/datagram.rs +++ b/rs/moq-net/tests/datagram.rs @@ -21,7 +21,7 @@ const PAYLOAD: &[u8] = b"datagram payload"; /// Build an origin producer, spawning its driver on the ambient runtime. fn produce_origin(hop: u64) -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); - tokio::spawn(driver.run(TokioRuntime::<()>::new())); + tokio::spawn(driver.run(TokioRuntime::new())); producer } diff --git a/rs/moq-net/tests/goaway.rs b/rs/moq-net/tests/goaway.rs index 035269e921..26961c5ce4 100644 --- a/rs/moq-net/tests/goaway.rs +++ b/rs/moq-net/tests/goaway.rs @@ -15,7 +15,7 @@ use support::mock::create_mock_session_pair; /// Build an origin producer, spawning its driver on the ambient runtime. fn produce_origin(hop: Hop) -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(hop)); - tokio::spawn(driver.run(support::harness::TokioRuntime::<()>::new())); + tokio::spawn(driver.run(support::harness::TokioRuntime::new())); producer } @@ -277,8 +277,10 @@ async fn duplicate_goaway_keeps_first_payload_moq_lite_04() { client.connect(support::harness::TokioRuntime::new(), client_transport), server.accept(support::harness::TokioRuntime::new(), server_transport) ); - let client_session = client_result.expect("client handshake failed"); - let _server_session = server_result.expect("server handshake failed"); + let (client_session, client_driver) = client_result.expect("client handshake failed"); + tokio::spawn(client_driver); + let (_server_session, server_driver) = server_result.expect("server handshake failed"); + tokio::spawn(server_driver); // First GOAWAY: observed with its URI. Waiting for the peer to close the // stream guarantees the control message was fully processed. diff --git a/rs/moq-net/tests/support/harness.rs b/rs/moq-net/tests/support/harness.rs index 11b303bc7b..85ead4fd0c 100644 --- a/rs/moq-net/tests/support/harness.rs +++ b/rs/moq-net/tests/support/harness.rs @@ -2,44 +2,28 @@ //! [`moq_net::Session`]s over the in-memory mock transport. //! //! The harness runs the full MoQ handshake (Client::connect + Server::accept) -//! over a [`MockSession`] pair and spawns both protocol drivers, giving tests +//! over a [`super::mock::MockSession`] pair and spawns both protocol drivers, giving tests //! two live sessions ready for pub/sub without any real QUIC or network I/O. #![allow(dead_code)] -use std::{marker::PhantomData, pin::Pin, task::Poll}; +use std::{pin::Pin, task::Poll}; use moq_net::{Client, Server, Session, Version, origin}; -use super::mock::{MockSession, create_mock_session_pair}; +use super::mock::create_mock_session_pair; -/// A tokio-backed [`moq_net::runtime::Runtime`] for these tests: machines are -/// spawned onto tokio and timers are tokio sleeps, so `tokio::time::pause` keeps -/// working. A copy of the crate's own `runtime::tokio_test` module, which is -/// `cfg(test)` and therefore invisible to integration tests. -pub struct TokioRuntime(PhantomData); +/// Tokio's clock and timers for these tests. +#[derive(Clone, Default)] +pub struct TokioRuntime; -impl TokioRuntime { +impl TokioRuntime { pub fn new() -> Self { - Self(PhantomData) + Self } } -impl Clone for TokioRuntime { - fn clone(&self) -> Self { - Self(PhantomData) - } -} - -impl Default for TokioRuntime { - fn default() -> Self { - Self::new() - } -} - -// Unbounded: timers don't involve the transport, so origin drivers can borrow -// a transportless handle (`TokioRuntime::<()>::new()`). -impl moq_net::runtime::Timers for TokioRuntime { +impl moq_net::runtime::Timers for TokioRuntime { type Timer = TokioTimer; fn timer(&self) -> Self::Timer { @@ -51,15 +35,6 @@ impl moq_net::runtime::Timers for TokioRuntime { } } -// Boxable: tokio work-steals, so the machine must be `Send`. -impl moq_net::runtime::Runtime for TokioRuntime { - type Transport = S; - - fn spawn(&self, machine: moq_net::runtime::Machine) { - tokio::spawn(machine); - } -} - /// The [`moq_net::runtime::Timer`] handed out by [`TokioRuntime`]. pub struct TokioTimer { at: Option, @@ -151,21 +126,25 @@ pub async fn connect_mock(opts: MockConnectOptions) -> MockPair { server = server.with_subscriber(subscribe); } - // Run both handshakes concurrently; the runtime spawns each side's machine + // Run both handshakes concurrently and spawn each side's driver // the moment its handshake resolves: on draft-17+ the server's accept blocks // on the client's SETUP, which only reaches the wire once the client's // machine is polled (and vice versa for the server's own SETUP). let client_fut = async { - client + let (session, driver) = client .connect(TokioRuntime::new(), client_transport) .await - .expect("client handshake failed") + .expect("client handshake failed"); + tokio::spawn(driver); + session }; let server_fut = async { - server + let (session, driver) = server .accept(TokioRuntime::new(), server_transport) .await - .expect("server handshake failed") + .expect("server handshake failed"); + tokio::spawn(driver); + session }; let (client_session, server_session) = tokio::join!(client_fut, server_fut); diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 76718741ed..5b58769a9e 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -757,7 +757,12 @@ async fn serve_connection( if let Some(publish) = grants.publish { request = request.with_subscriber(publish); } - let session = request.ok().await?; + let (session, driver) = request.ok().await?; + handle.spawn(async move { + if let Err(err) = driver.await { + tracing::debug!(%err, "session driver ended"); + } + }); let node_connection = peer_hop.map(|origin| serve.cluster.nodes.connect_inbound(id, origin)); tracing::info!(id, version = %session.version(), transport = %moq_tokio::server::Transport::Quic, "negotiated"); diff --git a/rs/moq-relay/src/websocket.rs b/rs/moq-relay/src/websocket.rs index 1a0323a367..c14a6dff05 100644 --- a/rs/moq-relay/src/websocket.rs +++ b/rs/moq-relay/src/websocket.rs @@ -175,14 +175,13 @@ where if let Some(publish) = publish { server = server.with_subscriber(publish); } - // Hold the session so it doesn't close early; the machine serves it in place - // (an Inline runtime hands it back instead of spawning), so its lifetime and - // teardown stay tied to this handler task. - let runtime = moq_tokio::runtime::Inline::new(); - let session = server - .accept(runtime.clone(), moq_tokio::transport::Session::new(ws)) + // Keep the driver in this task so cancellation tears down the transport. + let (session, mut driver) = server + .accept( + moq_tokio::runtime::Runtime::new(), + moq_tokio::transport::Session::new(ws), + ) .await?; - let mut driver = runtime.take().expect("accept hands the machine to its runtime"); // The handshake is done, so this is a MoQ session now: only now can a push // be serviced, and only now does the session appear in the live table. diff --git a/rs/moq-rtc/src/egress.rs b/rs/moq-rtc/src/egress.rs index 07676d14ce..a84ab4fd27 100644 --- a/rs/moq-rtc/src/egress.rs +++ b/rs/moq-rtc/src/egress.rs @@ -302,7 +302,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-rtc/src/server/mod.rs b/rs/moq-rtc/src/server/mod.rs index 88e549f2cd..e179710b91 100644 --- a/rs/moq-rtc/src/server/mod.rs +++ b/rs/moq-rtc/src/server/mod.rs @@ -302,7 +302,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-srt/src/dial.rs b/rs/moq-srt/src/dial.rs index 4f49072a91..8ee36fcbd1 100644 --- a/rs/moq-srt/src/dial.rs +++ b/rs/moq-srt/src/dial.rs @@ -148,7 +148,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-srt/src/server.rs b/rs/moq-srt/src/server.rs index 38c33b0c32..3e205dc0e9 100644 --- a/rs/moq-srt/src/server.rs +++ b/rs/moq-srt/src/server.rs @@ -819,7 +819,7 @@ mod tests { const OFFSET: u64 = 600_000_000; let (origin, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); let mut broadcast = origin.create_broadcast("rewind").unwrap(); broadcast.announce(moq_net::origin::Route::default()).unwrap(); let mut catalog = moq_mux::catalog::Producer::with_catalog( diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index 9a8e6b894a..84058c7279 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -146,7 +146,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index 847b97b2f9..0646ab44b6 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -480,7 +480,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-stats/src/consume.rs b/rs/moq-stats/src/consume.rs index cedf67e1f8..e7004095da 100644 --- a/rs/moq-stats/src/consume.rs +++ b/rs/moq-stats/src/consume.rs @@ -108,7 +108,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index b651ebd479..25f99c5a1a 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -887,7 +887,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index f60e110ae3..11cac2b897 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -394,9 +394,7 @@ impl Client { if url.scheme() == "tcp" { let session = crate::tcp::connect(url, &self.versions.alpns(), self.failover_delay, self.resolution_delay).await?; - return Ok(moq - .connect(crate::runtime::Runtime::new(), crate::transport::Session::new(session)) - .await?); + return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?); } // Unix domain socket (qmux, no TLS). Same-host only; the server can @@ -404,9 +402,7 @@ impl Client { #[cfg(all(feature = "uds", unix))] if url.scheme() == "unix" { let session = crate::unix::connect(url, &self.versions.alpns()).await?; - return Ok(moq - .connect(crate::runtime::Runtime::new(), crate::transport::Session::new(session)) - .await?); + return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?); } // A WebSocket URL names its transport. No QUIC backend can dial it, so there is @@ -431,9 +427,7 @@ impl Client { crate::iroh::Binding::H3 => self.moq.clone(), }; - return Ok(moq - .connect(crate::runtime::Runtime::new(), crate::transport::Session::new(session)) - .await?); + return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?); } #[cfg(feature = "noq")] @@ -455,7 +449,7 @@ impl Client { #[cfg(not(feature = "websocket"))] { let session = quic_handle.await?; - return Ok(moq.connect(crate::runtime::Runtime::new(), session).await?); + return Ok(connect_session(&moq, session).await?); } } @@ -478,7 +472,7 @@ impl Client { #[cfg(not(feature = "websocket"))] { let session = quic_handle.await?; - return Ok(moq.connect(crate::runtime::Runtime::new(), session).await?); + return Ok(connect_session(&moq, session).await?); } } @@ -495,7 +489,7 @@ impl Client { #[cfg(not(feature = "websocket"))] { let session = quic_handle.await?; - return Ok(moq.connect(crate::runtime::Runtime::new(), session).await?); + return Ok(connect_session(&moq, session).await?); } } @@ -513,10 +507,7 @@ impl Client { let alpns = self.versions.alpns(); let session = crate::websocket::connect(&self.websocket, &self.tls, self.tls_host_name.as_deref(), addr, &alpns).await?; - Ok(self - .moq - .connect(crate::runtime::Runtime::new(), crate::transport::Session::new(session)) - .await?) + Ok(connect_session(&self.moq, crate::transport::Session::new(session)).await?) } /// Race the QUIC dial against the WebSocket fallback, handshaking whichever wins. @@ -549,14 +540,10 @@ impl Client { }; match race_transport_connect(quic, websocket).await? { - TransportRace::Quic(quic) => Ok(moq.connect(crate::runtime::Runtime::new(), quic).await?), - TransportRace::WebSocket(websocket) => Ok(self - .moq - .connect( - crate::runtime::Runtime::new(), - crate::transport::Session::new(websocket), - ) - .await?), + TransportRace::Quic(quic) => Ok(connect_session(moq, quic).await?), + TransportRace::WebSocket(websocket) => { + Ok(connect_session(&self.moq, crate::transport::Session::new(websocket)).await?) + } } } } @@ -692,6 +679,25 @@ where } } +#[cfg(any( + feature = "noq", + feature = "quinn", + feature = "quiche", + feature = "iroh", + feature = "websocket", + feature = "tcp", + all(feature = "uds", unix) +))] +async fn connect_session( + client: &moq_net::Client, + transport: S, +) -> Result { + let (session, driver) = client.connect(crate::runtime::Runtime::new(), transport).await?; + use tracing::Instrument; + tokio::spawn(driver.instrument(tracing::Span::current())); + Ok(session) +} + #[cfg(test)] mod tests { use super::*; diff --git a/rs/moq-tokio/src/origin.rs b/rs/moq-tokio/src/origin.rs index 0320543b35..c199159475 100644 --- a/rs/moq-tokio/src/origin.rs +++ b/rs/moq-tokio/src/origin.rs @@ -14,7 +14,7 @@ /// [`Hop`](moq_net::Hop) id (which uses the default config). pub fn spawn(config: impl Into) -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(config.into()); - tokio::spawn(driver.run(crate::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(crate::runtime::Runtime::new())); producer } diff --git a/rs/moq-tokio/src/runtime.rs b/rs/moq-tokio/src/runtime.rs index 4d1c1769c3..6f59f3ccca 100644 --- a/rs/moq-tokio/src/runtime.rs +++ b/rs/moq-tokio/src/runtime.rs @@ -1,46 +1,19 @@ -//! The tokio runtime handle passed to moq-net's session entry points. +//! Tokio-backed timers for MoQ drivers. -use std::{marker::PhantomData, pin::Pin, task::Poll}; +use std::{pin::Pin, task::Poll}; -/// The [`moq_net::Runtime`] for tokio: protocol machines are `tokio::spawn`ed -/// and timers are tokio sleeps. -/// -/// One runtime drives one transport type (`S`), so each dial path constructs -/// its own with [`Runtime::new`]; it is a zero-sized handle. The default `S` -/// makes a bare `Runtime::new()` a transportless [`moq_net::Timers`] handle, -/// for work that only arms timers (the origin driver). -pub struct Runtime(PhantomData); +/// Supplies Tokio's clock and timers to MoQ drivers. +#[derive(Clone, Copy, Debug, Default)] +pub struct Runtime; -impl Runtime { - /// A handle for the transport type this dial produces. +impl Runtime { + /// A handle to Tokio's clock and timers. pub fn new() -> Self { - Self(PhantomData) + Self } } -impl Default for Runtime { - fn default() -> Self { - Self::new() - } -} - -impl Clone for Runtime { - fn clone(&self) -> Self { - *self - } -} - -impl Copy for Runtime {} - -impl std::fmt::Debug for Runtime { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("Runtime").finish() - } -} - -// Unbounded: timers don't involve the transport, so any `Runtime` (including -// the transportless default) hands them out. -impl moq_net::Timers for Runtime { +impl moq_net::Timers for Runtime { type Timer = Timer; fn timer(&self) -> Self::Timer { @@ -54,104 +27,6 @@ impl moq_net::Timers for Runtime { } } -// Boxable: tokio work-steals, so the machine must be `Send`. -impl moq_net::Runtime for Runtime { - type Transport = S; - - fn spawn(&self, machine: moq_net::runtime::Machine) { - // The session surfaces the result through `closed()`; the task's own - // result is nothing to keep. - use tracing::Instrument; - tokio::spawn(machine.instrument(tracing::Span::current())); - } -} - -/// A tokio-timer runtime that hands the machine back instead of spawning it. -/// -/// For callers that drive the session inline, in their own task, so its -/// lifetime and teardown stay tied to theirs (the relay's WebSocket handler -/// does this). Accept or connect with a clone, then [`take`](Self::take) the -/// machine and poll it yourself. -/// -/// One handle serves exactly one session. Clones share its slot, which is how -/// the machine gets back from the clone `accept`/`connect` took, so a second -/// session through the same handle panics rather than silently taking the -/// first one's place: whichever way that resolved, some session would end up -/// with the wrong driver or none at all, and a session nobody drives makes no -/// progress and never processes its shutdown. -pub struct Inline { - slot: std::sync::Arc>>>, - _transport: PhantomData, -} - -impl Inline { - /// An empty handle; the next accept/connect through it fills the slot. - pub fn new() -> Self { - Self { - slot: Default::default(), - _transport: PhantomData, - } - } - - /// The machine this handle's accept/connect handed over, if any. - /// - /// The session makes no progress until the caller polls it. - pub fn take(&self) -> Option> { - self.slot.lock().unwrap().take() - } -} - -impl Default for Inline { - fn default() -> Self { - Self::new() - } -} - -impl Clone for Inline { - fn clone(&self) -> Self { - Self { - slot: self.slot.clone(), - _transport: PhantomData, - } - } -} - -impl std::fmt::Debug for Inline { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("Inline").finish() - } -} - -impl moq_net::Timers for Inline { - type Timer = Timer; - - fn timer(&self) -> Self::Timer { - Timer { at: None, sleep: None } - } - - fn now(&self) -> moq_net::runtime::Instant { - tokio::time::Instant::now().into_std() - } -} - -impl moq_net::Runtime for Inline { - type Transport = S; - - fn spawn(&self, machine: moq_net::runtime::Machine) { - let mut slot = self.slot.lock().unwrap(); - if slot.is_some() { - // Unlock first: panicking under the guard would poison the slot and - // stop the caller who used this handle correctly from ever taking - // their machine, stranding the session this is meant to protect. - drop(slot); - panic!( - "an Inline runtime drives one session: take() the machine from the previous accept/connect, or use a fresh handle" - ); - } - *slot = Some(machine); - } -} - /// A tokio sleep driven through the [`moq_net::runtime::Timer`] contract. pub struct Timer { at: Option, diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 555d4b97dc..bb81160cfd 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -1186,7 +1186,7 @@ fn spawn_stream_request( /// every transport before the caller authorizes. The variant only distinguishes the /// underlying session type; all of them delegate identically. /// A pending moq-net request over transport `S`, driven by our tokio runtime. -type PendingRequest = moq_net::server::Handshake>; +type PendingRequest = moq_net::server::Handshake; pub(crate) enum RequestKind { #[cfg(feature = "noq")] @@ -1456,7 +1456,12 @@ impl Request { /// Accept the session, starting the MoQ session loops. pub async fn ok(self) -> crate::Result { - Ok(request_into!(self.kind, request => request.ok().await?)) + Ok(request_into!(self.kind, request => { + let (session, driver) = request.ok().await?; + use tracing::Instrument; + tokio::spawn(driver.instrument(tracing::Span::current())); + session + })) } /// Returns the network transport carrying this session. diff --git a/rs/moq-uring/README.md b/rs/moq-uring/README.md index 95d68f9f12..7846cbedc6 100644 --- a/rs/moq-uring/README.md +++ b/rs/moq-uring/README.md @@ -29,10 +29,10 @@ the UDP sockets bound through it. - **WebTransport**: browsers negotiate `h3` and `quic::web::Request` runs the HTTP/3 CONNECT handshake (SETTINGS, subprotocol selection, capsule close) over the same adapter via `web-transport-proto`. `quic::web::Session` is - the one transport type the runtime drives, raw or web (`Session::raw`), so - `connect_lite`/`accept_lite` run moq-lite sessions on the worker either - way, with stream and close codes mapped through the HTTP/3 error space in - web mode. + a raw or web transport (`Session::raw`). `connect_lite`/`accept_lite` return + the session and its driver; poll the driver or await it inside a + `Handle::spawn` task to run it on the worker. Web mode maps stream and close codes through the + HTTP/3 error space. - **qlog**: `quic::qlog::Sink` points a group of workers at a directory and `quic::Transport::qlog` turns capture on. The pinned worker never writes to the file: the QUIC stacks want a `Send + Sync` writer, which cannot hold the diff --git a/rs/moq-uring/benches/session_lite.rs b/rs/moq-uring/benches/session_lite.rs index eb8601d625..a1085a1385 100644 --- a/rs/moq-uring/benches/session_lite.rs +++ b/rs/moq-uring/benches/session_lite.rs @@ -172,11 +172,12 @@ mod linux { let conn = quic::server::accept(&server_handle, server_sock, &server_config) .await .expect("quic accept"); - let session = moq_net::Server::new() + let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) .accept_lite(server_handle.clone(), quic::web::Session::raw(conn)) .await .expect("accept_lite"); + let _ = driver.await; session.closed().await; }); @@ -186,11 +187,14 @@ mod linux { let conn = quic::client::connect(&handle, client_sock, &dial) .await .expect("quic connect"); - let session = moq_net::Client::new() + let (session, driver) = moq_net::Client::new() .with_subscriber(sub_origin.clone()) .connect_lite(handle.clone(), quic::web::Session::raw(conn)) .await .expect("connect_lite"); + handle.spawn(async move { + let _ = driver.await; + }); let bc = { let consumer = sub_origin.consume(); consumer.routed("bench").await.expect("broadcast announced"); diff --git a/rs/moq-uring/src/lib.rs b/rs/moq-uring/src/lib.rs index b7627cfc17..e934b57b85 100644 --- a/rs/moq-uring/src/lib.rs +++ b/rs/moq-uring/src/lib.rs @@ -16,8 +16,8 @@ //! 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`] is their -//! [`moq_net::Runtime`]). The stack underneath is the `noq` (default), +//! moq-lite sessions on the worker. [`Handle`] supplies [`moq_net::Timers`]; +//! callers run the returned drivers with [`Handle::spawn`]. The stack underneath is the `noq` (default), //! `quinn`, or `quiche` feature; the module is the same either way, and a build with //! neither leaves it out. //! diff --git a/rs/moq-uring/src/quic/web.rs b/rs/moq-uring/src/quic/web.rs index b9d3381e44..7ce0632dd2 100644 --- a/rs/moq-uring/src/quic/web.rs +++ b/rs/moq-uring/src/quic/web.rs @@ -7,7 +7,7 @@ //! selection) using [`web_transport_proto`]'s state machines, and //! [`Request::respond`] yields a [`Session`]. //! -//! [`Session`] is the one transport type the worker's runtime drives: +//! [`Session`] supports both raw QUIC and WebTransport on the worker: //! [`Session::raw`] wraps a raw-QUIC connection in the same type with the //! WebTransport layering disabled, so native peers and browsers run the same //! machinery. Stream and session error codes map through the HTTP/3 error @@ -419,7 +419,7 @@ struct State { /// A MoQ transport over the worker: raw QUIC, or a WebTransport session. /// -/// The runtime's one transport type. [`Request::respond`] builds the web +/// [`Request::respond`] builds the web /// flavor; [`Session::raw`] wraps a raw-QUIC [`Connection`] with the layering /// disabled. Clones share the session. pub struct Session { @@ -431,7 +431,7 @@ pub struct Session { } impl Session { - /// Wrap a raw-QUIC connection in the runtime's transport type, with no + /// Wrap a raw-QUIC connection in a session, with no /// WebTransport layering: streams and datagrams pass through untouched. pub fn raw(conn: Connection) -> Self { let protocol = web_transport_trait::poll::Session::protocol(&conn).map(str::to_owned); diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index 43c2302da4..44393d92a5 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -544,24 +544,6 @@ impl moq_net::Timers for Handle { } } -// Without a QUIC backend there is no transport to name, so the worker is a -// task and timer runtime only. -#[cfg(any(feature = "noq", feature = "quiche", feature = "quinn"))] -impl moq_net::Runtime for Handle { - type Transport = crate::quic::web::Session; - - fn spawn(&self, machine: moq_net::runtime::Machine) { - // The machine is `!Send` (its transport is), which is exactly what the - // worker's local spawn takes. Its result is the session outcome, which - // the `Session` handle also observes; here it is just the task ending. - self.spawn(async move { - if let Err(err) = machine.await { - tracing::debug!(%err, "session machine ended"); - } - }); - } -} - /// The running kernel release, for error messages. fn kernel_release() -> String { // SAFETY: all-zero is a valid utsname out-buffer. diff --git a/rs/moq-uring/tests/session.rs b/rs/moq-uring/tests/session.rs index 2dd1af2af8..ee0364c813 100644 --- a/rs/moq-uring/tests/session.rs +++ b/rs/moq-uring/tests/session.rs @@ -92,11 +92,12 @@ fn lite_session_over_the_worker() { let conn = quic::server::accept(&server_handle, server_sock, &server_config) .await .expect("quic accept"); - let session = moq_net::Server::new() + let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) .accept_lite(server_handle.clone(), quic::web::Session::raw(conn)) .await .expect("accept_lite"); + let _ = driver.await; // Serve until the client walks away. session.closed().await; }); @@ -113,11 +114,14 @@ fn lite_session_over_the_worker() { "negotiated ALPN" ); - let session = moq_net::Client::new() + let (session, driver) = moq_net::Client::new() .with_subscriber(sub.clone()) .connect_lite(handle.clone(), quic::web::Session::raw(conn)) .await .expect("connect_lite"); + handle.spawn(async move { + let _ = driver.await; + }); let bc = { let consumer = sub.consume(); @@ -209,11 +213,12 @@ fn two_lite_sessions_share_the_server_socket() { let pub_origin = pub_origin.clone(); let session_handle = server_handle.clone(); server_handle.spawn(async move { - let session = moq_net::Server::new() + let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) .accept_lite(session_handle, quic::web::Session::raw(conn)) .await .expect("accept_lite"); + let _ = driver.await; session.closed().await; }); } @@ -233,11 +238,14 @@ fn two_lite_sessions_share_the_server_socket() { let conn = quic::client::connect(&handle, client_sock, &dial) .await .expect("quic connect"); - let session = moq_net::Client::new() + let (session, driver) = moq_net::Client::new() .with_subscriber(sub.clone()) .connect_lite(handle.clone(), quic::web::Session::raw(conn)) .await .expect("connect_lite"); + handle.spawn(async move { + let _ = driver.await; + }); let bc = { let consumer = sub.consume(); diff --git a/rs/moq-uring/tests/web.rs b/rs/moq-uring/tests/web.rs index b8448c3e83..dc01dedde0 100644 --- a/rs/moq-uring/tests/web.rs +++ b/rs/moq-uring/tests/web.rs @@ -235,13 +235,14 @@ fn lite_session_over_webtransport() { Box::pin(async move { assert_eq!(session.protocol(), Some(PROTO), "negotiated subprotocol"); let (sub_origin, sub_driver) = origin::Producer::new(origin::Config::new(moq_net::Hop::random())); - let driver = tokio::spawn(sub_driver.run(moq_tokio::runtime::Runtime::<()>::new())); + let driver = tokio::spawn(sub_driver.run(moq_tokio::runtime::Runtime::new())); - let moq = moq_net::Client::new() + let (moq, session_driver) = moq_net::Client::new() .with_subscriber(sub_origin.clone()) .connect_lite(moq_tokio::runtime::Runtime::new(), session) .await .expect("connect_lite"); + tokio::spawn(session_driver); let bc = { let consumer = sub_origin.consume(); @@ -281,11 +282,14 @@ fn lite_session_over_webtransport() { } let session = request.respond(response).await.expect("respond"); - let session = moq_net::Server::new() + let (session, driver) = moq_net::Server::new() .with_publisher(&serve_origin) .accept_lite(handle.clone(), session) .await .expect("accept_lite"); + handle.spawn(async move { + let _ = driver.await; + }); session.closed().await; }) .expect("worker"); diff --git a/rs/moq-video/src/decode/consumer.rs b/rs/moq-video/src/decode/consumer.rs index 9ef9ffa745..84174ae36d 100644 --- a/rs/moq-video/src/decode/consumer.rs +++ b/rs/moq-video/src/decode/consumer.rs @@ -145,7 +145,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new())); + tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new())); } else { // A sync test: nothing polls the driver, and dropping it would tear // the origin down, so leak it and rely on the synchronous half. diff --git a/rs/moq-wasm/README.md b/rs/moq-wasm/README.md index 12dff5542b..11c4fa1bbd 100644 --- a/rs/moq-wasm/README.md +++ b/rs/moq-wasm/README.md @@ -17,9 +17,9 @@ that path is unrelated to this crate.) What works today: -- **The architecture is right.** `moq-net` is generic over - `web_transport_trait::poll::Session` and spawns via `web_async::spawn` (not - `tokio::spawn`), so it is not tied to native QUIC. +- **Executor-independent sessions.** `moq-net` is generic over + `web_transport_trait::poll::Session` and returns a driver for the caller to + run. `moq-wasm` spawns that driver via `web_async::spawn` on the browser. - **The browser transport needs no adapter**: `web-transport-wasm` implements the poll traits `moq-net` consumes, so `src/transport.rs` is just the dial (the ALPN list and the browser's two trust modes). diff --git a/rs/moq-wasm/src/lib.rs b/rs/moq-wasm/src/lib.rs index f659e38eda..ca9d762bab 100644 --- a/rs/moq-wasm/src/lib.rs +++ b/rs/moq-wasm/src/lib.rs @@ -9,11 +9,8 @@ //! which is the highest-value target (the `@moq/watch` use case). The publish //! path follows the same shape and is left as the obvious next step. //! -//! moq-net's timers and spawning come from the `moq_net::Runtime` passed at -//! connect; `runtime::Runtime` supplies both for the browser (wasmtimer-backed -//! timers, microtask spawning), so the consume path runs in the browser. -//! (`model/time.rs` has an unused wall-clock helper that isn't wasm-portable, -//! but nothing calls it, so it never runs. See README.md.) +//! `runtime::Runtime` supplies browser timers to moq-net. This crate spawns +//! the returned session drivers on the browser's microtask queue. // Browser-only crate. Empty on native so `cargo check --workspace` stays green. #![cfg(target_arch = "wasm32")] @@ -78,10 +75,12 @@ impl Session { web_async::spawn(origin_driver.run(runtime::Runtime)); let consumer = origin.consume(); let client = moq_net::Client::new().with_subscriber(origin); - // The runtime spawns the protocol machine on the microtask queue. The machine - // holds no session clone, so dropping this `Session` still closes the - // transport, which in turn ends the spawned task. - let inner = client.connect(runtime::Runtime, transport).await.map_err(js_err)?; + // The driver holds no session clone, so dropping this `Session` still + // closes the transport and ends the spawned task. + let (inner, driver) = client.connect(runtime::Runtime, transport).await.map_err(js_err)?; + web_async::spawn(async move { + let _ = driver.await; + }); Ok(Session { inner, consumer }) } diff --git a/rs/moq-wasm/src/runtime.rs b/rs/moq-wasm/src/runtime.rs index 00d9847e64..8aec93c9bb 100644 --- a/rs/moq-wasm/src/runtime.rs +++ b/rs/moq-wasm/src/runtime.rs @@ -1,9 +1,8 @@ -//! The browser runtime: microtask spawning and `wasmtimer`-backed timers. +//! Browser-backed timers for MoQ drivers. use std::{pin::Pin, task::Poll}; -/// The [`moq_net::Runtime`] for the browser: machines run on the microtask -/// queue via `web_async::spawn`, timers are `setTimeout`-backed sleeps. +/// Supplies the browser clock and timers to MoQ drivers. #[derive(Clone, Copy, Debug, Default)] pub struct Runtime; @@ -15,18 +14,6 @@ impl moq_net::Timers for Runtime { } } -impl moq_net::Runtime for Runtime { - type Transport = web_transport_wasm::Session; - - fn spawn(&self, machine: moq_net::runtime::Machine) { - web_async::spawn(async move { - // The session surfaces the result through `closed()`; nothing to do - // with it here. - let _ = machine.await; - }); - } -} - /// A `web_async::time::Sleep` driven through the [`moq_net::runtime::Timer`] /// contract. pub struct Timer {