From f5b0b378600bb47cfa76e4e396c5917b256c8bb9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 21 Sep 2026 04:51:14 -0700 Subject: [PATCH 1/2] refactor(net)!: return the next deadline from driver polls `time::Driver::poll(now, waiter)` now returns `Result, Error>`: the next deadline while live, and the terminal error once finished (`Error::Closed` for a clean finish), matching `Session::closed`. The separate `timeout()` accessor and `Output` type are gone. `moq_net::time::run` replaces the identical adapter loops in moq-tokio, moq-wasm, moq-ffi, and the test-runtime feature; moq-uring keeps its own on the worker timer heap. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 5 -- doc/lib/rs/moq-net.md | 13 +++-- rs/moq-e2ee/Cargo.toml | 1 - rs/moq-e2ee/tests/support/harness.rs | 2 +- rs/moq-ffi/src/lib.rs | 2 - rs/moq-ffi/src/origin.rs | 7 +-- rs/moq-ffi/src/runtime.rs | 28 ---------- rs/moq-ffi/src/session.rs | 2 +- rs/moq-hls/src/export/mod.rs | 2 +- rs/moq-hls/src/export/upstream.rs | 2 +- rs/moq-hls/src/server/mod.rs | 2 +- rs/moq-mux/Cargo.toml | 1 - rs/moq-mux/src/source.rs | 2 +- rs/moq-net/Cargo.toml | 7 --- rs/moq-net/src/client.rs | 14 ++--- rs/moq-net/src/driver.rs | 28 ++++------ rs/moq-net/src/lib.rs | 15 ++--- rs/moq-net/src/model/origin.rs | 36 ++++++------ rs/moq-net/src/server.rs | 2 +- rs/moq-net/src/time.rs | 83 ++++++++++++++-------------- rs/moq-net/tests/support/harness.rs | 2 +- rs/moq-relay/src/websocket.rs | 28 +++++----- rs/moq-rtc/src/egress.rs | 2 +- rs/moq-srt/Cargo.toml | 1 - rs/moq-srt/src/dial.rs | 2 +- rs/moq-srt/src/server.rs | 2 +- rs/moq-srt/src/ts.rs | 2 +- rs/moq-stats/Cargo.toml | 1 - rs/moq-stats/src/aggregate.rs | 2 +- rs/moq-stats/src/consume.rs | 2 +- rs/moq-stats/src/produce.rs | 2 +- rs/moq-tokio/src/client.rs | 2 +- rs/moq-tokio/src/lib.rs | 1 - rs/moq-tokio/src/origin.rs | 2 +- rs/moq-tokio/src/runtime.rs | 29 ---------- rs/moq-tokio/src/server.rs | 2 +- rs/moq-uring/benches/session_lite.rs | 2 +- rs/moq-uring/src/worker.rs | 11 ++-- rs/moq-uring/tests/session.rs | 8 +-- rs/moq-uring/tests/web.rs | 6 +- rs/moq-video/Cargo.toml | 1 - rs/moq-video/src/decode/consumer.rs | 2 +- rs/moq-wasm/README.md | 12 ++-- rs/moq-wasm/src/lib.rs | 11 ++-- rs/moq-wasm/src/runtime.rs | 28 ---------- 45 files changed, 155 insertions(+), 262 deletions(-) delete mode 100644 rs/moq-ffi/src/runtime.rs delete mode 100644 rs/moq-tokio/src/runtime.rs delete mode 100644 rs/moq-wasm/src/runtime.rs diff --git a/Cargo.lock b/Cargo.lock index 164a68870f..8cb632f412 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4574,7 +4574,6 @@ dependencies = [ "moq-loc", "moq-msf", "moq-net", - "moq-tokio", "mp4-atom", "mpeg2ts", "num_enum", @@ -4606,7 +4605,6 @@ dependencies = [ "getrandom 0.4.3", "kio", "loom", - "moq-net", "moq-pattern", "num_enum", "rand 0.10.2", @@ -4769,7 +4767,6 @@ dependencies = [ "hang", "moq-mux", "moq-net", - "moq-tokio", "srt-protocol", "srt-tokio", "thiserror 2.0.20", @@ -4784,7 +4781,6 @@ dependencies = [ "futures", "moq-json", "moq-net", - "moq-tokio", "serde", "serde_json", "thiserror 2.0.20", @@ -4935,7 +4931,6 @@ dependencies = [ "moq-mux", "moq-net", "moq-nvenc", - "moq-tokio", "moq-vaapi", "ndk", "objc2 0.6.4", diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 6e7b57e8bb..1406bff752 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -42,18 +42,19 @@ grammar lives on the concept page. `Client::connect(now, transport)`, `Server::accept(now, transport)`, and `server::Handshake::ok()` return `(Session, Driver)`. `moq-net` never spawns tasks or reads the clock: the caller polls the driver and supplies the time. +`moq_net::time::run` does that on tokio or in the browser. ```rust let now = tokio::time::Instant::now().into_std(); let (session, driver) = client.connect(now, transport).await?; -tokio::spawn(moq_tokio::runtime::run(driver)); +tokio::spawn(moq_net::time::run(driver)); ``` A custom event loop calls `driver.poll(now, waiter)` with a nondecreasing -`moq_net::time::Instant`. After `Pending`, wait for external activity or the -instant returned by `driver.timeout()` (`None` means no timer is armed), then -poll again with fresh time. Tests drive the same interface with explicitly -advanced instants. +`moq_net::time::Instant`. `Ok(Some(at))` asks to be polled again by `at` or +on external activity, `Ok(None)` only on external activity, and `Err` is the +terminal error (`Error::Closed` for a clean finish): stop polling. Tests drive +the same interface with explicitly advanced instants. Dropping the last session handle requests closure on the next poll. Dropping the driver cancels the session. `moq-tokio` and `moq-wasm` drive sessions for @@ -61,7 +62,7 @@ their callers. `origin::Producer::new` returns a driver with the same `time::Driver` interface. It calls `cache::Pool::gc(now)` after each poll and folds the next -cleanup time into its own `timeout()`. A standalone pool needs `gc(now)` +cleanup time into its returned deadline. A standalone pool needs `gc(now)` called by its owner, at least by the returned deadline; `None` means expiry is disabled. diff --git a/rs/moq-e2ee/Cargo.toml b/rs/moq-e2ee/Cargo.toml index 4a6ff8c374..051d7fcadd 100644 --- a/rs/moq-e2ee/Cargo.toml +++ b/rs/moq-e2ee/Cargo.toml @@ -25,7 +25,6 @@ zeroize = { workspace = true } [dev-dependencies] criterion = { workspace = true } hex = { workspace = true } -moq-net = { workspace = true, features = ["test-runtime"] } serde_json = { workspace = true } tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] } web-transport-trait = { workspace = true } diff --git a/rs/moq-e2ee/tests/support/harness.rs b/rs/moq-e2ee/tests/support/harness.rs index dac549d653..3eba29225a 100644 --- a/rs/moq-e2ee/tests/support/harness.rs +++ b/rs/moq-e2ee/tests/support/harness.rs @@ -11,7 +11,7 @@ use moq_net::{Client, Server, Session, Version, origin}; use super::mock::create_mock_session_pair; -pub use moq_net::time::test::run; +pub use moq_net::time::run; pub fn now() -> moq_net::time::Instant { tokio::time::Instant::now().into_std() diff --git a/rs/moq-ffi/src/lib.rs b/rs/moq-ffi/src/lib.rs index c8c806c56b..03820bc813 100644 --- a/rs/moq-ffi/src/lib.rs +++ b/rs/moq-ffi/src/lib.rs @@ -26,8 +26,6 @@ mod log; pub mod media; pub mod origin; pub mod producer; -#[cfg(target_arch = "wasm32")] -mod runtime; #[cfg(not(target_arch = "wasm32"))] pub mod server; pub mod session; diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 1ea84675a0..12ffb67491 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -206,10 +206,9 @@ impl MoqOriginProducer { /// Build an origin producer, spawning its driver on the FFI runtime. 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(moq_tokio::runtime::run(driver)); - #[cfg(target_arch = "wasm32")] - crate::ffi::spawn(crate::runtime::run(driver)); + crate::ffi::spawn(async move { + moq_net::time::run(driver).await; + }); producer } diff --git a/rs/moq-ffi/src/runtime.rs b/rs/moq-ffi/src/runtime.rs deleted file mode 100644 index 6bc9252224..0000000000 --- a/rs/moq-ffi/src/runtime.rs +++ /dev/null @@ -1,28 +0,0 @@ -//! Run MoQ drivers with the runtime clock and timer. - -use std::task::Poll; - -/// Run an explicit-time driver on the runtime's clock and timer. -pub async fn run(mut driver: D) -> D::Output { - let mut timer = None; - moq_net::kio::wait(|waiter| { - loop { - let now = web_async::time::Instant::now(); - if let Poll::Ready(result) = driver.poll(now, waiter) { - return Poll::Ready(result); - } - let Some(at) = driver.timeout() else { - timer = None; - return Poll::Pending; - }; - let sleep = timer.get_or_insert_with(|| Box::pin(web_async::time::sleep_until(at))); - if sleep.deadline() != at { - sleep.as_mut().reset(at); - } - if waiter.poll_future(sleep.as_mut()).is_pending() { - return Poll::Pending; - } - } - }) - .await -} diff --git a/rs/moq-ffi/src/session.rs b/rs/moq-ffi/src/session.rs index 75c513b911..fda4430810 100644 --- a/rs/moq-ffi/src/session.rs +++ b/rs/moq-ffi/src/session.rs @@ -301,7 +301,7 @@ impl Client { .await?; crate::ffi::spawn(async move { - let _ = crate::runtime::run(driver).await; + moq_net::time::run(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 5fb94d44e9..7edf278c69 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -362,7 +362,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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 1ae02aec59..025bae2b40 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::origin::Config::default()); - let driver = tokio::spawn(moq_tokio::runtime::run(driver)); + let driver = tokio::spawn(moq_net::time::run(driver)); 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 a2e158cf82..c4a13be663 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::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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/Cargo.toml b/rs/moq-mux/Cargo.toml index 65b64d6566..7b1549ba48 100644 --- a/rs/moq-mux/Cargo.toml +++ b/rs/moq-mux/Cargo.toml @@ -42,5 +42,4 @@ webm-iterable = "0.7" [dev-dependencies] futures = { workspace = true } -moq-tokio = { workspace = true } tokio = { workspace = true, features = ["test-util"] } diff --git a/rs/moq-mux/src/source.rs b/rs/moq-mux/src/source.rs index a5fd11a05f..8a2c66e220 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::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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/Cargo.toml b/rs/moq-net/Cargo.toml index 5f6b2db4f4..51ce7330db 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -16,10 +16,6 @@ categories = ["multimedia", "network-programming", "web-programming"] ignored = ["getrandom"] [features] -# The deterministic runtime::Test, for tests in this crate and downstream ones. -# A feature rather than cfg(test) because integration tests and other crates' -# tests compile against the normal library build. -test-runtime = [] # Exposes the wire codecs to the `fuzz/` harness through the hidden `fuzz` module. # Off by default and hidden from the docs; nothing depends on it but that harness. fuzz = [] @@ -41,9 +37,6 @@ web-transport-trait = { workspace = true } [dev-dependencies] criterion = { workspace = true } -# Enable the test runtime for our own unit and integration tests. A crate may -# dev-depend on itself to turn features on for its test builds only. -moq-net = { path = ".", features = ["test-runtime"] } serde_json = { workspace = true } # test-util (tokio::time::pause/advance) is test-only and is NOT supported on # wasm, so it must not leak into the normal dependency feature set. diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 1ef2ae0776..dbb9c197f5 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -682,7 +682,7 @@ mod tests { .connect(tokio::time::Instant::now().into_std(), fake.clone()) .await .unwrap(); - tokio::spawn(crate::time::test::run(driver)); + tokio::spawn(crate::time::run(driver)); // Verify the client setup was encoded using Draft14 framing (ALPN_LITE fallback path). let mut setup_bytes = Bytes::from(fake.control_writes()); @@ -759,7 +759,7 @@ mod tests { assert_eq!(session.version(), Version::Lite(lite::Version::Lite04)); // Construction leaves the driver idle until the caller polls it. - assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_pending()); + assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok()); // The caller drops their only session clone; the machine observes the // last handle going away and closes the transport. @@ -982,7 +982,7 @@ mod tests { let runtime = crate::runtime::Test::new(); let (session, mut driver) = futures::executor::block_on(client.connect_lite(runtime.now(), local)).unwrap(); - assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_pending()); + assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok()); fn assert_send_sync(_: &T) {} assert_send_sync(&session); @@ -1009,7 +1009,7 @@ mod tests { let runtime = crate::runtime::Test::new(); let (session, mut driver) = futures::executor::block_on(server.accept_lite(runtime.now(), local)).unwrap(); assert_eq!(session.version(), Version::Lite(lite::Version::Lite04)); - assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_pending()); + assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok()); drop(session); assert!(fake.state.close_events.lock().unwrap().is_empty()); @@ -1048,7 +1048,7 @@ mod tests { .connect(tokio::time::Instant::now().into_std(), fake.clone()) .await .unwrap(); - tokio::spawn(crate::time::test::run(driver)); + tokio::spawn(crate::time::run(driver)); // The construction-time snapshot, before the machine sampled anything. assert_eq!( @@ -1081,7 +1081,7 @@ mod tests { .connect(tokio::time::Instant::now().into_std(), fake.clone()) .await .unwrap(); - tokio::spawn(crate::time::test::run(driver)); + tokio::spawn(crate::time::run(driver)); assert!( session.send_bandwidth().is_none(), "no send-rate estimate, so nothing samples on its own" @@ -1112,7 +1112,7 @@ mod tests { .connect(tokio::time::Instant::now().into_std(), fake.clone()) .await .unwrap(); - tokio::spawn(crate::time::test::run(driver)); + tokio::spawn(crate::time::run(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 index 23f357e11d..d00f74d785 100644 --- a/rs/moq-net/src/driver.rs +++ b/rs/moq-net/src/driver.rs @@ -8,9 +8,10 @@ use crate::time::{Clock, Instant}; /// Drives a session with caller-supplied time. /// /// Returned by [`crate::Client::connect`] and [`crate::Server::accept`]. Call -/// [`poll`](Self::poll) when external activity wakes the waiter or when -/// [`timeout`](Self::timeout) is reached. Supply nondecreasing instants. -/// Completion is cached, so subsequent polls return the same result. +/// [`poll`](Self::poll) when external activity wakes the waiter or when the +/// returned deadline is reached, supplying nondecreasing instants; see +/// [`crate::time::Driver`] for the contract. Completion is cached, so +/// subsequent polls return the same error. /// /// It holds no session handle: dropping the last [`crate::Session`] requests /// closure on the next poll. Dropping the driver cancels the session, and @@ -59,9 +60,13 @@ impl Driver { /// Process ready work at `now`, registering for external activity. /// /// Panics if `now` is earlier than the previous poll or construction time. - pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Poll> { + pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error> { self.clock.advance(now); - self.state.poll(waiter) + match self.state.poll(waiter) { + Poll::Ready(Ok(())) => Err(Error::Closed), + Poll::Ready(Err(err)) => Err(err), + Poll::Pending => Ok(self.clock.timeout()), + } } } @@ -103,20 +108,9 @@ impl State { } impl crate::time::Driver for Driver { - type Output = Result<(), Error>; - fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Poll { + fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error> { self.poll(now, waiter) } - fn timeout(&self) -> Option { - self.timeout() - } -} - -impl Driver { - /// The next instant the caller must poll at, or none while no timer is armed. - pub fn timeout(&self) -> Option { - self.clock.timeout() - } } impl std::fmt::Debug for Driver { diff --git a/rs/moq-net/src/lib.rs b/rs/moq-net/src/lib.rs index a4ddb6e0db..d1b755dbbb 100644 --- a/rs/moq-net/src/lib.rs +++ b/rs/moq-net/src/lib.rs @@ -47,11 +47,12 @@ //! last producer signals consumers that no more updates are coming. //! //! ## Driving and time -//! This library never spawns tasks or schedules runtime timers. [`Client::connect`] -//! and [`Server::accept`] take an initial [`time::Instant`] and return +//! This library never spawns tasks or reads the clock. [`Client::connect`] and +//! [`Server::accept`] take an initial [`time::Instant`] and return //! `(Session, Driver)`. Poll the [`Driver`] with the current instant and a -//! [`kio::Waiter`], then wake on external activity or at [`Driver::timeout`]. -//! The last [`Session`] drop requests closure; dropping the driver cancels it. +//! [`kio::Waiter`], then wake on external activity or at the deadline it +//! returns; [`time::run`] does exactly that on tokio or the browser. The last +//! [`Session`] drop requests closure; dropping the driver cancels it. //! //! [`origin::Producer::new`] also returns a producer and driver. Its driver runs //! route changes, serving, linger, teardown, and the origin's cache expiration. @@ -59,9 +60,9 @@ //! clear their expiration timestamp for the next cleanup pass. Datagrams use a bounded //! FIFO; model read/write APIs take no wall-clock time. //! -//! Both drivers implement [`time::Driver`]. `moq-tokio` and `moq-wasm` -//! supply runtime adapters, and `moq-uring` can drive thread-local transports. -//! Tests can advance time simply by supplying a later instant. +//! Both drivers implement [`time::Driver`]. `moq-uring` drives thread-local +//! transports on its own timer heap. Tests advance time by supplying a later +//! instant. #![warn(missing_docs)] // The browser transport is `!Send`, so on wasm the shared state behind these `Arc`s is diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 7ba2e9b607..134dff9d25 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -1121,7 +1121,6 @@ impl Producer { }, timers, pool, - gc: None, }; (producer, driver) } @@ -1614,8 +1613,8 @@ impl Drop for AnnounceProducer { /// Drives origin lifecycle work and cache expiration with caller-supplied time. /// -/// Returned by [`Producer::new`]. Poll on external activity or at [`Self::timeout`], -/// supplying nondecreasing instants. Route changes, track serving, linger, +/// Returned by [`Producer::new`]. Poll on external activity or at the deadline +/// it returns, supplying nondecreasing instants. Route changes, track serving, linger, /// failover, and teardown run here; exact lookups and eligible announcements /// update synchronously in [`Producer::create_broadcast`]. /// @@ -1631,7 +1630,6 @@ pub struct Driver { // The cache pool this origin's groups charge into, swept on a wall-clock // cadence so its idle window binds a track whose publisher stopped writing. pool: cache::Pool, - gc: Option, } /// Lifecycle work and the state it tears down. @@ -1649,27 +1647,25 @@ struct DriverState { impl Driver { /// Process ready origin work using caller-supplied monotonic time. - pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Poll<()> { + /// + /// See [`crate::time::Driver`] for the contract. Finishes with + /// [`Error::Closed`] once every producer handle has dropped and the + /// remaining lifecycle work has drained. + pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error> { self.timers.advance(now); let result = self.state.poll(waiter); - self.gc = self.pool.gc(now); - result - } - - /// The next instant to poll, or none when only external activity can make progress. - pub fn timeout(&self) -> Option { - self.timers.timeout().into_iter().chain(self.gc).min() + let gc = self.pool.gc(now); + if result.is_ready() { + return Err(Error::Closed); + } + Ok(self.timers.timeout().into_iter().chain(gc).min()) } } impl crate::time::Driver for Driver { - type Output = (); - fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Poll<()> { + fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error> { self.poll(now, waiter) } - fn timeout(&self) -> Option { - self.timeout() - } } impl DriverState { @@ -3964,7 +3960,9 @@ 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(crate::time::test::run(driver)); + web_async::spawn(async move { + crate::time::run(driver).await; + }); } 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. @@ -5210,7 +5208,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 = crate::time::test::run(driver); + let run = crate::time::run(driver); drop(producer); tokio::time::timeout(Duration::from_secs(5), run) .await diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 32db6a5f05..9ca4de32fc 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -1108,7 +1108,7 @@ mod tests { .unwrap(); let (session, driver) = request.ok().await.unwrap(); - tokio::spawn(crate::time::test::run(driver)); + tokio::spawn(crate::time::run(driver)); for _ in 0..100 { if occurrences(&log, b"local-route") > 0 { diff --git a/rs/moq-net/src/time.rs b/rs/moq-net/src/time.rs index cfcb74fcda..bef0ba27b4 100644 --- a/rs/moq-net/src/time.rs +++ b/rs/moq-net/src/time.rs @@ -1,5 +1,6 @@ //! Caller-supplied time for protocol and model drivers. +use crate::Error; use crate::runtime::{Timer as _, Timers}; use std::{ collections::BTreeMap, @@ -10,13 +11,48 @@ use std::{ pub use crate::runtime::Instant; /// A state machine polled with caller-supplied time. +/// +/// Each poll advances the driver to `now`, processes ready work, and registers +/// `waiter` for external activity. `Ok(Some(at))` asks to be polled again by +/// `at` (or sooner, on a wake); `Ok(None)` means only external activity can +/// make progress. `Err` is terminal: the driver has finished and must not be +/// polled again. A clean finish is [`Error::Closed`]. pub trait Driver { - /// The result produced when this driver finishes. - type Output; - /// Advance time and process ready work, registering for external activity. - fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Poll; - /// The next time to poll, or none when only external activity can make progress. - fn timeout(&self) -> Option; + /// Advance to `now` and process ready work. + fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error>; +} + +/// Run a driver to completion on the ambient runtime, sleeping until each deadline. +/// +/// Tokio on native, `setTimeout` in the browser. Resolves with the driver's +/// terminal error, [`Error::Closed`] for a clean finish. +pub async fn run(mut driver: D) -> Error { + let mut timer: Option>> = None; + kio::wait(|waiter| { + loop { + let now = web_async::time::Instant::now(); + #[cfg(not(target_family = "wasm"))] + let now = now.into_std(); + let at = match driver.poll(now, waiter) { + Ok(Some(at)) => at, + Ok(None) => { + timer = None; + return Poll::Pending; + } + Err(err) => return Poll::Ready(err), + }; + #[cfg(not(target_family = "wasm"))] + let at = web_async::time::Instant::from_std(at); + let sleep = timer.get_or_insert_with(|| Box::pin(web_async::time::sleep_until(at))); + if sleep.deadline() != at { + sleep.as_mut().reset(at); + } + if waiter.poll_future(sleep.as_mut()).is_pending() { + return Poll::Pending; + } + } + }) + .await } /// A private clock shared only by work owned by one driver. @@ -180,38 +216,3 @@ mod tests { clock.advance(now - Duration::from_secs(1)); } } - -/// Runtime adapter for integration tests. -#[cfg(any(test, feature = "test-runtime"))] -#[doc(hidden)] -pub mod test { - use std::task::Poll; - /// Run an explicit-time driver on the runtime's clock and timer. - pub async fn run(mut driver: D) -> D::Output { - let mut timer = None; - crate::kio::wait(|waiter| { - loop { - let now = web_async::time::Instant::now(); - #[cfg(not(target_family = "wasm"))] - let now = now.into_std(); - if let Poll::Ready(result) = driver.poll(now, waiter) { - return Poll::Ready(result); - } - let Some(at) = driver.timeout() else { - timer = None; - return Poll::Pending; - }; - #[cfg(not(target_family = "wasm"))] - let at = web_async::time::Instant::from_std(at); - let sleep = timer.get_or_insert_with(|| Box::pin(web_async::time::sleep_until(at))); - if sleep.deadline() != at { - sleep.as_mut().reset(at); - } - if waiter.poll_future(sleep.as_mut()).is_pending() { - return Poll::Pending; - } - } - }) - .await - } -} diff --git a/rs/moq-net/tests/support/harness.rs b/rs/moq-net/tests/support/harness.rs index dac549d653..3eba29225a 100644 --- a/rs/moq-net/tests/support/harness.rs +++ b/rs/moq-net/tests/support/harness.rs @@ -11,7 +11,7 @@ use moq_net::{Client, Server, Session, Version, origin}; use super::mock::create_mock_session_pair; -pub use moq_net::time::test::run; +pub use moq_net::time::run; pub fn now() -> moq_net::time::Instant { tokio::time::Instant::now().into_std() diff --git a/rs/moq-relay/src/websocket.rs b/rs/moq-relay/src/websocket.rs index 006bb37162..07199da29f 100644 --- a/rs/moq-relay/src/websocket.rs +++ b/rs/moq-relay/src/websocket.rs @@ -183,7 +183,7 @@ where ) .await?; - let driver = moq_tokio::runtime::run(driver); + let driver = moq_net::time::run(driver); tokio::pin!(driver); // The handshake is done, so this is a MoQ session now: only now can a push @@ -198,21 +198,15 @@ where } }; tokio::select! { - res = &mut driver => { - lease.close( - match &res { - Ok(()) => "closed".to_string(), - Err(err) => err.to_string(), - }, - crate::connection::session_bytes(&session), - ); - return res.map_err(Into::into); + err = &mut driver => { + lease.close(err.to_string(), crate::connection::session_bytes(&session)); + return ended(err); } why = lease.ended() => { tracing::info!(%why, "lease ended, closing session"); session.abort(moq_net::Error::Unauthorized); // Drive the teardown so the close reaches the peer. - let res = driver.await.map_err(Into::into); + let res = ended(driver.await); lease.close(why, crate::connection::session_bytes(&session)); return res; } @@ -224,8 +218,8 @@ where let drain = shutdown.drain_session(&session); let mut drain = std::pin::pin!(drain); let res = tokio::select! { - res = &mut driver => res.map_err(Into::into), - _ = &mut drain => driver.await.map_err(Into::into), + err = &mut driver => ended(err), + _ = &mut drain => ended(driver.await), }; lease.close("shutdown", crate::connection::session_bytes(&session)); return res; @@ -250,6 +244,14 @@ where /// /// A client that offers no subprotocol at all is left alone: it upgrades and /// negotiates the moq version over moq-lite SETUP instead. +/// The driver's terminal error as a session outcome: a clean close is not a failure. +fn ended(err: moq_net::Error) -> anyhow::Result<()> { + match err { + moq_net::Error::Closed => Ok(()), + err => Err(err.into()), + } +} + fn negotiate_subprotocol(ws: WebSocketUpgrade, alpns: &[&str]) -> Result { let supported = supported_subprotocols(alpns); diff --git a/rs/moq-rtc/src/egress.rs b/rs/moq-rtc/src/egress.rs index bec8cad0f8..8f91f43131 100644 --- a/rs/moq-rtc/src/egress.rs +++ b/rs/moq-rtc/src/egress.rs @@ -298,7 +298,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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/Cargo.toml b/rs/moq-srt/Cargo.toml index 98e89fee91..5b85caef0f 100644 --- a/rs/moq-srt/Cargo.toml +++ b/rs/moq-srt/Cargo.toml @@ -29,5 +29,4 @@ tracing = { workspace = true } [dev-dependencies] # Builds the synthetic broadcasts the egress tests mux back to TS. hang = { workspace = true } -moq-tokio = { workspace = true } srt-protocol = "0.4" diff --git a/rs/moq-srt/src/dial.rs b/rs/moq-srt/src/dial.rs index 0d6d27e34b..1b7a94b187 100644 --- a/rs/moq-srt/src/dial.rs +++ b/rs/moq-srt/src/dial.rs @@ -146,7 +146,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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 35759aa90b..7894acb35d 100644 --- a/rs/moq-srt/src/server.rs +++ b/rs/moq-srt/src/server.rs @@ -835,7 +835,7 @@ mod tests { const OFFSET: u64 = 600_000_000; let (origin, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); let mut broadcast = origin.create_broadcast("rewind").unwrap(); broadcast.announce(moq_net::origin::Route::default()).unwrap(); let mut catalog = moq_mux::catalog::Producer::new( diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index bcd42025cc..baafbb3ef1 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -145,7 +145,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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/Cargo.toml b/rs/moq-stats/Cargo.toml index a088897a48..078baaa5e0 100644 --- a/rs/moq-stats/Cargo.toml +++ b/rs/moq-stats/Cargo.toml @@ -23,5 +23,4 @@ web-async = { workspace = true } [dev-dependencies] futures = { workspace = true } -moq-tokio = { workspace = true } tokio = { workspace = true, features = ["macros", "rt", "test-util", "time"] } diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index d9b0b786d6..8fdccb8088 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::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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 80b33ddbd1..2dcf2b0566 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::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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 4720731514..519d241b9f 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -883,7 +883,7 @@ mod tests { fn produce_origin() -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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 d738909827..04358722bb 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -602,7 +602,7 @@ async fn connect_session( .connect(tokio::time::Instant::now().into_std(), transport) .await?; use tracing::Instrument; - tokio::spawn(crate::runtime::run(driver).instrument(tracing::Span::current())); + tokio::spawn(moq_net::time::run(driver).instrument(tracing::Span::current())); Ok(session) } diff --git a/rs/moq-tokio/src/lib.rs b/rs/moq-tokio/src/lib.rs index 3fd5cd762c..f520b86565 100644 --- a/rs/moq-tokio/src/lib.rs +++ b/rs/moq-tokio/src/lib.rs @@ -46,7 +46,6 @@ pub mod origin; pub mod quic; #[cfg(any(feature = "noq", feature = "tcp", feature = "websocket"))] mod resolve; -pub mod runtime; #[cfg(feature = "_transport")] pub mod server; #[doc(hidden)] diff --git a/rs/moq-tokio/src/origin.rs b/rs/moq-tokio/src/origin.rs index 2c1fe78f2e..aa4398b62b 100644 --- a/rs/moq-tokio/src/origin.rs +++ b/rs/moq-tokio/src/origin.rs @@ -17,7 +17,7 @@ pub fn spawn() -> moq_net::origin::Producer { /// Build and spawn an origin producer with an explicit configuration. pub fn spawn_config(config: moq_net::origin::Config) -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(config); - tokio::spawn(crate::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); producer } diff --git a/rs/moq-tokio/src/runtime.rs b/rs/moq-tokio/src/runtime.rs deleted file mode 100644 index f8e268918c..0000000000 --- a/rs/moq-tokio/src/runtime.rs +++ /dev/null @@ -1,29 +0,0 @@ -//! Run MoQ drivers with the runtime clock and timer. - -use std::task::Poll; - -/// Run an explicit-time driver on the runtime's clock and timer. -pub async fn run(mut driver: D) -> D::Output { - let mut timer = None; - moq_net::kio::wait(|waiter| { - loop { - let now = tokio::time::Instant::now().into_std(); - if let Poll::Ready(result) = driver.poll(now, waiter) { - return Poll::Ready(result); - } - let Some(at) = driver.timeout() else { - timer = None; - return Poll::Pending; - }; - let at = tokio::time::Instant::from_std(at); - let sleep = timer.get_or_insert_with(|| Box::pin(tokio::time::sleep_until(at))); - if sleep.deadline() != at { - sleep.as_mut().reset(at); - } - if waiter.poll_future(sleep.as_mut()).is_pending() { - return Poll::Pending; - } - } - }) - .await -} diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 28aa32bf19..75cd4b43d4 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -1303,7 +1303,7 @@ impl Request { Ok(request_into!(self.kind, request => { let (session, driver) = request.ok().await?; use tracing::Instrument; - tokio::spawn(crate::runtime::run(driver).instrument(tracing::Span::current())); + tokio::spawn(moq_net::time::run(driver).instrument(tracing::Span::current())); session })) } diff --git a/rs/moq-uring/benches/session_lite.rs b/rs/moq-uring/benches/session_lite.rs index c276e337d1..df792a776f 100644 --- a/rs/moq-uring/benches/session_lite.rs +++ b/rs/moq-uring/benches/session_lite.rs @@ -99,7 +99,7 @@ mod linux { .build() .expect("tokio runtime"); rt.block_on(async move { - tokio::join!(moq_tokio::runtime::run(pub_driver), moq_tokio::runtime::run(sub_driver)); + tokio::join!(moq_net::time::run(pub_driver), moq_net::time::run(sub_driver)); }); }); diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index c1ea3b7501..1294accbcb 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -542,15 +542,16 @@ impl Handle { crate::Timer::from_heap(self.shared.timers.clone()) } - /// Run a MoQ driver with this worker's timer and monotonic clock. - pub async fn run(&self, mut driver: D) -> D::Output { + /// Run a MoQ driver with this worker's timer and monotonic clock, resolving + /// with its terminal error. + pub async fn run(&self, mut driver: D) -> moq_net::Error { let mut timer = self.timer(); kio::wait(|waiter| { loop { - if let Poll::Ready(result) = driver.poll(Instant::now(), waiter) { - return Poll::Ready(result); + match driver.poll(Instant::now(), waiter) { + Ok(at) => timer.set(at), + Err(err) => return Poll::Ready(err), } - timer.set(driver.timeout()); if timer.poll(waiter).is_pending() { return Poll::Pending; } diff --git a/rs/moq-uring/tests/session.rs b/rs/moq-uring/tests/session.rs index 2700a4e578..3e8c8af321 100644 --- a/rs/moq-uring/tests/session.rs +++ b/rs/moq-uring/tests/session.rs @@ -46,7 +46,7 @@ fn lite_session_over_the_worker() { .build() .expect("tokio runtime"); rt.block_on(async move { - tokio::join!(moq_tokio::runtime::run(pub_driver), moq_tokio::runtime::run(sub_driver)); + tokio::join!(moq_net::time::run(pub_driver), moq_net::time::run(sub_driver)); }); }); @@ -167,9 +167,9 @@ fn two_lite_sessions_share_the_server_socket() { .expect("tokio runtime"); rt.block_on(async move { tokio::join!( - moq_tokio::runtime::run(pub_driver), - moq_tokio::runtime::run(sub_a_driver), - moq_tokio::runtime::run(sub_b_driver), + moq_net::time::run(pub_driver), + moq_net::time::run(sub_a_driver), + moq_net::time::run(sub_b_driver), ); }); }); diff --git a/rs/moq-uring/tests/web.rs b/rs/moq-uring/tests/web.rs index 0f68a358d2..a4624039a5 100644 --- a/rs/moq-uring/tests/web.rs +++ b/rs/moq-uring/tests/web.rs @@ -216,7 +216,7 @@ fn lite_session_over_webtransport() { .enable_time() .build() .expect("tokio runtime"); - rt.block_on(moq_tokio::runtime::run(pub_driver)); + rt.block_on(moq_net::time::run(pub_driver)); }); let broadcast = pub_origin.create_broadcast("test").expect("create broadcast"); @@ -235,14 +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::default()); - let driver = tokio::spawn(moq_tokio::runtime::run(sub_driver)); + let driver = tokio::spawn(moq_net::time::run(sub_driver)); let (moq, session_driver) = moq_net::Client::new() .with_subscriber(sub_origin.clone()) .connect_lite(std::time::Instant::now(), moq_tokio::transport::Session::new(session)) .await .expect("connect_lite"); - tokio::spawn(moq_tokio::runtime::run(session_driver)); + tokio::spawn(moq_net::time::run(session_driver)); let bc = { let consumer = sub_origin.consume(); diff --git a/rs/moq-video/Cargo.toml b/rs/moq-video/Cargo.toml index 8505de0e02..55f232d88f 100644 --- a/rs/moq-video/Cargo.toml +++ b/rs/moq-video/Cargo.toml @@ -236,7 +236,6 @@ windows = { version = "0.62", features = [ # `poll!`, to cancel an `encode::Sink` request exactly where a `select!` would. futures = { workspace = true } h264-reader = "0.8.0" -moq-tokio = { workspace = true } # Driving `encode::Sink` from a plain test thread, the way an FFI caller does. pollster = { workspace = true } tracing-test = { workspace = true } diff --git a/rs/moq-video/src/decode/consumer.rs b/rs/moq-video/src/decode/consumer.rs index aa4a67bc45..dd7d50834c 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::origin::Config::default()); if tokio::runtime::Handle::try_current().is_ok() { - tokio::spawn(moq_tokio::runtime::run(driver)); + tokio::spawn(moq_net::time::run(driver)); } 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 6276089dd4..e195d32b69 100644 --- a/rs/moq-wasm/README.md +++ b/rs/moq-wasm/README.md @@ -19,8 +19,8 @@ What works today: - **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 a runtime adapter via `web_async::spawn`, supplying - time and scheduling the driver's next wakeup on the browser. + run. `moq-wasm` spawns `moq_net::time::run` via `web_async::spawn`, which + supplies the browser clock and sleeps until the driver's next deadline. - **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). @@ -45,11 +45,11 @@ What works today: transport is `!Send`, but `SessionInner` used to hard-code `Send`. `web_async::MaybeSendBoxFuture` picks a `Send` boxed future on native and a local boxed future on wasm. Native behavior is unchanged. -3. Drivers accept `moq_net::time::Instant` and report their next timeout. - This crate supplies the browser clock and one re-armable sleep through - `web_async::time`, backed by `performance.now()` and `setTimeout`. +3. Drivers accept `moq_net::time::Instant` and return their next deadline. + `moq_net::time::run` supplies the clock and one re-armable sleep through + `web_async::time`, backed by `performance.now()` and `setTimeout` here. `moq-net` handles session sampling, linger, probes, and cache expiration - using the supplied time. It does not create runtime timers or spawn tasks. + using the supplied time and never spawns tasks. ### Timestamp fallback diff --git a/rs/moq-wasm/src/lib.rs b/rs/moq-wasm/src/lib.rs index b69051e465..9d2a392f05 100644 --- a/rs/moq-wasm/src/lib.rs +++ b/rs/moq-wasm/src/lib.rs @@ -9,8 +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. //! -//! `runtime::run` drives moq-net with the browser clock and timer. This crate spawns -//! the returned session drivers on the browser's microtask queue. +//! `moq_net::time::run` drives moq-net with the browser clock and timer; this +//! crate spawns it on the browser's microtask queue. // Browser-only crate. Empty on native so `cargo check --workspace` stays green. #![cfg(target_arch = "wasm32")] @@ -21,7 +21,6 @@ use std::rc::Rc; use js_sys::Uint8Array; use wasm_bindgen::prelude::*; -pub mod runtime; pub mod transport; /// Map any displayable error into a JS exception. @@ -72,7 +71,9 @@ impl Session { // Wire a subscribe origin so the session has somewhere to insert the // broadcasts the remote announces; keep a consumer to read them. let (origin, origin_driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); - web_async::spawn(crate::runtime::run(origin_driver)); + web_async::spawn(async move { + moq_net::time::run(origin_driver).await; + }); let consumer = origin.consume(); let client = moq_net::Client::new().with_subscriber(origin); // The driver holds no session clone, so dropping this `Session` still @@ -82,7 +83,7 @@ impl Session { .await .map_err(js_err)?; web_async::spawn(async move { - let _ = crate::runtime::run(driver).await; + moq_net::time::run(driver).await; }); Ok(Session { inner, consumer }) } diff --git a/rs/moq-wasm/src/runtime.rs b/rs/moq-wasm/src/runtime.rs deleted file mode 100644 index 6bc9252224..0000000000 --- a/rs/moq-wasm/src/runtime.rs +++ /dev/null @@ -1,28 +0,0 @@ -//! Run MoQ drivers with the runtime clock and timer. - -use std::task::Poll; - -/// Run an explicit-time driver on the runtime's clock and timer. -pub async fn run(mut driver: D) -> D::Output { - let mut timer = None; - moq_net::kio::wait(|waiter| { - loop { - let now = web_async::time::Instant::now(); - if let Poll::Ready(result) = driver.poll(now, waiter) { - return Poll::Ready(result); - } - let Some(at) = driver.timeout() else { - timer = None; - return Poll::Pending; - }; - let sleep = timer.get_or_insert_with(|| Box::pin(web_async::time::sleep_until(at))); - if sleep.deadline() != at { - sleep.as_mut().reset(at); - } - if waiter.poll_future(sleep.as_mut()).is_pending() { - return Poll::Pending; - } - } - }) - .await -} From 1ee85ea17f4a6daa2575f183a76b7ae354d4cdc2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 21 Sep 2026 05:07:03 -0700 Subject: [PATCH 2/2] refactor(net): use kio::Lock instead of web_async::Lock Leaves web-async serving only the browser clock in moq-net, and puts these locks under loom's model. kio::Lock gains the Default impl the swap needs. Co-Authored-By: Claude Opus 5 --- rs/kio/src/lock.rs | 6 ++++++ rs/moq-net/src/ietf/subscriber.rs | 2 +- rs/moq-net/src/lite/subscriber.rs | 2 +- rs/moq-net/src/model/broadcast.rs | 4 ++-- rs/moq-net/src/model/origin.rs | 6 ++---- rs/moq-net/src/model/resume.rs | 4 ++-- rs/moq-net/src/stats.rs | 2 +- rs/moq-net/tests/loom.rs | 8 ++++---- rs/moq-wasm/README.md | 7 +++---- 9 files changed, 22 insertions(+), 19 deletions(-) diff --git a/rs/kio/src/lock.rs b/rs/kio/src/lock.rs index f031021d9d..eb36f84b43 100644 --- a/rs/kio/src/lock.rs +++ b/rs/kio/src/lock.rs @@ -95,6 +95,12 @@ impl Clone for Lock { } } +impl Default for Lock { + fn default() -> Self { + Self::new(T::default()) + } +} + impl fmt::Debug for Lock { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self.inner.try_lock() { diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 99c5832b82..67a7b9baee 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -16,7 +16,7 @@ use crate::{ use super::{Message, Version, cluster, error::request, peer}; -use web_async::Lock; +use kio::Lock; const TRACK_ALIAS_TIMEOUT: Duration = Duration::from_secs(1); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 331b0758d6..c8bbadf0cc 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -16,7 +16,7 @@ use crate::{ use super::Version; -use web_async::Lock; +use kio::Lock; pub(super) struct SubscriberConfig { pub runtime: crate::time::Clock, diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index bf863efaa4..0897a446cb 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -512,7 +512,7 @@ struct Alive { // The advertisement of the broadcast's exact path, owned here so it retracts // with the broadcast: on finish, abort, or the last producer-side handle // dropping. `None` for a standalone broadcast. - announcer: web_async::Lock>, + announcer: kio::Lock>, } impl Alive { @@ -520,7 +520,7 @@ impl Alive { Arc::new(Self { token: kio::Producer::default(), state, - announcer: web_async::Lock::new(None), + announcer: kio::Lock::new(None), }) } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 134dff9d25..46db5febe1 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -10,8 +10,8 @@ use std::{ time::Duration, }; +use kio::Lock; use rand::RngExt; -use web_async::Lock; use super::{Requests, WeakCache, WeakEntry}; use crate::{ @@ -3960,9 +3960,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(async move { - crate::time::run(driver).await; - }); + tokio::spawn(crate::time::run(driver)); } 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/model/resume.rs b/rs/moq-net/src/model/resume.rs index 02c455cb1c..0123e60cc2 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -583,7 +583,7 @@ impl Consumer { state: self.state.clone(), sequence, options: options.into().unwrap_or_default(), - inner: web_async::Lock::new(None), + inner: kio::Lock::new(None), }) } @@ -651,7 +651,7 @@ pub struct Fetching { // break the type recursion with `track::Fetching` (which can wrap a resume // [`Fetching`]). #[allow(clippy::type_complexity)] - inner: web_async::Lock)>>, + inner: kio::Lock)>>, } impl Fetching { diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index ca93bf6acb..ed3e5fd8a7 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -113,8 +113,8 @@ use std::{ }, }; +use kio::Lock; use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use web_async::Lock; use crate::{AsPath, PathOwned, Pattern, Patterns}; diff --git a/rs/moq-net/tests/loom.rs b/rs/moq-net/tests/loom.rs index e55c21160e..bc4e81825a 100644 --- a/rs/moq-net/tests/loom.rs +++ b/rs/moq-net/tests/loom.rs @@ -11,10 +11,10 @@ //! asserted directly. //! //! Know the boundary before trusting a pass: only kio's primitives are swapped for -//! loom's (see `kio/src/sync.rs`). moq-net's own `web_async::Lock`s and bare -//! `std::sync::atomic` counters, like the ones in `model/cache.rs`, still execute -//! under loom (its threads are cooperative) but loom does not permute around them. -//! So these models cover the kio handoff every handle is built on, not every +//! loom's (see `kio/src/sync.rs`). moq-net's bare `std::sync::atomic` counters and +//! `std::sync::Mutex`es, like the ones in `model/cache.rs`, still execute under +//! loom (its threads are cooperative) but loom does not permute around them. So +//! these models cover the kio handoff every handle is built on, not every //! critical section moq-net owns. //! //! Run with `just rs loom`; the whole file compiles away without `cfg(loom)`. diff --git a/rs/moq-wasm/README.md b/rs/moq-wasm/README.md index e195d32b69..7a0609c710 100644 --- a/rs/moq-wasm/README.md +++ b/rs/moq-wasm/README.md @@ -53,10 +53,9 @@ What works today: ### Timestamp fallback -`model/time.rs` uses `web_async::time::{Instant, SystemTime}` for timestamp -generation. Native keeps the Tokio-backed instant so paused-time tests still -work; browser wasm uses wasmtimer-backed clocks, avoiding the `std::time` and -Tokio paths that panic or lack a driver on `wasm32-unknown-unknown`. +`moq_net::time::Instant` is `std::time::Instant` on native and the +wasmtimer-backed instant from `web_async::time` in the browser, where +`std::time::Instant` panics. `model/time.rs` anchors its timestamps on it. ### Out of scope here: moq-mux