diff --git a/quest/m1/README.md b/quest/m1/README.md index f99627545fe..48dc84854d1 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -129,7 +129,6 @@ tracks, so per-group stats stop costing cluster announces. - [Track live signal](/quest/m1/track-live.md) - a reader can wait for its track's feed to be live, so a player skips old cached groups while a fresh upstream subscription starts - [Producers drop the phantom catalog type](/quest/m1/catalog-phantom.md) - moq-video, moq-audio and moq-mux producers lose an `E: CatalogExt` parameter that only `PhantomData` holds - [JSON encoder options without a downcast](/quest/m1/json-encoder-options.md) - moq-mux's `json_snapshot` takes its `deltas` switch through a typed path any custom catalog entry can use, not an `Any` downcast -- [moq-net compiles once](/quest/m1/net-erase-transport.md) - the session drivers stop monomorphizing per transport and per crate, saving several MiB of the relay, FFI library, `libmoq` and CLI - [Audio devices on ARM](/quest/m1/audio-devices-arm.md) - `just rs audio-devices` passes on the hosted ARM runner - [Tooling tests gate every merge](/quest/m1/gh-test-gate.md) - every queue Check runs `just gh test`, so a change to the CI selector or release tooling cannot merge without its own tests - [Wayland display id](/quest/m1/wayland-display-id.md) - a requested display id off X11 is refused or honored, never silently replaced by a portal pick shared across publications diff --git a/quest/m1/audit-compile-time.md b/quest/m1/audit-compile-time.md index aecb99b11f8..bc391e9d8d1 100644 --- a/quest/m1/audit-compile-time.md +++ b/quest/m1/audit-compile-time.md @@ -24,8 +24,8 @@ incremental). The [CI workflow audit](/quest/m2/audit-ci.md) owns job-level CI cost and caching; the [dependency audit](/quest/m2/audit-deps.md) owns deduplicating and upgrading dependencies. -What to check first: whether [moq-net compiles once](/quest/m1/net-erase-transport.md) -landed. If so, rerun `cargo llvm-lines` on the moq-tokio lib and compare the +What to check first: [moq-net compiles once](https://github.com/moq-dev/moq/pull/6176) +landed, so rerun `cargo llvm-lines` on the moq-tokio lib and compare the transport share below, then rerun the thin-LTO size comparison against the smaller archive. Then whether `cargo-llvm-lines` joined the pinned shell. @@ -137,7 +137,7 @@ Not measured: thin sizes for linux_arm64, darwin_arm64 and windows_amd64; build time above two jobs; wall-clock runtime. Decision (maintainer, 2026-10-20): keep fat LTO, and revisit once -[moq-net compiles once](/quest/m1/net-erase-transport.md) shrinks the archive. +[moq-net compiles once](https://github.com/moq-dev/moq/pull/6176) shrinks the archive. Thin saves about 40% of the final unit's wall time on two jobs, but release builds run only nightly and on release PRs, off the merge-queue gate. It costs 10 MiB of the Go archive's headroom under a hard publish limit while that @@ -146,7 +146,7 @@ builds join the PR gate. ### Results of the m0 run -- Planned: [moq-net compiles once](/quest/m1/net-erase-transport.md) removes +- Planned: [moq-net compiles once](https://github.com/moq-dev/moq/pull/6176) removes the per-transport copies, the largest attributable dev IR (43% of moq-tokio) and a share of the release fat-LTO module. - Explained, no action: the release `moq` unit is all LLVM backend work on one diff --git a/quest/m1/audit-size.md b/quest/m1/audit-size.md index 7d916912263..99c59f1d168 100644 --- a/quest/m1/audit-size.md +++ b/quest/m1/audit-size.md @@ -19,9 +19,9 @@ Only a change that needs a maintainer decision (public API, scope) or is independently landable and too large to review alongside the rest becomes a follow-up through `/quest-plan`. -Checkpointed 2026-10-21, a day after the m0 run, with nothing to plan but -the quests below. The run finishes after -[moq-net compiles once](/quest/m1/net-erase-transport.md) lands, so the audit +Checkpointed 2026-10-21, a day after the m0 run, with nothing to plan. The +run waited for moq-net to compile once, which landed in +[moq-dev/moq#6176](https://github.com/moq-dev/moq/pull/6176), so the audit before the breaking cut sees its result and the rest of m1. What to check first: whether the nightly `sizes` job (`sh/size/`) has run. @@ -30,7 +30,7 @@ The Linux release-link fix 2026-10-22 UTC; verify the first scheduled Nightly after it before treating the sizes history as available. Once it runs, its history replaces the local method below. Then -re-attribute the native rows for the erasure quest, and measure `@moq/net` +re-attribute the native rows after the erasure, and measure `@moq/net` with the track compression codec in `js/net`. ### Baselines @@ -66,13 +66,14 @@ run's FFI (31.86 MiB) and CLI (48.14 MiB) figures for it do not reproduce. from the message parameter codec (`parameters.ts`, `filter.ts`), so a lite session never loads them. First load fell 5.1 to 5.5 KB gzip in every bundle; the same test fails if SETUP reaches `parameters.ts` again. -- Planned: [moq-net compiles once](/quest/m1/net-erase-transport.md). On - 7d2a91fe66, `moq-net` is 9.5 MiB of the relay's text and 7.7 MiB of the FFI - library's; one copy of each function would be about 2.5 MiB in both, so the - ceiling is about 7.0 and 5.2 MiB. In the relay, the binary crate (a 16-line - `main.rs`) instantiates 3.1 MiB of it, `moq-tokio` 3.2 MiB and the library - 1.9 MiB; identical instantiations repeated across crates are 5.0 MiB. They - are not byte-identical after LTO: `-Wl,--icf=all` saves only 0.23 MiB. +- Fixed: moq-net boxes the transport, so its session drivers compile once + instead of per transport and per instantiating crate + ([moq-dev/moq#6176](https://github.com/moq-dev/moq/pull/6176)). Stripped + profiling builds fell 31.8 to 22.5 MiB (relay), 49.1 to 42.5 MiB (CLI), + 28.0 to 21.8 MiB (`libmoq_ffi.so`), 254.6 to 207.8 MiB (unstripped + `libmoq.a`) and 862 to 786 KB gzipped (`@moq/wasm`). Before it, the relay + binary crate, `moq-tokio` and the library each instantiated their own copy; + `-Wl,--icf=all` saved only 0.23 MiB, since the copies differ after LTO. - Planned: [the catalog publish path compiles once](/quest/m2/catalog-publish-once.md) (`moq-json`'s diff serializer, +0.14 MiB of the CLI this run). - Explained, no action: native growth since 2b28f5eac is `moq-net` features @@ -102,10 +103,6 @@ v0 mangling suffix) on local `profiling` builds of both revisions, the wasm name section from a `wasm-release` build with `CARGO_PROFILE_WASM_RELEASE_STRIP=none`, and esbuild metafiles from `sh/size/bundles.ts` for JS. -## Required - -- [moq-net compiles once](/quest/m1/net-erase-transport.md) - re-measure the native rows once the drivers compile once - ## Related - [FFI size profile](/quest/m2/ffi-size-profile.md) - whether the bindings ship at `opt-level = "s"` diff --git a/quest/m1/net-erase-transport.md b/quest/m1/net-erase-transport.md deleted file mode 100644 index cdc800d255b..00000000000 --- a/quest/m1/net-erase-transport.md +++ /dev/null @@ -1,46 +0,0 @@ -# [L] moq-net compiles its protocol code once - -## Goal - -`moq-net`'s session drivers compile once per binary, not once per transport -session type and per crate that drives them. The relay, `libmoq_ffi.so`, -`libmoq`, the CLI and `@moq/wasm` shrink by the duplicated copies, they -compile faster for the same reason, and the relay fan-out benchmark shows no -regression beyond its noise. - -## Plan - -Found by the 2026-10 size audit (main at -2b28f5eac). The drivers are generic over the session, so each backend -(QUIC/WebTransport, qmux) gets a full copy, the IETF `ControlStreamAdapter` -doubles the IETF half, and every crate that awaits a `moq-tokio` future or -calls `moq_net::Server`/`Client` with a concrete session codegens its own copy: -release builds share no generics across crates. `moq-net` is 9.1 of 23.5 MiB of -Rust text in the relay (eight copies), 6.4 of 19.2 MiB in the FFI library and -7.4 of 36.5 MiB in the CLI. Estimated saving: 5 to 6 MiB of the relay, 4 to -5 MiB of the FFI library, similar for `libmoq` and the CLI. - -Open decision (public API): `Driver`, `server::Handshake` and friends -carry the session type. - -1. Drop the type parameter: the public types take a boxed session trait object - at their constructor. Breaking, and the API gets simpler. -2. Keep the generic signatures as thin entry points that erase at once. - Non-breaking, but it is a compatibility shim the repo rules refuse. -3. Leave it, accepting the size. - -Recommendation: 1, if the dynamic dispatch prices out. Measure it first with -`bench/relay.sh` on `bench/workloads/fanout-heavy.toml` and the `moq-net` -session benchmark, before and after. The async bodies must end up -non-generic inside `moq-net`, or each awaiting crate still instantiates them. -Re-measure with the audit's method: `nm` symbol attribution by crate path and -by instantiating crate on an unstripped `profiling` build. - -Compile time (2026-10-20 compile-time audit, -[#6106](https://github.com/moq-dev/moq/pull/6106)): moq-tokio's dev build is -1,661,503 lines of LLVM IR, 43.2% of it moq-net session code instantiated per -transport and again under `ControlStreamAdapter`. The PR reports -`cargo llvm-lines` for moq-tokio and moq-relay before and after, which is -deterministic, and a clean build of both crates before and after, run back to -back and reported as a ratio with the load average. Fewer IR lines alone do not -prove a faster build if the work moves to another compiler phase. diff --git a/quest/m1/net-internals.md b/quest/m1/net-internals.md index 60f8e866413..96a85cc7e2f 100644 --- a/quest/m1/net-internals.md +++ b/quest/m1/net-internals.md @@ -25,7 +25,3 @@ Dropped: renaming the over-long `session_delivery_*` bench IDs, since it loses their history for a cosmetic gain. Public API: none. Wire: none. - -## Required - -- [moq-net compiles once](/quest/m1/net-erase-transport.md) - boxes the transport these cleanups build on diff --git a/quest/m2/ietf-adapter-session-errors.md b/quest/m2/ietf-adapter-session-errors.md index 597ca50a207..971be369437 100644 --- a/quest/m2/ietf-adapter-session-errors.md +++ b/quest/m2/ietf-adapter-session-errors.md @@ -29,7 +29,3 @@ ProtocolViolation while a uni or datagram poll is pending, and the client must see a terminal error and not redial. Public API: none. Wire: none. - -## Related - -- [moq-net compiles once](/quest/m1/net-erase-transport.md) - its open PR [moq-dev/moq#6176](https://github.com/moq-dev/moq/pull/6176) touches the same adapter diff --git a/quest/m2/uring-transport-machines.md b/quest/m2/uring-transport-machines.md index ca93b95acd5..f35208644e1 100644 --- a/quest/m2/uring-transport-machines.md +++ b/quest/m2/uring-transport-machines.md @@ -3,8 +3,8 @@ ## Goal The relay fanout benchmark measures what `moq_uring::transport::Local`'s -per-call thread check costs on the ring once -[moq-net compiles once](/quest/m1/net-erase-transport.md) lands, and the +per-call thread check costs on the ring now that +[moq-net compiles once](https://github.com/moq-dev/moq/pull/6176), and the check is removed only if it shows up. `transport::poll::Boxable` stays the single bound. @@ -24,10 +24,6 @@ check, and remove it only if the difference shows. Public API: none unless the check goes, which removes `moq_uring::transport::Local`. Wire: none. -## Required - -- [moq-net compiles once](/quest/m1/net-erase-transport.md) - boxes the transport the check wraps - ## Related - [Relay io_uring packages](/quest/m2/relay-io-uring-package.md) - not blocked on this; `Local` already gives parity, and only its cost is measured here diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index c39554a5641..a16f363a660 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -781,18 +781,12 @@ mod tests { .with_subscriber(to.clone()); tokio::join!( async { - let (session, driver) = client - .connect_lite(moq_net::time::now(), client_transport) - .await - .unwrap(); + let (session, driver) = client.connect(moq_net::time::now(), client_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session }, async { - let (session, driver) = server - .accept_lite(moq_net::time::now(), server_transport) - .await - .unwrap(); + let (session, driver) = server.accept(moq_net::time::now(), server_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session } diff --git a/rs/moq-mux/src/container/test_util.rs b/rs/moq-mux/src/container/test_util.rs index f0bb717a357..48155d9f4e7 100644 --- a/rs/moq-mux/src/container/test_util.rs +++ b/rs/moq-mux/src/container/test_util.rs @@ -259,18 +259,12 @@ impl Link { // Each driver runs as soon as its handshake resolves: the accept waits on the client's SETUP. let sessions = tokio::join!( async { - let (session, driver) = client - .connect_lite(moq_net::time::now(), client_transport) - .await - .unwrap(); + let (session, driver) = client.connect(moq_net::time::now(), client_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session }, async { - let (session, driver) = server - .accept_lite(moq_net::time::now(), server_transport) - .await - .unwrap(); + let (session, driver) = server.accept(moq_net::time::now(), server_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session } @@ -304,18 +298,12 @@ impl Link { .with_publisher(from.consume()); let sessions = tokio::join!( async { - let (session, driver) = client - .connect_lite(moq_net::time::now(), client_transport) - .await - .unwrap(); + let (session, driver) = client.connect(moq_net::time::now(), client_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session }, async { - let (session, driver) = server - .accept_lite(moq_net::time::now(), server_transport) - .await - .unwrap(); + let (session, driver) = server.accept(moq_net::time::now(), server_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session } diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index cbd8d10165a..5b362199db7 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,14 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Sessions box their transport (#6176) + +`Driver` and `server::Handshake` lose their transport type parameter: write `moq_net::Driver` and `moq_net::server::Handshake`. +`Client::connect_lite`, `Server::accept_lite`, and `Server::accept_request_lite` are gone; call `connect`, `accept`, or `accept_request`, which accept every ALPN. +Every session now takes a `transport::poll::Boxable` transport, whose streams must also be `Sync` on native. +A `!Send` transport wraps itself in a type that checks the thread on use, such as `moq_uring::transport::Local`. +`Client::connect` and `Server::accept_request` are plain functions returning `impl Future`, so call them as before with `.await`. + ### `track::Producer::largest()` is a getter `track::Producer::largest(&self)` returns the track's largest location now, `None` before any, instead of streaming each newer one; `track::Producer::poll_largest` is gone. diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 3e31d37bec7..90c851323c7 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -1,5 +1,7 @@ +use crate::driver::Transport; use crate::origin; use crate::time::{Deadlines, Instant}; +use crate::transport::poll::Session as _; use crate::{ ALPN_14, ALPN_15, ALPN_16, ALPN_17, ALPN_18, ALPN_19, ALPN_20, ALPN_21, ALPN_22, ALPN_LITE, ALPN_LITE_03, ALPN_LITE_04, ALPN_LITE_05, ALPN_LITE_06, ALPN_LITE_07_WIP, Consume, Error, NEGOTIATED, Session, Version, Versions, @@ -193,16 +195,13 @@ impl Client { /// Start a lite session on an already-negotiated version: build our SETUP, /// wire the origins, and return the session and its driver. - fn start_lite( + fn start_lite( &self, runtime: Deadlines, - session: S, + session: Transport, version: lite::Version, peer_hop: crate::Hop, - ) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Session, - { + ) -> Result<(Session, crate::Driver), Error> { let (publish, subscribe) = self.origins(peer_hop); // Advertise our capabilities (we report what the transport measures; we @@ -256,39 +255,19 @@ impl Client { )) } - /// Perform the MoQ handshake for moq-lite only, over any transport. - /// - /// Unlike [`connect`](Self::connect) this puts no thread-affinity bound on - /// the transport, so a pinned `!Send` transport works and yields a `!Send` - /// machine that stays on its thread. The trade is protocol scope: only a - /// moq-lite ALPN is accepted, since the moq-transport driver still needs a - /// [`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, now: Instant, session: S) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Session, - { - let runtime = Deadlines::new(now); - let version = match session.protocol() { - Some(ALPN_LITE_07_WIP) => lite::Version::Lite07, - Some(ALPN_LITE_06) => lite::Version::Lite06, - Some(ALPN_LITE_05) => lite::Version::Lite05, - Some(ALPN_LITE_04) => lite::Version::Lite04, - Some(ALPN_LITE_03) => lite::Version::Lite03, - _ => return Err(Error::Version), - }; - self.versions.select(Version::Lite(version)).ok_or(Error::Version)?; - self.start_lite(runtime, session, version, self.assigned_hop()) - } - /// Perform the MoQ handshake, returning the [`Session`] and its [`Driver`](crate::Driver). /// /// Poll the returned driver with nondecreasing time, starting at `now`. - pub async fn connect(&self, now: Instant, mut session: S) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Boxable, - { + pub fn connect( + &self, + now: Instant, + session: impl crate::transport::poll::Boxable, + ) -> impl Future> { + // Box before any protocol code runs, so it is compiled once rather than per transport. + self.connect_boxed(now, Transport::new(session)) + } + + async fn connect_boxed(&self, now: Instant, mut session: Transport) -> Result<(Session, crate::Driver), Error> { let runtime = Deadlines::new(now); let peer_hop = self.assigned_hop(); let (publish, subscribe) = self.origins(peer_hop); @@ -1000,233 +979,6 @@ mod tests { assert!(matches!(futures::executor::block_on(session.closed()), Error::Cancel)); } - /// A transport made deliberately `!Send` by an `Rc` marker on the session and - /// both stream types: compiling at all is the point, proving the lite path - /// never demands thread mobility of any transport piece. - #[derive(Clone)] - struct LocalSession { - inner: FakeSession, - _local: std::rc::Rc<()>, - } - - struct LocalSend { - inner: FakeSendStream, - _local: std::rc::Rc<()>, - } - - struct LocalRecv { - inner: FakeRecvStream, - _local: std::rc::Rc<()>, - } - - impl crate::transport::poll::Session for LocalSession { - type SendStream = LocalSend; - type RecvStream = LocalRecv; - type Error = FakeError; - - fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { - self.inner.poll_accept_uni(cx).map_ok(|stream| LocalRecv { - inner: stream, - _local: self._local.clone(), - }) - } - - fn poll_accept_bi( - &mut self, - cx: &mut Context<'_>, - ) -> Poll> { - self.inner.poll_accept_bi(cx).map_ok(|(send, recv)| { - ( - LocalSend { - inner: send, - _local: self._local.clone(), - }, - LocalRecv { - inner: recv, - _local: self._local.clone(), - }, - ) - }) - } - - fn poll_open_bi( - &mut self, - cx: &mut Context<'_>, - options: crate::transport::poll::SendOptions<'_, crate::transport::poll::Ungrouped>, - ) -> Poll> { - self.inner.poll_open_bi(cx, options).map_ok(|(send, recv)| { - ( - LocalSend { - inner: send, - _local: self._local.clone(), - }, - LocalRecv { - inner: recv, - _local: self._local.clone(), - }, - ) - }) - } - - fn poll_open_uni( - &mut self, - cx: &mut Context<'_>, - options: crate::transport::poll::SendOptions<'_, crate::transport::poll::Ungrouped>, - ) -> Poll> { - self.inner.poll_open_uni(cx, options).map_ok(|stream| LocalSend { - inner: stream, - _local: self._local.clone(), - }) - } - - fn send_group(&mut self, _: i64) -> Option { - None - } - - fn send_datagram( - &mut self, - payload: Bytes, - options: crate::transport::poll::SendOptions<'_, crate::transport::poll::Ungrouped>, - ) -> Result<(), Self::Error> { - self.inner.send_datagram(payload, options) - } - - fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { - self.inner.poll_recv_datagram(cx) - } - - fn max_datagram_size(&self) -> usize { - self.inner.max_datagram_size() - } - - fn binding(&self) -> crate::transport::Binding { - crate::transport::Binding::InBand - } - fn protocol(&self) -> Option<&str> { - self.inner.protocol() - } - - fn close(&mut self, code: u32, reason: &str) { - self.inner.close(code, reason); - } - - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll { - self.inner.poll_closed(cx) - } - - fn stats(&self) -> impl crate::transport::Stats { - self.inner.stats() - } - } - - impl crate::transport::poll::SendStream for LocalSend { - type Error = FakeError; - type Group = crate::transport::poll::Ungrouped; - - fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { - self.inner.poll_write(cx, buf) - } - - fn set_priority(&mut self, order: i64) { - self.inner.set_priority(order); - } - - fn set_group(&mut self, group: &Self::Group) { - match *group {} - } - - fn finish(&mut self) -> Result<(), Self::Error> { - self.inner.finish() - } - - fn reset(&mut self, code: u32) { - self.inner.reset(code); - } - - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { - crate::transport::poll::SendStream::poll_closed(&mut self.inner, cx) - } - } - - impl crate::transport::poll::RecvStream for LocalRecv { - type Error = FakeError; - type Lease<'a> = crate::transport::poll::BufferedLease<'a>; - - fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Self::Error>> { - self.inner.poll_lease(cx, want) - } - - fn set_priority(&mut self, priority: i64) { - self.inner.set_priority(priority); - } - - fn stop(&mut self, code: u32) { - crate::transport::poll::RecvStream::stop(&mut self.inner, code); - } - } - - // The point of the lite-only entry: a `!Send` transport yields a `!Send` - // 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] - fn connect_lite_over_a_send_less_transport() { - let fake = FakeSession::new(Some(ALPN_LITE_04), Vec::new()); - let local = LocalSession { - inner: fake.clone(), - _local: std::rc::Rc::new(()), - }; - let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - - let runtime = crate::time::Deadlines::new(crate::time::Instant::now()); - let (session, mut driver) = futures::executor::block_on(client.connect_lite(runtime.now(), local)).unwrap(); - assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok()); - - fn assert_send_sync(_: &T) {} - assert_send_sync(&session); - - session.abort(Error::Cancel); - let _ = driver.poll(runtime.now(), &kio::Waiter::noop()); - assert_eq!(fake.state.close_events.read()[0].0, SessionError::Cancel.to_code()); - } - - // The server-side twin: a `!Send` transport accepts a lite session whose - // driver the caller polls directly. - #[test] - fn accept_lite_over_a_send_less_transport() { - let fake = FakeSession::new(Some(ALPN_LITE_04), Vec::new()); - let local = LocalSession { - inner: fake.clone(), - _local: std::rc::Rc::new(()), - }; - let server = crate::Server::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - - let runtime = crate::time::Deadlines::new(crate::time::Instant::now()); - 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_ok()); - - drop(session); - assert!(fake.state.close_events.read().is_empty()); - let _ = driver.poll(runtime.now(), &kio::Waiter::noop()); - assert_eq!(fake.state.close_events.read()[0].0, SessionError::Cancel.to_code()); - } - - // The lite-only entry refuses everything that still needs the boxed ietf - // driver, instead of silently negotiating it. - #[test] - fn connect_lite_refuses_ietf_alpns() { - let fake = FakeSession::new(Some(ALPN_19), Vec::new()); - let local = LocalSession { - inner: fake, - _local: std::rc::Rc::new(()), - }; - let client = Client::new(); - let runtime = crate::time::Deadlines::new(crate::time::Instant::now()); - let result = futures::executor::block_on(client.connect_lite(runtime.now(), local)); - assert!(matches!(result, Err(Error::Version))); - } - // `stats()` reads the machine's latest sample and primes the sampler, so a // periodic poller observes fresh counters without consuming the bandwidth // channel. diff --git a/rs/moq-net/src/driver.rs b/rs/moq-net/src/driver.rs index d560f799e80..f32f4c3ae9c 100644 --- a/rs/moq-net/src/driver.rs +++ b/rs/moq-net/src/driver.rs @@ -15,31 +15,27 @@ use crate::time::{Deadlines, Instant}; /// /// 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. +/// [`crate::Session::closed`] resolves with [`Error::Cancel`]. It is `Send` on +/// native, like the [`Boxable`](crate::transport::poll::Boxable) transport it drives. #[must_use = "the session makes no progress unless its driver is polled"] -pub struct Driver { - state: State, +pub struct Driver { + state: State, clock: Deadlines, } /// 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>), +pub(crate) enum Protocol { + /// Boxed for size only. + Lite(Box>), Ietf(crate::ietf::Driver), } +/// The transport every session runs on, boxed once by `Client` or `Server`. +pub(crate) type Transport = crate::transport::boxed::Session; + /// Protocol and lifecycle work owned by the driver. -pub(crate) struct State { - pub(crate) protocol: Protocol, +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 @@ -47,13 +43,13 @@ pub(crate) struct State { // 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>, + 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(clock: Deadlines, state: State) -> Self { +impl Driver { + pub(crate) fn new(clock: Deadlines, state: State) -> Self { Self { state, clock } } @@ -72,7 +68,7 @@ impl Driver { } } -impl Protocol { +impl Protocol { pub(crate) fn local_close(&self) -> std::sync::Arc { match self { Self::Lite(driver) => driver.local_close.clone(), @@ -104,7 +100,7 @@ impl Protocol { } } -impl State { +impl State { fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { if let Some(supervisor) = &mut self.supervisor && supervisor.poll(waiter).is_ready() @@ -145,16 +141,23 @@ impl State { } } -impl crate::time::Driver for Driver { +impl crate::time::Driver for Driver { fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error> { self.poll(now, waiter) } } -impl std::fmt::Debug for Driver { +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() } } + +// The relay and the bindings spawn drivers on multi-threaded runtimes. +#[cfg(not(target_family = "wasm"))] +const _: () = { + const fn assert_send() {} + assert_send::(); +}; diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index ca2e15e11d3..8e99324e2a1 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1,4 +1,5 @@ use crate::origin; +use crate::transport::poll::Session as _; use crate::{ Error, Hop, SessionError, StreamError, coding::{Decode, DecodeError, Encode, Reader, Stream, Writer}, @@ -112,10 +113,7 @@ impl std::future::Future for Driver { } } -pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> -where - S: crate::transport::poll::Boxable, -{ +pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> { let Config { runtime, stats, @@ -233,11 +231,14 @@ where }; let control = Control::new(request_id_max, client).with_window(limits.requests(), client, version); let adapter = ControlStreamAdapter::new(session.clone(), control.clone(), version).with_stats(&stats); + // Boxed like a draft-17+ transport, so the publisher and subscriber below + // are the same instantiation as theirs. + let virtual_session = crate::driver::Transport::new(adapter.clone()); // No AUTH on these drafts, but a limit still reaches both halves. let mut publisher = Publisher::new( runtime.clone(), - adapter.clone(), + virtual_session.clone(), publish, control.clone(), peer_hop, @@ -253,7 +254,7 @@ where let mut subscriber = Subscriber::new( runtime.clone(), - adapter.clone(), + virtual_session.clone(), subscribe, control, peer_hop, @@ -293,7 +294,7 @@ where } drop(tasks); - let dispatch_session = adapter.clone(); + let dispatch_session = virtual_session.clone(); let sub_ns = subscriber.clone(); let sub_ns_adapter = adapter.clone(); @@ -301,7 +302,7 @@ where // completion); the task set draining is the one clean exit. let mut adapter_run = std::pin::pin!(err_only(adapter.run(setup.reader, setup.writer, goaway.clone()))); let mut unis = Unis::new( - adapter.clone(), + virtual_session.clone(), subscriber.clone(), UniSetup { peer: None, @@ -320,7 +321,8 @@ where None, version ))); - let mut datagrams = std::pin::pin!(err_only(run_datagrams(adapter.clone(), subscriber.clone()))); + let mut datagrams = + std::pin::pin!(err_only(run_datagrams(virtual_session.clone(), subscriber.clone()))); // Unsolicited PUBLISH_NAMESPACE unless the peer requires solicitation; // see `Publisher::run_publish_namespaces`. let mut pub_ns_run = std::pin::pin!(err_only(publisher.clone().run_publish_namespaces())); @@ -334,14 +336,20 @@ where Version::Draft16 => { let mut sub_ns_adapter = sub_ns_adapter; let (send, recv) = sub_ns_adapter.open_native_bi().await?; - let mut reader = crate::coding::Reader::new(recv, version); + let mut reader = crate::coding::Reader::new( + crate::transport::boxed::RecvStream::new(recv), + version, + ); reader.set_priority(crate::priority::RECV_CONTROL); Stream { - writer: crate::coding::Writer::new(send, version), + writer: crate::coding::Writer::new( + crate::transport::boxed::SendStream::new(send), + version, + ), reader, } } - _ => Stream::open(&mut sub_ns_adapter.clone(), version).await?, + _ => Stream::open(&mut crate::driver::Transport::new(sub_ns_adapter), version).await?, }; if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix, snapshot).await { // The peer breaking the protocol is fatal, and the driver @@ -1610,7 +1618,7 @@ mod tests { let (driver, _) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1666,7 +1674,7 @@ mod tests { let (driver, _) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1713,7 +1721,7 @@ mod tests { let (driver, _) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1809,7 +1817,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1873,7 +1881,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1938,8 +1946,9 @@ mod tests { }; let gate = kio::Producer::new(true); - let mut session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); + let session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); let log = session.log.clone(); + let mut session = crate::driver::Transport::new(session); let setup = Stream::open(&mut session, version) .await .expect("open the control stream"); @@ -2036,7 +2045,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2140,7 +2149,7 @@ mod tests { let (driver, goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2225,7 +2234,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2322,7 +2331,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2412,7 +2421,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2456,7 +2465,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2723,7 +2732,7 @@ mod tests { start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2867,7 +2876,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2936,7 +2945,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, client: false, @@ -3029,9 +3038,10 @@ mod tests { let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let consumer = origin.consume(); - let mut session = + let session = crate::lite::test_transport::ScriptedSession::new(publish_namespace_then_two_withdrawals(VERSION).await); let log = session.log.clone(); + let mut session = crate::driver::Transport::new(session); let setup = Stream::open(&mut session, VERSION) .await .expect("open the control stream"); @@ -3147,13 +3157,14 @@ mod tests { ), ] { let payload = subscribe_with_token(version, token); - let mut session = match version { + let session = match version { Version::Draft14 => crate::lite::test_transport::ScriptedSession::new(payload), _ => { crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(vec![payload]) } }; let log = session.log.clone(); + let mut session = crate::driver::Transport::new(session); let setup = match version { Version::Draft14 => Some(Stream::open(&mut session, version).await.unwrap()), _ => None, @@ -3394,7 +3405,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -3497,7 +3508,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index a76eae490e8..27f25956092 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -1,10 +1,10 @@ //! Accepting a MoQ session, including the paused handshake that inspects the //! peer's SETUP before granting origins. -use crate::transport::{MaybeSend, MaybeSync}; - +use crate::driver::Transport; use crate::origin; use crate::time::{Deadlines, Instant}; +use crate::transport::poll::Session as _; use crate::{ ALPN_14, ALPN_15, ALPN_16, ALPN_17, ALPN_18, ALPN_19, ALPN_20, ALPN_21, ALPN_22, ALPN_LITE, ALPN_LITE_03, ALPN_LITE_04, ALPN_LITE_05, ALPN_LITE_06, ALPN_LITE_07_WIP, Consume, Error, NEGOTIATED, Role, Session, @@ -143,18 +143,15 @@ impl Server { /// Start a lite session on an accepted transport: wire the origins, answer /// with our SETUP, and return the session and its driver. - fn start_lite( + fn start_lite( &self, runtime: Deadlines, - session: S, + session: Transport, version: lite::Version, - client_setup: Option>, + client_setup: Option>, peer_hop: Option, auth: crate::auth::Handle, - ) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Session, - { + ) -> Result<(Session, crate::Driver), Error> { let (publish, subscribe) = self.stat_tagged_origins(); // We report what the transport actually measures; a server never @@ -206,41 +203,7 @@ impl Server { )) } - /// Perform the MoQ handshake for moq-lite only, over any transport. - /// - /// Same trade as [`Client::connect_lite`](crate::Client::connect_lite): no - /// thread-affinity bound on the transport, so a pinned `!Send` transport - /// works, and only a moq-lite ALPN is accepted (anything else is refused - /// 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, now: Instant, session: S) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Session, - { - self.accept_request_lite(now, session).await?.ok().await - } - - /// Begin the moq-lite handshake, pausing like - /// [`accept_request`](Self::accept_request) but for moq-lite ALPNs only, - /// 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, now: Instant, session: S) -> Result, Error> - where - S: crate::transport::poll::Session, - { - let mut refused = session.clone(); - self.handshake_lite(now, session) - .await - .inspect_err(|err| close(&mut refused, err, None)) - } - - async fn handshake_lite(&self, now: Instant, mut session: S) -> Result, Error> - where - S: crate::transport::poll::Session, - { - let runtime = Deadlines::new(now); + async fn handshake_lite(&self, runtime: Deadlines, mut session: Transport) -> Result { let (path, role, origin, handshake) = match session.protocol() { Some(alpn @ (ALPN_LITE_05 | ALPN_LITE_06 | ALPN_LITE_07_WIP)) => { let version = match alpn { @@ -300,7 +263,7 @@ impl Server { PausedHandshake::LiteBare { version, .. } | PausedHandshake::LiteSetup { version, .. } => { version.has_auth() } - PausedHandshake::Boxed(_) => false, + PausedHandshake::IetfModern(_) | PausedHandshake::Legacy(_) => false, }); Ok(Handshake { path, @@ -325,12 +288,11 @@ impl Server { /// 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, now: Instant, session: S) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, - { + pub async fn accept( + &self, + now: Instant, + session: impl crate::transport::poll::Boxable, + ) -> Result<(Session, crate::Driver), Error> { self.accept_request(now, session).await?.ok().await } @@ -347,25 +309,23 @@ impl Server { /// /// A SETUP that fails to parse or negotiate closes the session with the matching code, /// so the peer learns why instead of seeing a bare disconnect. - pub async fn accept_request(&self, now: Instant, session: S) -> Result, Error> - where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, - { + pub fn accept_request( + &self, + now: Instant, + session: impl crate::transport::poll::Boxable, + ) -> impl Future> { + // Box before any protocol code runs, so it is compiled once rather than per transport. + self.accept_boxed(Deadlines::new(now), Transport::new(session)) + } + + async fn accept_boxed(&self, runtime: Deadlines, session: Transport) -> Result { let mut refused = session.clone(); - self.handshake(now, session) + self.handshake(runtime, session) .await .inspect_err(|err| close(&mut refused, err, None)) } - async fn handshake(&self, now: Instant, mut session: S) -> Result, Error> - where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, - { - let runtime = Deadlines::new(now); + async fn handshake(&self, runtime: Deadlines, mut session: Transport) -> Result { let (encoding, supported) = match session.protocol() { Some(alpn @ (ALPN_22 | ALPN_21 | ALPN_20 | ALPN_19 | ALPN_18 | ALPN_17)) => { let draft = match alpn { @@ -401,10 +361,8 @@ impl Server { .ok_or(Error::Version)?; (v, v.into()) } - // Every lite ALPN goes through the same entry point, which is also - // what a `!Send` transport calls directly. Some(ALPN_LITE_07_WIP | ALPN_LITE_06 | ALPN_LITE_05 | ALPN_LITE_04 | ALPN_LITE_03) => { - return self.handshake_lite(now, session).await; + return self.handshake_lite(runtime, session).await; } Some(ALPN_LITE) | None => { let supported = self.versions.filter(&NEGOTIATED.into()).ok_or(Error::Version)?; @@ -472,31 +430,26 @@ impl Server { inner: Some(RequestInner { server: self.clone(), runtime, - handshake: PausedHandshake::Boxed(Box::new(PausedLegacy { + handshake: PausedHandshake::Legacy(PausedLegacy { session, stream, version, request_id_max, peer_declared, client_setup_size, - })), + }), }), }) } /// 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: Deadlines, - mut session: S, + mut session: Transport, version: ietf::Version, - ) -> Result, Error> - where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, - { + ) -> Result { let peer_setup = ietf::accept_setup(&mut session, version).await?; Ok(Handshake { path: peer_setup.path.clone(), @@ -512,11 +465,11 @@ impl Server { inner: Some(RequestInner { server: self.clone(), runtime, - handshake: PausedHandshake::Boxed(Box::new(PausedIetfModern { + handshake: PausedHandshake::IetfModern(PausedIetfModern { session, version, peer_setup, - })), + }), }), }) } @@ -529,7 +482,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, @@ -542,146 +495,110 @@ pub struct Handshake { /// take the peer's token requests before any arrive. auth: crate::auth::Handle, // Taken by `ok`/`close`; `Drop` rejects the handshake if neither ran. - inner: Option>, + inner: Option, } /// The parts of a [`Handshake`] consumed by [`Handshake::ok`] / [`Handshake::close`]. -struct RequestInner { +struct RequestInner { server: Server, /// Supplies the clock and timers for the accepted session. runtime: Deadlines, - handshake: PausedHandshake, + 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 }, + LiteBare { + session: Transport, + version: lite::Version, + }, /// moq-lite 05+: the client's Setup Stream has been read. `ok()` starts the /// session, seeding the SETUP back so PROBE gating resolves. LiteSetup { - session: S, + session: Transport, version: lite::Version, - client_setup: lite::AcceptedSetup, + client_setup: lite::AcceptedSetup, }, - /// An IETF (or legacy bidi-SETUP) handshake, boxed where its - /// thread-affinity bounds held. The boxing is what keeps [`Handshake`] and - /// its lite path free of those bounds: the ietf machinery erases its - /// futures, which forces a per-target `Send` choice a pinned `!Send` - /// transport cannot satisfy, so the choice is made here, at construction, - /// where the caller proved the bounds. - Boxed(Box>), + IetfModern(PausedIetfModern), + Legacy(PausedLegacy), } -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 { - /// Complete the handshake with the final server config. - fn ok( - self: Box, - server: Server, - runtime: Deadlines, - peer_hop: Option, - auth: crate::auth::Handle, - ) -> Accept; - - /// Reject the handshake, closing the transport with `err`'s wire code. - fn close(self: Box, err: Error); -} - -/// Modern IETF (17/18): the client's SETUP (with its request path) has been +/// Modern IETF (17+): the client's SETUP (with its request path) has been /// read off its uni stream; `ok` starts the session, handing that stream back /// for GOAWAY monitoring. -struct PausedIetfModern { - session: S, +struct PausedIetfModern { + session: Transport, version: ietf::Version, - peer_setup: ietf::PeerSetup, + peer_setup: ietf::PeerSetup, } -impl Paused for PausedIetfModern -where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, -{ +impl PausedIetfModern { + /// Complete the handshake with the final server config. fn ok( - self: Box, + self, server: Server, runtime: Deadlines, peer_hop: Option, auth: crate::auth::Handle, - ) -> Accept { - use crate::util::MaybeBoxedExt as _; - async move { - let Self { - session, - version, - peer_setup, - } = *self; - server.stats.control(stats::Control::Session, peer_setup.size, true); - let (publish, subscribe) = server.stat_tagged_origins(); - - // The client's SETUP was read at the pause; hand the stream back - // for GOAWAY. A server never advertises a path, hence `None`. - let (protocol, goaway) = ietf::start(ietf::Config { - runtime: runtime.clone(), + ) -> Result<(Session, crate::Driver), Error> { + let Self { + session, + version, + peer_setup, + } = self; + server.stats.control(stats::Control::Session, peer_setup.size, true); + let (publish, subscribe) = server.stat_tagged_origins(); + + // The client's SETUP was read at the pause; hand the stream back + // for GOAWAY. A server never advertises a path, hence `None`. + let (protocol, goaway) = ietf::start(ietf::Config { + runtime: runtime.clone(), + stats: server.stats.clone(), + limits: server.limits, + session: session.clone(), + setup: None, + request_id_max: None, + client: false, + publish, + subscribe, + peer_hop, + // Only the dialing side prices a link. + cost: None, + version, + path: None, + authority: None, + peer_setup_stream: Some(peer_setup.stream), + peer_declared: Some(peer_setup.declared), + auth: auth.clone(), + early_unis: peer_setup.early, + extensions: server.extensions, + })?; + tracing::debug!(?version, "connected"); + Ok(Session::new( + runtime, + session, + version.into(), + None, + crate::driver::Protocol::Ietf(protocol), + crate::session::Handles { stats: server.stats.clone(), - limits: server.limits, - session: session.clone(), - setup: None, - request_id_max: None, - client: false, - publish, - subscribe, - peer_hop, - // Only the dialing side prices a link. - cost: None, - version, - path: None, - authority: None, - peer_setup_stream: Some(peer_setup.stream), - peer_declared: Some(peer_setup.declared), - auth: auth.clone(), - early_unis: peer_setup.early, - extensions: server.extensions, - })?; - tracing::debug!(?version, "connected"); - Ok(Session::new( - runtime, - session, - version.into(), - None, - crate::driver::Protocol::Ietf(protocol), - crate::session::Handles { - stats: server.stats.clone(), - goaway, - auth, - estimate: None, - bandwidth: crate::bandwidth::Allocator::default(), - }, - )) - } - .maybe_boxed() - } - - fn close(mut self: Box, err: Error) { - close(&mut self.session, &err, Some(self.version.into())); + goaway, + auth, + estimate: None, + bandwidth: crate::bandwidth::Allocator::default(), + }, + )) } } /// Legacy IETF (draft 14-16) and lite 01/02: the client SETUP has been read /// off the bidi stream (including its request path) but the server SETUP /// hasn't been sent; `ok` finishes it. -struct PausedLegacy { - session: S, - stream: Stream, +struct PausedLegacy { + session: Transport, + stream: Stream, version: Version, request_id_max: Option, /// What the client's SETUP declared, for the options `ok` acts on. @@ -690,143 +607,128 @@ struct PausedLegacy { client_setup_size: usize, } -impl Paused for PausedLegacy -where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, -{ - fn ok( - self: Box, +impl PausedLegacy { + /// Send the server SETUP and start the session. + async fn ok( + self, server: Server, runtime: Deadlines, peer_hop: Option, auth: crate::auth::Handle, - ) -> Accept { - use crate::util::MaybeBoxedExt as _; - async move { - let Self { - session, - stream, - version, - request_id_max, - peer_declared, - client_setup_size, - } = *self; - let mut stream = stream.with_stats(&server.stats); - server.stats.control(stats::Control::Session, client_setup_size, true); - let (publish, subscribe) = server.stat_tagged_origins(); - - // Encode parameters using the version-appropriate type. - let parameters = match version { - Version::Ietf(v) => { - let mut parameters = ietf::Parameters::default(); - // The client's requests, admitted up to our limits and granted back as they close. - parameters.set_varint( - ietf::ParameterVarInt::MaxRequestId, - ietf::initial_max_request_id(server.limits.requests(), true), - ); - parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); - if server.extensions.solicit { - ietf::solicit::into_setup(&mut parameters, v); - } - ietf::hidden::into_setup(&mut parameters, v); - ietf::max_age::into_setup(&mut parameters, v); - ietf::active_count::into_setup(&mut parameters, v); - ietf::flate::into_setup(&mut parameters, v); - parameters.encode_bytes(v)? - } - Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?, - }; - - let server_setup = setup::Server { - version: version.into(), - parameters, - }; - stream.writer.encode(&server_setup).await?; - - let (recv_bw, protocol, goaway, auth, estimate) = match version { - Version::Lite(v) => { - let stream = stream.with_version(v); - // Pre-lite-05: no Setup Stream, so nothing to advertise or seed. - let start = lite::start(lite::Config { - runtime: runtime.clone(), - stats: server.stats.clone(), - client: false, - limits: server.limits, - session: session.clone(), - setup_stream: Some(stream), - publish, - subscribe, - peer_hop, - version: v, - our_setup: lite::Setup::default(), - peer_setup: None, - auth, - })?; - ( - start.recv_bandwidth, - crate::driver::Protocol::Lite(Box::new(start.driver)), - start.goaway, - start.auth, - Some(start.estimate), - ) - } - Version::Ietf(v) => { - let stream = stream.with_version(v); - // Draft 14-16: path came in the bidi SETUP, no uni SETUP to hand back. - let (protocol, goaway) = ietf::start(ietf::Config { - runtime: runtime.clone(), - stats: server.stats.clone(), - limits: server.limits, - session: session.clone(), - setup: Some(stream), - request_id_max, - client: false, - publish, - subscribe, - peer_hop, - cost: None, - version: v, - path: None, - authority: None, - peer_setup_stream: None, - peer_declared: Some(peer_declared), - auth: auth.clone(), - early_unis: Vec::new(), - extensions: server.extensions, - })?; - (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, None) + ) -> Result<(Session, crate::Driver), Error> { + let Self { + session, + stream, + version, + request_id_max, + peer_declared, + client_setup_size, + } = self; + let mut stream = stream.with_stats(&server.stats); + server.stats.control(stats::Control::Session, client_setup_size, true); + let (publish, subscribe) = server.stat_tagged_origins(); + + // Encode parameters using the version-appropriate type. + let parameters = match version { + Version::Ietf(v) => { + let mut parameters = ietf::Parameters::default(); + // The client's requests, admitted up to our limits and granted back as they close. + parameters.set_varint( + ietf::ParameterVarInt::MaxRequestId, + ietf::initial_max_request_id(server.limits.requests(), true), + ); + parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); + if server.extensions.solicit { + ietf::solicit::into_setup(&mut parameters, v); } - }; + ietf::hidden::into_setup(&mut parameters, v); + ietf::max_age::into_setup(&mut parameters, v); + ietf::active_count::into_setup(&mut parameters, v); + ietf::flate::into_setup(&mut parameters, v); + parameters.encode_bytes(v)? + } + Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?, + }; - Ok(Session::new( - runtime, - session, - version, - recv_bw, - protocol, - crate::session::Handles { + let server_setup = setup::Server { + version: version.into(), + parameters, + }; + stream.writer.encode(&server_setup).await?; + + let (recv_bw, protocol, goaway, auth, estimate) = match version { + Version::Lite(v) => { + let stream = stream.with_version(v); + // Pre-lite-05: no Setup Stream, so nothing to advertise or seed. + let start = lite::start(lite::Config { + runtime: runtime.clone(), stats: server.stats.clone(), - goaway, + client: false, + limits: server.limits, + session: session.clone(), + setup_stream: Some(stream), + publish, + subscribe, + peer_hop, + version: v, + our_setup: lite::Setup::default(), + peer_setup: None, auth, - estimate, - bandwidth: crate::bandwidth::Allocator::default(), - }, - )) - } - .maybe_boxed() - } + })?; + ( + start.recv_bandwidth, + crate::driver::Protocol::Lite(Box::new(start.driver)), + start.goaway, + start.auth, + Some(start.estimate), + ) + } + Version::Ietf(v) => { + let stream = stream.with_version(v); + // Draft 14-16: path came in the bidi SETUP, no uni SETUP to hand back. + let (protocol, goaway) = ietf::start(ietf::Config { + runtime: runtime.clone(), + stats: server.stats.clone(), + limits: server.limits, + session: session.clone(), + setup: Some(stream), + request_id_max, + client: false, + publish, + subscribe, + peer_hop, + cost: None, + version: v, + path: None, + authority: None, + peer_setup_stream: None, + peer_declared: Some(peer_declared), + auth: auth.clone(), + early_unis: Vec::new(), + extensions: server.extensions, + })?; + (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, None) + } + }; - fn close(mut self: Box, err: Error) { - close(&mut self.session, &err, Some(self.version)); + Ok(Session::new( + runtime, + session, + version, + recv_bw, + protocol, + crate::session::Handles { + stats: server.stats.clone(), + goaway, + auth, + estimate, + bandwidth: crate::bandwidth::Allocator::default(), + }, + )) } } -impl Handshake -where - S: crate::transport::poll::Session, -{ +impl Handshake { /// The request path the client advertised in its SETUP. /// /// Empty when the client advertised none: either it sent an empty path, or the @@ -920,14 +822,14 @@ where self } - fn inner_mut(&mut self) -> &mut RequestInner { + fn inner_mut(&mut self) -> &mut RequestInner { self.inner.as_mut().expect("request already responded") } /// Accept the session, returning the [`Session`] and its [`Driver`](crate::Driver). /// /// Poll or spawn the returned driver to run the session. - pub async fn ok(mut self) -> Result<(Session, crate::Driver), Error> { + pub async fn ok(mut self) -> Result<(Session, crate::Driver), Error> { let peer_hop = Some(self.assigned_hop); let auth = self.auth.clone(); let RequestInner { @@ -948,7 +850,9 @@ where server.stats.control(stats::Control::Session, client_setup.size, true); server.start_lite(runtime, session, version, Some(client_setup), peer_hop, auth) } - PausedHandshake::Boxed(paused) => paused.ok(server, runtime, peer_hop, auth).await, + PausedHandshake::IetfModern(paused) => paused.ok(server, runtime, peer_hop, auth), + // Boxed so the bidi SETUP write of these old versions doesn't size every caller's future. + PausedHandshake::Legacy(paused) => Box::pin(paused.ok(server, runtime, peer_hop, auth)).await, } } @@ -959,25 +863,26 @@ where } } -impl RequestInner { +impl RequestInner { fn close(self, err: Error) { let (mut session, version) = match self.handshake { - PausedHandshake::LiteBare { session, version } => (session, version), - PausedHandshake::LiteSetup { session, version, .. } => (session, version), - PausedHandshake::Boxed(paused) => return paused.close(err), + PausedHandshake::LiteBare { session, version } => (session, version.into()), + PausedHandshake::LiteSetup { session, version, .. } => (session, version.into()), + PausedHandshake::IetfModern(paused) => (paused.session, paused.version.into()), + PausedHandshake::Legacy(paused) => (paused.session, paused.version), }; - close(&mut session, &err, Some(version.into())); + close(&mut session, &err, Some(version)); } } /// Close `session` with `err`'s wire code. -fn close(session: &mut S, err: &Error, version: Option) { +fn close(session: &mut Transport, err: &Error, version: Option) { let error = SessionError::from(err); let code = version.map_or_else(|| error.to_code(), |version| version.session_code(&error)); session.close(code, &err.to_string()); } -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) { @@ -1433,7 +1338,7 @@ mod tests { let observed = session.clone(); let result = Server::new() .with_versions(alpn.parse::().unwrap().into()) - .accept_request_lite(moq_clock::now(), session) + .accept_request(moq_clock::now(), session) .await; assert!( matches!(result, Err(Error::ProtocolViolation)), @@ -1448,10 +1353,7 @@ mod tests { async fn uri_binding_accepts_setup_without_path() { let mut session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]); session.binding = crate::transport::Binding::Uri; - let request = Server::new() - .accept_request_lite(moq_clock::now(), session) - .await - .unwrap(); + let request = Server::new().accept_request(moq_clock::now(), session).await.unwrap(); assert_eq!(request.path(), ""); } @@ -1472,10 +1374,7 @@ mod tests { [lite05_group(), lite05_setup(Some("/team/room"), None, None)], ); let stops = session.stops.clone(); - let request = Server::new() - .accept_request_lite(moq_clock::now(), session) - .await - .unwrap(); + let request = Server::new().accept_request(moq_clock::now(), session).await.unwrap(); assert_eq!(request.path(), "/team/room"); assert!(stops.lock().unwrap().is_empty(), "the early stream must stay open"); } @@ -1484,10 +1383,7 @@ mod tests { async fn accept_request_reads_buffered_setup_after_transport_close() { let mut session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/closed"), None, None)]); crate::transport::poll::Session::close(&mut session, SessionError::Cancel.to_code(), "closed"); - let request = Server::new() - .accept_request_lite(moq_clock::now(), session) - .await - .unwrap(); + let request = Server::new().accept_request(moq_clock::now(), session).await.unwrap(); assert_eq!(request.path(), "/closed"); } @@ -1495,10 +1391,7 @@ mod tests { async fn accepted_lite_setup_refuses_a_second_setup_stream() { let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]); let transport = session.clone(); - let request = Server::new() - .accept_request_lite(moq_clock::now(), session) - .await - .unwrap(); + let request = Server::new().accept_request(moq_clock::now(), session).await.unwrap(); let (_session, mut driver) = request.ok().await.unwrap(); transport.uni.lock().unwrap().push_back(vec![1]); let _ = driver.poll(moq_clock::now(), &kio::Waiter::noop()); @@ -1532,12 +1425,12 @@ mod tests { inner: Some(RequestInner { server: Server::new().with_publisher(&origin), runtime: Deadlines::new(moq_clock::now()), - handshake: PausedHandshake::Boxed(Box::new(PausedIetfModern { - session: transport, + handshake: PausedHandshake::IetfModern(PausedIetfModern { + session: Transport::new(transport), version, peer_setup: ietf::PeerSetup { stream: crate::coding::Reader::new( - crate::lite::test_transport::PendingRecv, + crate::transport::boxed::RecvStream::new(crate::lite::test_transport::PendingRecv), Version::Ietf(version), ), size: 0, @@ -1546,7 +1439,7 @@ mod tests { declared: ietf::peer::Peer::default(), early: Vec::new(), }, - })), + }), }), }; let assigned = request.assigned_hop; diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index 5d217ea7cbc..8bb85dbee38 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -3,6 +3,7 @@ use std::{sync::Arc, task::Poll, time::Duration}; use crate::transport::Stats as _; +use crate::transport::poll::Session as _; use crate::{Error, SessionError, Version, auth, bandwidth, coding::StreamCodes, goaway}; @@ -315,17 +316,14 @@ pub(super) struct Handles { } impl Session { - pub(super) fn new( + pub(super) fn new( runtime: crate::time::Deadlines, - session: S, + session: crate::driver::Transport, version: Version, recv_bandwidth: Option, - protocol: crate::driver::Protocol, + protocol: crate::driver::Protocol, handles: Handles, - ) -> (Self, crate::Driver) - where - S: crate::transport::poll::Session, - { + ) -> (Self, crate::Driver) { let Handles { goaway, auth, @@ -409,15 +407,15 @@ impl Session { /// anyone is consuming them, and enforces GOAWAY deadlines independently of writes. /// /// Finishes once the transport reports closed; everything else is moot then. -pub(crate) struct Supervisor { +pub(crate) struct Supervisor { goaway: Option, version: Version, local_close: Arc, runtime: crate::time::Deadlines, - session: S, + session: crate::driver::Transport, // A dedicated clone for the close watch, since each pending poll operation // needs its own handle. - closed_watch: S, + closed_watch: crate::driver::Transport, /// Handle-side close requests; `None` once one was executed (only the /// first close matters, and the channel closing is the last handle /// dropping). @@ -461,7 +459,7 @@ enum SamplerMode { }, } -impl Supervisor { +impl Supervisor { const POLL_INTERVAL: Duration = Duration::from_millis(100); pub(crate) fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { diff --git a/rs/moq-net/src/transport.rs b/rs/moq-net/src/transport.rs index b81613ee311..48fb9cdda33 100644 --- a/rs/moq-net/src/transport.rs +++ b/rs/moq-net/src/transport.rs @@ -3,11 +3,14 @@ //! Implement these traits for a transport, or wrap a backend at the application //! boundary. `moq-tokio`, `moq-wasm`, and `moq-uring` provide their own adapters, and //! [`memory`] links two peers in one process. -//! The async helpers only drive the required poll methods. Thread-local transports -//! stay supported; [`poll::Boxable`] adds native thread bounds for boxed drivers. +//! The async helpers only drive the required poll methods. Sessions take a +//! [`poll::Boxable`] transport, which adds native thread bounds; wrap a thread-local +//! transport in a type that checks the thread on use (`moq-uring` has one). use std::time::Duration; +#[cfg(feature = "async")] +pub(crate) mod boxed; pub mod memory; /// A trait that is Send on native targets and empty on WASM. @@ -239,12 +242,10 @@ pub mod poll { /// operations release retained resources; shared resources must not be held /// across a wait. Implementations register the supplied waker before Pending. pub trait Session: Clone + 'static { - /// The outgoing stream type. Only the poll half is required, so a `!Send` - /// session can hang `!Send` streams off it. + /// The outgoing stream type. type SendStream: SendStream; - /// The incoming stream type. Only the poll half is required, so a `!Send` - /// session can hang `!Send` streams off it. + /// The incoming stream type. type RecvStream: RecvStream; /// The error type for every operation on this session. @@ -450,14 +451,16 @@ pub mod poll { /// A pending [`SendStream::closed`]. SendClosed, SendStream, poll_closed, Result<(), S::Error>); - /// A transport whose session, streams, and errors can be captured by the - /// boxed drivers (`Send` boxes on native): what the moq-transport path - /// requires until it too becomes named machines. Implemented automatically. + /// A transport that [`Client`](crate::Client) and [`Server`](crate::Server) can box: + /// the session, its streams, send groups, and errors are thread-safe on native. + /// Implemented automatically. + /// + /// Boxing is what compiles the session code once per binary rather than once per + /// transport. pub trait Boxable: Session< - SendStream: MaybeSend + SendStream, - RecvStream: MaybeSend, - Error: MaybeSend, + SendStream: MaybeSend + MaybeSync + SendStream, + RecvStream: MaybeSend + MaybeSync, > + MaybeSend + MaybeSync { @@ -465,9 +468,8 @@ pub mod poll { impl Boxable for S where S: Session< - SendStream: MaybeSend + SendStream, - RecvStream: MaybeSend, - Error: MaybeSend, + SendStream: MaybeSend + MaybeSync + SendStream, + RecvStream: MaybeSend + MaybeSync, > + MaybeSend + MaybeSync { diff --git a/rs/moq-net/src/transport/boxed.rs b/rs/moq-net/src/transport/boxed.rs new file mode 100644 index 00000000000..e4272edab18 --- /dev/null +++ b/rs/moq-net/src/transport/boxed.rs @@ -0,0 +1,560 @@ +//! The one transport type the session code is compiled for. +//! +//! [`Client`](crate::Client) and [`Server`](crate::Server) box the caller's transport +//! here before running any protocol code, so the lite and ietf drivers are +//! monomorphized once per binary instead of once per transport and per crate that +//! awaits them. Each transport only instantiates the thin forwarding impls below. + +use std::any::Any; +use std::task::{Context, Poll}; + +use bytes::{Buf, Bytes}; + +use super::poll::{self, SendOptions}; +use super::{Binding, CloseKind, MaybeSend, MaybeSync, Probe, Wire}; + +/// A boxed [`poll::Boxable`] session. +pub(crate) struct Session(Box); + +impl Session { + pub(crate) fn new(session: S) -> Self { + Self(Box::new(session)) + } +} + +impl Clone for Session { + fn clone(&self) -> Self { + Self(self.0.clone_box()) + } +} + +/// A boxed transport error, keeping both code registries and the message. +#[derive(Debug)] +pub(crate) struct Error(Box); + +impl Error { + fn new(err: impl super::Error) -> Self { + Self(Box::new(err)) + } +} + +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.0.source() + } +} + +impl super::Error for Error { + fn close_kind(&self) -> CloseKind { + self.0.close_kind() + } + + fn session_error(&self) -> Option<(u32, String)> { + self.0.session_error() + } + + fn stream_error(&self) -> Option { + self.0.stream_error() + } +} + +/// A boxed send group. It only ever goes back to the session that opened it. +pub(crate) struct Group(Box); + +impl Clone for Group { + fn clone(&self) -> Self { + Self(self.0.clone_box()) + } +} + +impl poll::SendGroup for Group { + fn set_priority(&mut self, priority: i64) { + self.0.set_priority(priority) + } +} + +/// A boxed outgoing stream. +pub(crate) struct SendStream(Box); + +impl SendStream { + /// Box a stream whose send group, if any, is a [`Group`]. + pub(crate) fn new(stream: impl poll::SendStream + MaybeSend + MaybeSync) -> Self { + Self(Box::new(stream)) + } +} + +/// A boxed incoming stream. +pub(crate) struct RecvStream(Box); + +impl RecvStream { + pub(crate) fn new(stream: impl poll::RecvStream + MaybeSend + MaybeSync) -> Self { + Self(Box::new(stream)) + } +} + +/// A boxed [`poll::Lease`], borrowed from a [`RecvStream`]. +pub(crate) struct Lease<'a>(Box); + +/// The transport's connection stats, read once. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct Stats { + bytes_sent: Option, + bytes_received: Option, + bytes_lost: Option, + packets_sent: Option, + packets_received: Option, + packets_lost: Option, + rtt: Option, + estimated_send_rate: Option, + streams_evicted: Option, +} + +impl Stats { + fn read(stats: impl super::Stats) -> Self { + Self { + bytes_sent: stats.bytes_sent(), + bytes_received: stats.bytes_received(), + bytes_lost: stats.bytes_lost(), + packets_sent: stats.packets_sent(), + packets_received: stats.packets_received(), + packets_lost: stats.packets_lost(), + rtt: stats.rtt(), + estimated_send_rate: stats.estimated_send_rate(), + streams_evicted: stats.streams_evicted(), + } + } +} + +impl super::Stats for Stats { + fn bytes_sent(&self) -> Option { + self.bytes_sent + } + fn bytes_received(&self) -> Option { + self.bytes_received + } + fn bytes_lost(&self) -> Option { + self.bytes_lost + } + fn packets_sent(&self) -> Option { + self.packets_sent + } + fn packets_received(&self) -> Option { + self.packets_received + } + fn packets_lost(&self) -> Option { + self.packets_lost + } + fn rtt(&self) -> Option { + self.rtt + } + fn estimated_send_rate(&self) -> Option { + self.estimated_send_rate + } + fn streams_evicted(&self) -> Option { + self.streams_evicted + } +} + +trait ErasedSession: MaybeSend + MaybeSync { + fn clone_box(&self) -> Box; + fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll>; + fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll>; + fn poll_open_uni( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll>; + fn poll_open_bi( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll>; + fn send_group(&mut self, priority: i64) -> Option; + fn send_datagram(&mut self, payload: Bytes, options: SendOptions<'_, Group>) -> Result<(), Error>; + fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll>; + fn max_datagram_size(&self) -> usize; + fn protocol(&self) -> Option<&str>; + fn binding(&self) -> Binding; + fn close(&mut self, code: u32, reason: &str); + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll; + fn stats(&self) -> Stats; + fn wire(&self) -> Option; + fn probe(&self) -> Option; +} + +trait ErasedGroup: MaybeSend + MaybeSync { + fn clone_box(&self) -> Box; + fn set_priority(&mut self, priority: i64); + fn as_any(&self) -> &dyn Any; +} + +trait ErasedSend: MaybeSend + MaybeSync { + fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll>; + fn poll_write_buf(&mut self, cx: &mut Context<'_>, buf: &mut dyn Buf) -> Poll>; + fn set_priority(&mut self, priority: i64); + fn set_group(&mut self, group: &Group); + fn finish(&mut self) -> Result<(), Error>; + fn reset(&mut self, code: u32); + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll>; +} + +trait ErasedRecv: MaybeSend + MaybeSync { + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Error>>; + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Error>>; + fn set_priority(&mut self, priority: i64); + fn stop(&mut self, code: u32); +} + +// Not `MaybeSend`: a lease may hold its transport's lock, and never outlives a poll. +trait ErasedLease: Buf { + fn chunks(&self) -> Box + '_>; + fn is_fin(&self) -> bool; + fn release(self: Box); +} + +type GroupOf = <::SendStream as poll::SendStream>::Group; + +/// Recover the transport's own group from a boxed one. +fn unbox<'a, G: 'static>(options: SendOptions<'a, Group>) -> SendOptions<'a, G> { + SendOptions { + group: options.group.map(|group| { + group + .0 + .as_any() + .downcast_ref::() + .expect("send group belongs to another transport") + }), + priority: options.priority, + } +} + +fn bi((send, recv): poll::BiStreams) -> (SendStream, RecvStream) { + (SendStream(Box::new(send)), RecvStream(Box::new(recv))) +} + +impl ErasedSession for S { + fn clone_box(&self) -> Box { + Box::new(self.clone()) + } + + fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::Session::poll_accept_uni(self, cx) + .map(|res| res.map(|recv| RecvStream(Box::new(recv))).map_err(Error::new)) + } + + fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::Session::poll_accept_bi(self, cx).map(|res| res.map(bi::).map_err(Error::new)) + } + + fn poll_open_uni( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll> { + poll::Session::poll_open_uni(self, cx, unbox(options)) + .map(|res| res.map(|send| SendStream(Box::new(send))).map_err(Error::new)) + } + + fn poll_open_bi( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll> { + poll::Session::poll_open_bi(self, cx, unbox(options)).map(|res| res.map(bi::).map_err(Error::new)) + } + + fn send_group(&mut self, priority: i64) -> Option { + poll::Session::send_group(self, priority).map(|group| Group(Box::new(group))) + } + + fn send_datagram(&mut self, payload: Bytes, options: SendOptions<'_, Group>) -> Result<(), Error> { + poll::Session::send_datagram(self, payload, unbox::>(options)).map_err(Error::new) + } + + fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::Session::poll_recv_datagram(self, cx).map_err(Error::new) + } + + fn max_datagram_size(&self) -> usize { + poll::Session::max_datagram_size(self) + } + + fn protocol(&self) -> Option<&str> { + poll::Session::protocol(self) + } + + fn binding(&self) -> Binding { + poll::Session::binding(self) + } + + fn close(&mut self, code: u32, reason: &str) { + poll::Session::close(self, code, reason) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll { + poll::Session::poll_closed(self, cx).map(Error::new) + } + + fn stats(&self) -> Stats { + Stats::read(poll::Session::stats(self)) + } + + fn wire(&self) -> Option { + poll::Session::wire(self) + } + + fn probe(&self) -> Option { + poll::Session::probe(self) + } +} + +impl ErasedGroup for G { + fn clone_box(&self) -> Box { + Box::new(self.clone()) + } + + fn set_priority(&mut self, priority: i64) { + poll::SendGroup::set_priority(self, priority) + } + + fn as_any(&self) -> &dyn Any { + self + } +} + +impl + MaybeSend + MaybeSync> ErasedSend for T { + fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + poll::SendStream::poll_write(self, cx, buf).map_err(Error::new) + } + + fn poll_write_buf(&mut self, cx: &mut Context<'_>, mut buf: &mut dyn Buf) -> Poll> { + // `&mut dyn Buf` forwards `copy_to_bytes`, so a `Bytes` source stays zero-copy. + poll::SendStream::poll_write_buf(self, cx, &mut buf).map_err(Error::new) + } + + fn set_priority(&mut self, priority: i64) { + poll::SendStream::set_priority(self, priority) + } + + fn set_group(&mut self, group: &Group) { + let group = group + .0 + .as_any() + .downcast_ref::() + .expect("send group belongs to another transport"); + poll::SendStream::set_group(self, group) + } + + fn finish(&mut self) -> Result<(), Error> { + poll::SendStream::finish(self).map_err(Error::new) + } + + fn reset(&mut self, code: u32) { + poll::SendStream::reset(self, code) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::SendStream::poll_closed(self, cx).map_err(Error::new) + } +} + +impl ErasedRecv for T { + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Error>> { + poll::RecvStream::poll_lease(self, cx, want) + .map(|res| res.map(|lease| Lease(Box::new(lease))).map_err(Error::new)) + } + + // Forwarded rather than left to the provided method, so reading a chunk never boxes a lease. + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Error>> { + poll::RecvStream::poll_read_chunk(self, cx, max).map_err(Error::new) + } + + fn set_priority(&mut self, priority: i64) { + poll::RecvStream::set_priority(self, priority) + } + + fn stop(&mut self, code: u32) { + poll::RecvStream::stop(self, code) + } +} + +impl ErasedLease for L { + fn chunks(&self) -> Box + '_> { + Box::new(poll::Lease::chunks(self)) + } + + fn is_fin(&self) -> bool { + poll::Lease::is_fin(self) + } + + fn release(self: Box) { + poll::Lease::release(*self) + } +} + +impl poll::Session for Session { + type SendStream = SendStream; + type RecvStream = RecvStream; + type Error = Error; + + fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_accept_uni(cx) + } + + fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_accept_bi(cx) + } + + fn poll_open_uni( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll> { + self.0.poll_open_uni(cx, options) + } + + fn poll_open_bi( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll> { + self.0.poll_open_bi(cx, options) + } + + fn send_group(&mut self, priority: i64) -> Option { + self.0.send_group(priority) + } + + fn send_datagram(&mut self, payload: Bytes, options: SendOptions<'_, Group>) -> Result<(), Error> { + self.0.send_datagram(payload, options) + } + + fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_recv_datagram(cx) + } + + fn max_datagram_size(&self) -> usize { + self.0.max_datagram_size() + } + + fn protocol(&self) -> Option<&str> { + self.0.protocol() + } + + fn binding(&self) -> Binding { + self.0.binding() + } + + fn close(&mut self, code: u32, reason: &str) { + self.0.close(code, reason) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll { + self.0.poll_closed(cx) + } + + fn stats(&self) -> impl super::Stats { + self.0.stats() + } + + fn wire(&self) -> Option { + self.0.wire() + } + + fn probe(&self) -> Option { + self.0.probe() + } +} + +impl poll::SendStream for SendStream { + type Error = Error; + type Group = Group; + + fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + self.0.poll_write(cx, buf) + } + + fn poll_write_buf(&mut self, cx: &mut Context<'_>, buf: &mut B) -> Poll> { + self.0.poll_write_buf(cx, buf) + } + + fn set_priority(&mut self, priority: i64) { + self.0.set_priority(priority) + } + + fn set_group(&mut self, group: &Group) { + self.0.set_group(group) + } + + fn finish(&mut self) -> Result<(), Error> { + self.0.finish() + } + + fn reset(&mut self, code: u32) { + self.0.reset(code) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_closed(cx) + } +} + +impl poll::RecvStream for RecvStream { + type Error = Error; + type Lease<'a> = Lease<'a>; + + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Error>> { + self.0.poll_lease(cx, want) + } + + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Error>> { + self.0.poll_read_chunk(cx, max) + } + + fn set_priority(&mut self, priority: i64) { + self.0.set_priority(priority) + } + + fn stop(&mut self, code: u32) { + self.0.stop(code) + } +} + +impl Buf for Lease<'_> { + fn remaining(&self) -> usize { + self.0.remaining() + } + + fn chunk(&self) -> &[u8] { + self.0.chunk() + } + + fn advance(&mut self, n: usize) { + self.0.advance(n) + } + + // Forwarded so a transport's zero-copy override survives the box. + fn copy_to_bytes(&mut self, n: usize) -> Bytes { + self.0.copy_to_bytes(n) + } +} + +impl poll::Lease for Lease<'_> { + fn chunks(&self) -> impl Iterator { + self.0.chunks() + } + + fn is_fin(&self) -> bool { + self.0.is_fin() + } + + fn release(self) { + self.0.release() + } +} diff --git a/rs/moq-net/src/transport/memory.rs b/rs/moq-net/src/transport/memory.rs index 5323e5ce68b..91ec7e502b8 100644 --- a/rs/moq-net/src/transport/memory.rs +++ b/rs/moq-net/src/transport/memory.rs @@ -680,7 +680,7 @@ mod tests { #[test] fn bounds() { - fn boxable>() {} + fn boxable() {} boxable::(); } diff --git a/rs/moq-relay/MIGRATE.md b/rs/moq-relay/MIGRATE.md index 866a7e1d2d0..1ac5f299e8c 100644 --- a/rs/moq-relay/MIGRATE.md +++ b/rs/moq-relay/MIGRATE.md @@ -20,6 +20,10 @@ Upgrade any client that still speaks only `webtransport` over WebSocket before t `uring::Workers::serve(cluster, auth, shutdown, sessions)` is now `spawn` with the same arguments, which sets every worker up without accepting, followed by `serve()`, which starts them accepting. Install anything that must be in place before the first session (`Relay::run` installs its signal handlers) between the two. +### `Accepted` takes any transport's handshake (#6176) + +`Accepted` loses its transport type parameter, since `moq_net::server::Handshake` has none: write `Accepted` instead of `Accepted`. + ### `supervise` takes a `Serving` guard (#6144) `moq_relay::supervise` takes the `shutdown::Serving` from `Observer::serve` in place of the `Observer`. diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 09fda630dfe..9fa787a0181 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -1075,9 +1075,9 @@ pub fn request_for(auth: &Auth, request: &moq_tokio::server::Request) -> Request } /// Build an auth request from the path, credential, and role carried in SETUP. -pub(crate) fn request_for_setup( +pub(crate) fn request_for_setup( auth: &Auth, - setup: &moq_net::server::Handshake, + setup: &moq_net::server::Handshake, transport: moq_auth::Transport, ) -> Request { let (path, query) = match setup.path().split_once('?') { diff --git a/rs/moq-relay/src/incoming.rs b/rs/moq-relay/src/incoming.rs index 86ea3e3d492..2761d7c70e0 100644 --- a/rs/moq-relay/src/incoming.rs +++ b/rs/moq-relay/src/incoming.rs @@ -2,7 +2,6 @@ use crate::auth; use axum::http; -use moq_net::transport::poll; /// A session whose SETUP awaits the relay's reply: what a [`Connection`](crate::Connection) serves. /// @@ -11,20 +10,20 @@ pub trait Incoming: Sealed {} impl Incoming for moq_tokio::server::Request {} -impl Incoming for Accepted {} +impl Incoming for Accepted {} /// A session accepted on any moq-net transport, its SETUP awaiting the relay's reply. /// /// The auth decider sees the path, query, token, and role the SETUP carried, so the /// client must send its path in-band. -pub struct Accepted { - handshake: moq_net::server::Handshake, +pub struct Accepted { + handshake: moq_net::server::Handshake, transport: moq_auth::Transport, } -impl Accepted { +impl Accepted { /// Wrap `handshake`, reported to the auth decider as arriving over `transport`. - pub fn new(handshake: moq_net::server::Handshake, transport: moq_auth::Transport) -> Self { + pub fn new(handshake: moq_net::server::Handshake, transport: moq_auth::Transport) -> Self { Self { handshake, transport } } } @@ -145,15 +144,15 @@ impl Sealed for moq_tokio::server::Request { } } -impl Sealed for Accepted { +impl Sealed for Accepted { /// Polled on the connection's own task, so it needs no runtime to spawn on. - type Driver = moq_net::Driver; + type Driver = moq_net::Driver; fn remote(&self) -> Option { None } - fn transport(&self) -> impl std::fmt::Display + use { + fn transport(&self) -> impl std::fmt::Display + use<> { self.transport.as_str() } diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 371bea3f427..29ea532414c 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -725,16 +725,10 @@ async fn serve_connection( Some(_) => moq_net::transport::Binding::Uri, None => moq_net::transport::Binding::InBand, }; - let session = moq_uring::transport::Session::new(transport, binding); + // moq-net boxes the transport with a native `Send` bound, which only a `Local` + // session meets; it still never leaves this worker. + let session = moq_uring::transport::Local::new(moq_uring::transport::Session::new(transport, binding)); let now = moq_net::time::now(); - // moq-lite on its own ALPN runs unboxed on this worker. Everything else - // (moq-transport, and the versions that negotiate in a bidi SETUP, which a - // browser offering no subprotocol falls back to) boxes its futures with a - // native `Send` bound, which only a `Local` session meets. - let unboxed = alpn - .as_deref() - .and_then(moq_net::Version::from_alpn) - .is_some_and(|version| version.is_lite()); let mut accepted = Accepted { handle, conn, @@ -746,24 +740,12 @@ async fn serve_connection( alpn, url, }; - match unboxed { - true => { - let request = accepted - .deadline - .bound(server.accept_request_lite(now, session)) - .await - .context("moq handshake failed")?; - serve_request(accepted, request, lifecycle).await - } - false => { - let request = accepted - .deadline - .bound(server.accept_request(now, moq_uring::transport::Local::new(session))) - .await - .context("moq handshake failed")?; - serve_request(accepted, request, lifecycle).await - } - } + let request = accepted + .deadline + .bound(server.accept_request(now, session)) + .await + .context("moq handshake failed")?; + serve_request(accepted, request, lifecycle).await } /// A connection past its transport handshake, with the facts read off it @@ -780,14 +762,14 @@ struct Accepted { url: Option, } -/// A started session's supervision, boxed so both handshake paths return one type. +/// A started session's supervision. type Supervision = futures::future::BoxFuture<'static, anyhow::Result<()>>; /// Authenticate on the shared runtime, then start the session on the worker. /// Returns the session's supervision, which runs on the shared runtime too. -async fn serve_request( +async fn serve_request( accepted: Accepted, - request: moq_net::server::Handshake, + request: moq_net::server::Handshake, lifecycle: &Lifecycle, ) -> anyhow::Result { let Accepted { diff --git a/rs/moq-room/src/chat.rs b/rs/moq-room/src/chat.rs index b9d2db7e208..a2ec0d24806 100644 --- a/rs/moq-room/src/chat.rs +++ b/rs/moq-room/src/chat.rs @@ -255,18 +255,12 @@ mod tests { .with_subscriber(to.clone()); tokio::join!( async { - let (session, driver) = client - .connect_lite(moq_net::time::now(), client_transport) - .await - .unwrap(); + let (session, driver) = client.connect(moq_net::time::now(), client_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session }, async { - let (session, driver) = server - .accept_lite(moq_net::time::now(), server_transport) - .await - .unwrap(); + let (session, driver) = server.accept(moq_net::time::now(), server_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session } diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 935e76922f3..575c612ba55 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -741,7 +741,7 @@ impl Server { let (request, deadline) = accept_request(&setup.server, session, deadline).await?; // Only WebTransport carries a request URL; raw QUIC puts the path in the SETUP. let transport = match url { Some(_) => Transport::WebTransport, None => Transport::Quic }; - Ok(Request { transport, url, identity, authority, link, kind: RequestKind::Quic(Box::new(request)), quic, deadline, active }) + Ok(Request { transport, url, identity, authority, link, handshake: Box::new(request), quic, deadline, active }) }.boxed()); } } @@ -765,7 +765,7 @@ impl Server { identity, authority, link, - kind: RequestKind::Iroh(Box::new(request)), + handshake: Box::new(request), #[cfg(feature = "quic")] quic: None, deadline, @@ -935,7 +935,7 @@ impl Listener { alpn: Some(protocol), ..Default::default() }, - kind: RequestKind::Qmux(Box::new(request)), + handshake: Box::new(request), #[cfg(feature = "quic")] quic: None, deadline, @@ -1053,11 +1053,9 @@ async fn accept_request( server: &moq_net::Server, session: S, deadline: Option, -) -> crate::Result<(PendingRequest, Option)> +) -> crate::Result<(moq_net::server::Handshake, Option)> where S: moq_net::transport::poll::Boxable + Send + Sync, - S::SendStream: moq_net::transport::MaybeSync, - S::RecvStream: moq_net::transport::MaybeSync, { let deadline = deadline.map(|at| Deadline::new(at, session.clone())); let accept = async { Ok(server.accept_request(moq_net::time::now(), session).await?) }; @@ -1079,7 +1077,7 @@ pub(crate) async fn accept_websocket( authority, identity: options.identity, link: options.link, - kind: RequestKind::Qmux(Box::new(request)), + handshake: Box::new(request), #[cfg(feature = "quic")] quic: None, deadline, @@ -1572,7 +1570,7 @@ async fn stream_request( authority: None, identity: None, link, - kind: RequestKind::Qmux(Box::new(request)), + handshake: Box::new(request), #[cfg(feature = "quic")] quic: None, deadline, @@ -1584,29 +1582,6 @@ async fn stream_request( } } -/// An accepted connection whose MoQ SETUP has already been exchanged. -/// -/// Every backend drives the transport connect *and* the MoQ handshake up front, so the -/// [`path`](Request::path)/[`role`](Request::role) a client advertised are available on -/// 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; - -pub(crate) enum RequestKind { - #[cfg(feature = "quic")] - Quic(Box>>), - #[cfg(feature = "iroh")] - Iroh(Box>>), - #[cfg(any( - feature = "tcp", - all(feature = "uds", unix), - feature = "websocket", - feature = "webrtc" - ))] - Qmux(Box>), -} - /// The transport-level facts a backend captures while accepting a connection, before the /// MoQ SETUP. Grouped so the shared accept loop builds a [`Request`] from named fields /// rather than a wide tuple. @@ -1658,7 +1633,8 @@ pub struct Request { identity: Option, /// The link facts the transport could see. link: Link, - kind: RequestKind, + /// The paused moq-net handshake; the same type whatever the transport. + handshake: Box, /// The connection under a QUIC or WebTransport request, for [`Request::quic`]. #[cfg(feature = "quic")] quic: Option, @@ -1678,63 +1654,6 @@ pub enum Reject { App(u16), } -/// Delegate a read-only call to the inner [`moq_net::server::Handshake`], whatever the transport. -macro_rules! request_ref { - ($self:expr, $r:ident => $body:expr) => { - match &$self.kind { - #[cfg(feature = "quic")] - RequestKind::Quic($r) => $body, - #[cfg(feature = "iroh")] - RequestKind::Iroh($r) => $body, - #[cfg(any( - feature = "tcp", - all(feature = "uds", unix), - feature = "websocket", - feature = "webrtc" - ))] - RequestKind::Qmux($r) => $body, - } - }; -} - -/// Delegate a consuming call whose arms all yield the same type (e.g. `ok`, `close`). -macro_rules! request_into { - ($kind:expr, $r:ident => $body:expr) => { - match $kind { - #[cfg(feature = "quic")] - RequestKind::Quic($r) => $body, - #[cfg(feature = "iroh")] - RequestKind::Iroh($r) => $body, - #[cfg(any( - feature = "tcp", - all(feature = "uds", unix), - feature = "websocket", - feature = "webrtc" - ))] - RequestKind::Qmux($r) => $body, - } - }; -} - -/// Delegate a consuming builder call, rebuilding the same variant from the returned request. -macro_rules! request_map { - ($kind:expr, $r:ident => $body:expr) => { - match $kind { - #[cfg(feature = "quic")] - RequestKind::Quic($r) => RequestKind::Quic(Box::new($body)), - #[cfg(feature = "iroh")] - RequestKind::Iroh($r) => RequestKind::Iroh(Box::new($body)), - #[cfg(any( - feature = "tcp", - all(feature = "uds", unix), - feature = "websocket", - feature = "webrtc" - ))] - RequestKind::Qmux($r) => RequestKind::Qmux(Box::new($body)), - } - }; -} - impl Request { /// Reject the session. The transport is already accepted, so this closes the /// just-established MoQ session rather than answering the transport handshake: @@ -1744,19 +1663,19 @@ impl Request { Reject::Unauthorized | Reject::Forbidden => moq_net::Error::Unauthorized, Reject::App(code) => moq_net::Error::App(code), }; - request_into!(self.kind, request => request.close(err)); + self.handshake.close(err); Ok(()) } /// Publish the given origin to the session. pub fn with_publisher(mut self, publish: impl moq_net::Consume) -> Self { - self.kind = request_map!(self.kind, request => request.with_publisher(publish)); + *self.handshake = (*self.handshake).with_publisher(publish); self } /// Subscribe to the given origin from the session. pub fn with_subscriber(mut self, subscribe: moq_net::origin::Producer) -> Self { - self.kind = request_map!(self.kind, request => request.with_subscriber(subscribe)); + *self.handshake = (*self.handshake).with_subscriber(subscribe); self } @@ -1764,19 +1683,19 @@ impl Request { /// [`moq_net::server::Handshake::with_peer_hop`]. Derive it from [`Self::peer_identity`], /// never from something coarser. pub fn with_peer_hop(mut self, hop: moq_net::Hop) -> Self { - self.kind = request_map!(self.kind, request => request.with_peer_hop(hop)); + *self.handshake = (*self.handshake).with_peer_hop(hop); self } /// Attach a per-connection [`moq_net::stats::Session`] context to this session. pub fn with_stats(mut self, stats: moq_net::stats::Session) -> Self { - self.kind = request_map!(self.kind, request => request.with_stats(stats)); + *self.handshake = (*self.handshake).with_stats(stats); self } /// Set request caps after authorizing the peer and before answering SETUP. pub fn with_limits(mut self, limits: moq_net::session::Limits) -> Self { - self.kind = request_map!(self.kind, request => request.with_limits(limits)); + *self.handshake = (*self.handshake).with_limits(limits); self } @@ -1807,21 +1726,27 @@ impl Request { /// otherwise hold open. pub async fn ok(self) -> crate::Result<(Session, tokio::task::JoinHandle>)> { let Request { - kind, deadline, active, .. + handshake, + deadline, + active, + .. } = self; let ok = async { - Ok(request_into!(kind, request => { - let (session, driver) = request.ok().await?; - use tracing::Instrument; - let driver = tokio::spawn(async move { + let (session, driver) = handshake.ok().await?; + use tracing::Instrument; + let driver = tokio::spawn( + async move { crate::panic::isolate(async move { let _active = active; moq_net::time::run(driver).await; Ok(()) - }).await.unwrap_or(Err(Error::Panic)) - }.instrument(tracing::Span::current())); - (session, driver) - })) + }) + .await + .unwrap_or(Err(Error::Panic)) + } + .instrument(tracing::Span::current()), + ); + Ok((session, driver)) }; Deadline::bound(deadline.as_ref(), ok).await } @@ -1861,7 +1786,7 @@ impl Request { // An empty SETUP path means the client advertised none, so fall back to the // request URL. URL-carrying bindings are the ones that must not send a path at // all, so this never discards a path the client meant us to use. - let setup = request_ref!(self, r => r.path()); + let setup = self.handshake.path(); let path = if setup.is_empty() { self.url.as_ref().map(Url::path).unwrap_or("") } else { @@ -1874,7 +1799,7 @@ impl Request { /// /// Query values can contain credentials. Avoid logging this value. pub fn query(&self) -> Option<&str> { - let setup = request_ref!(self, r => r.path()); + let setup = self.handshake.path(); if setup.is_empty() { self.url.as_ref().and_then(Url::query) } else { @@ -1908,12 +1833,12 @@ impl Request { /// Available on every transport. Use it to reject a token that lacks the scope for /// the client's intended direction. pub fn role(&self) -> Option { - request_ref!(self, r => r.role()) + self.handshake.role() } /// Answer the peer's tokens before accepting and starting its protocol driver. pub fn auth(&self) -> moq_net::auth::Handle { - request_ref!(self, r => r.auth()) + self.handshake.auth() } /// The Hop ID the peer declared in its SETUP (moq-lite-05+). @@ -1924,7 +1849,7 @@ impl Request { /// Self-declared, so treat it as a correlation hint rather than an /// authenticated identity: authorize on the token or client certificate. pub fn peer_hop(&self) -> Option { - request_ref!(self, r => r.peer_hop()) + self.handshake.peer_hop() } /// The credential a moq-transport client presented in its SETUP's `AUTHORIZATION @@ -1932,7 +1857,7 @@ impl Request { /// /// Like [`query`](Self::query), it can hold a credential. Avoid logging it. pub fn token(&self) -> Option<&moq_net::setup::Token> { - request_ref!(self, r => r.token()) + self.handshake.token() } /// The client certificate chain the peer presented, if any, validated diff --git a/rs/moq-uring/MIGRATE.md b/rs/moq-uring/MIGRATE.md index 8ff58a08bce..a7a01f45e60 100644 --- a/rs/moq-uring/MIGRATE.md +++ b/rs/moq-uring/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Wrap sessions in `transport::Local` (#6176) + +moq-net dropped `connect_lite`, `accept_lite`, and `accept_request_lite`, and every session now needs a `Send` transport. +Wrap the worker's session as `transport::Local::new(transport::Session::new(conn, binding))` and call `connect`, `accept`, or `accept_request`. + ### Timers run on moq-clock The worker is a moq-clock backend, so its deadlines are `moq_clock::Instant`s instead of `std::time::Instant`s. diff --git a/rs/moq-uring/README.md b/rs/moq-uring/README.md index ef313719d76..0c4337111e3 100644 --- a/rs/moq-uring/README.md +++ b/rs/moq-uring/README.md @@ -32,12 +32,11 @@ local (`!Send`) task set, and the UDP sockets bound through it. over the same adapter via `web-transport-proto`. `quic::web::Session` is a raw or web transport (`Session::raw`). Wrap it with `transport::Session::new(session, binding)`, using `moq_net::transport::Binding::Uri` - for WebTransport or `InBand` for raw QUIC, before `connect_lite`/`accept_lite`, which return - the session and its driver; poll the driver or await it inside a - `Handle::spawn` task to run it on the worker. moq-transport and the - SETUP-negotiated moq-lite versions box their futures with a native `Send` - bound, so wrap the session once more in `transport::Local`, which checks - the thread on every use, before `accept_request`/`connect`. Web mode maps stream and close codes through the + for WebTransport or `InBand` for raw QUIC. moq-net boxes the transport with a + native `Send` bound, so wrap the session once more in `transport::Local`, which + checks the thread on every use, before `connect`/`accept`. They 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 diff --git a/rs/moq-uring/benches/session_lite.rs b/rs/moq-uring/benches/session_lite.rs index e59725e755d..005a71fd3ab 100644 --- a/rs/moq-uring/benches/session_lite.rs +++ b/rs/moq-uring/benches/session_lite.rs @@ -140,15 +140,15 @@ mod linux { .expect("quic accept"); let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) - .accept_lite( + .accept( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("accept_lite"); + .expect("accept"); let _ = server_handle.run(driver).await; session.closed().await; }); @@ -159,15 +159,15 @@ mod linux { let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); let (session, driver) = moq_net::Client::new() .with_subscriber(sub_origin.clone()) - .connect_lite( + .connect( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("connect_lite"); + .expect("connect"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await; diff --git a/rs/moq-uring/src/lib.rs b/rs/moq-uring/src/lib.rs index 9ba152d82b7..b4a3f96065d 100644 --- a/rs/moq-uring/src/lib.rs +++ b/rs/moq-uring/src/lib.rs @@ -14,10 +14,10 @@ //! //! [`quic`] stacks a sans-IO QUIC stack on that path: a [`quic::Endpoint`] //! serves many connections on one socket (demuxed by connection id, dials -//! included), each a [`quic::Connection`] wrapped by [`transport::Session`], -//! so `moq_net::Client::connect_lite` and `Server::accept_lite` run real -//! moq-lite sessions on the worker, and [`transport::Local`] lets the -//! `Send`-bounded moq-transport paths run there too. The socket is the identity: whatever is +//! included), each a [`quic::Connection`] wrapped by [`transport::Session`] +//! and [`transport::Local`], which meets moq-net's `Send` bound by checking the +//! thread on every use, so `moq_net::Client::connect` and `Server::accept` run +//! real MoQ sessions on the worker. The socket is the identity: whatever is //! built on it runs on the worker that adopted it, and a socket adopted as a //! member of a steered reuseport group ([`udp::Bound`]) issues connection ids //! that steer back to it. [`Handle::run`] supplies time and schedules driver diff --git a/rs/moq-uring/src/quic/client.rs b/rs/moq-uring/src/quic/client.rs index af87e4597fa..0417298723a 100644 --- a/rs/moq-uring/src/quic/client.rs +++ b/rs/moq-uring/src/quic/client.rs @@ -54,8 +54,8 @@ impl Config { /// [`connect`](super::Endpoint::connect) through it. The connection's driver /// runs as a task on the worker that adopted `socket`. Wrap the returned /// [`Connection`] with [`crate::transport::Session::new`] and -/// [`moq_net::transport::Binding::InBand`] to hand it to -/// `moq_net::Client::connect_lite`, or use the stream API directly. +/// [`moq_net::transport::Binding::InBand`], then [`crate::transport::Local`], to +/// hand it to `moq_net::Client::connect`, or use the stream API directly. pub async fn connect(socket: udp::Socket, config: &Config) -> Result { let endpoint = super::Endpoint::new(socket, super::endpoint::Config::default())?; endpoint.connect(config).await diff --git a/rs/moq-uring/src/quic/mod.rs b/rs/moq-uring/src/quic/mod.rs index 4974ad785fa..675335908b1 100644 --- a/rs/moq-uring/src/quic/mod.rs +++ b/rs/moq-uring/src/quic/mod.rs @@ -7,9 +7,9 @@ //! connection's timeout, and wakes stream waiters. The returned //! [`Connection`] implements [`web_transport_trait::poll`], so //! wrapping it with `crate::transport::Session::new` and -//! [`moq_net::transport::Binding::InBand`] lets -//! `moq_net::Client::connect_lite` / `Server::accept_lite` run real moq-lite -//! sessions on the worker; everything is `Rc`-shared and `!Send` by design. +//! [`moq_net::transport::Binding::InBand`], then `crate::transport::Local`, lets +//! `moq_net::Client::connect` / `Server::accept` run real MoQ sessions on the +//! worker; everything is `Rc`-shared and `!Send` by design. //! //! An [`Endpoint`] serves many connections on one socket, demuxed by //! connection id; [`client::connect`] and [`server::accept`] are its diff --git a/rs/moq-uring/src/transport/local.rs b/rs/moq-uring/src/transport/local.rs index 97cd2372a09..e8f9a50ceb9 100644 --- a/rs/moq-uring/src/transport/local.rs +++ b/rs/moq-uring/src/transport/local.rs @@ -5,13 +5,12 @@ use std::task::{Context, Poll}; /// A worker's transport, usable where moq-net wants a `Send` one. /// -/// moq-transport boxes its futures with a native `Send` bound, which a worker's +/// moq-net boxes every transport with a native `Send` bound, which a worker's /// `!Send` [`Session`](super::Session) cannot meet, so wrap the session in this /// before `Server::accept_request` or `Client::connect`. The session, its /// streams, and its send groups never leave the worker that polls them, so /// this checks the thread on every use and panics anywhere else instead of -/// proving it at compile time. moq-lite needs none of it: hand the session to -/// `accept_request_lite` or `connect_lite` directly. +/// proving it at compile time. #[derive(Clone)] pub struct Local(SendWrapper); @@ -102,6 +101,14 @@ impl poll::Session for Local { fn stats(&self) -> impl moq::Stats { self.0.stats() } + + fn wire(&self) -> Option { + self.0.wire() + } + + fn probe(&self) -> Option { + self.0.probe() + } } impl poll::SendStream for Local { @@ -172,7 +179,7 @@ mod tests { use super::*; use moq_net::transport::memory; - /// The point of the wrapper: a worker's `!Send` session meets moq-transport's bounds. + /// The point of the wrapper: a worker's `!Send` session meets moq-net's bounds. #[cfg(feature = "quic")] #[test] fn meets_the_boxed_driver_bounds() { diff --git a/rs/moq-uring/tests/integration/receive_credit.rs b/rs/moq-uring/tests/integration/receive_credit.rs index b3d1759681f..08a3d44272d 100644 --- a/rs/moq-uring/tests/integration/receive_credit.rs +++ b/rs/moq-uring/tests/integration/receive_credit.rs @@ -120,12 +120,15 @@ fn partial_frame_gives_way() { .expect("quic accept"); let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) - .accept_lite( + .accept( moq_net::time::Instant::now(), - moq_uring::transport::Session::new(quic::web::Session::raw(conn), moq_net::transport::Binding::InBand), + moq_uring::transport::Local::new(moq_uring::transport::Session::new( + quic::web::Session::raw(conn), + moq_net::transport::Binding::InBand, + )), ) .await - .expect("accept_lite"); + .expect("accept"); let _ = server_handle.run(driver).await; session.closed().await; }); @@ -136,15 +139,15 @@ fn partial_frame_gives_way() { let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); let (session, driver) = moq_net::Client::new() .with_subscriber(sub.clone()) - .connect_lite( + .connect( moq_net::time::Instant::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("connect_lite"); + .expect("connect"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await; @@ -292,9 +295,12 @@ fn javascript_partial_frame_gives_way() { .expect("respond"); let (session, driver) = moq_net::Server::new() .with_subscriber(sub.clone()) - .accept_lite( + .accept( moq_net::time::Instant::now(), - moq_uring::transport::Session::new(session, moq_net::transport::Binding::Uri), + moq_uring::transport::Local::new(moq_uring::transport::Session::new( + session, + moq_net::transport::Binding::Uri, + )), ) .await .expect("accept"); diff --git a/rs/moq-uring/tests/integration/session.rs b/rs/moq-uring/tests/integration/session.rs index eceb3bd7ca3..79f1f9b0f1e 100644 --- a/rs/moq-uring/tests/integration/session.rs +++ b/rs/moq-uring/tests/integration/session.rs @@ -2,7 +2,7 @@ //! //! A publisher and a subscriber session run on one worker (two sockets, two //! [`moq_uring::quic::Connection`]s over loopback), negotiated end to end by -//! `moq_net::Server::accept_lite` / `Client::connect_lite`, with a broadcast, +//! `moq_net::Server::accept` / `Client::connect`, with a broadcast, //! a track, and a frame flowing through the model. //! //! Origin drivers run through Tokio's explicit-time adapter while session @@ -93,12 +93,15 @@ fn lite_session_over_the_worker() { .expect("quic accept"); let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) - .accept_lite( + .accept( moq_net::time::now(), - moq_uring::transport::Session::new(quic::web::Session::raw(conn), moq_net::transport::Binding::InBand), + moq_uring::transport::Local::new(moq_uring::transport::Session::new( + quic::web::Session::raw(conn), + moq_net::transport::Binding::InBand, + )), ) .await - .expect("accept_lite"); + .expect("accept"); let _ = server_handle.run(driver).await; // Serve until the client walks away. session.closed().await; @@ -116,15 +119,15 @@ fn lite_session_over_the_worker() { let (session, driver) = moq_net::Client::new() .with_subscriber(sub.clone()) - .connect_lite( + .connect( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("connect_lite"); + .expect("connect"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await; @@ -228,15 +231,15 @@ fn two_lite_sessions_share_the_server_socket() { server_handle.spawn(async move { let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) - .accept_lite( + .accept( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("accept_lite"); + .expect("accept"); let _ = session_handle.run(driver).await; session.closed().await; }); @@ -257,15 +260,15 @@ fn two_lite_sessions_share_the_server_socket() { let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); let (session, driver) = moq_net::Client::new() .with_subscriber(sub.clone()) - .connect_lite( + .connect( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("connect_lite"); + .expect("connect"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await; diff --git a/rs/moq-uring/tests/integration/web.rs b/rs/moq-uring/tests/integration/web.rs index b334a60a880..a58e13ab733 100644 --- a/rs/moq-uring/tests/integration/web.rs +++ b/rs/moq-uring/tests/integration/web.rs @@ -284,12 +284,12 @@ fn lite_session_over_webtransport() { let (moq, session_driver) = moq_net::Client::new() .with_subscriber(sub_origin.clone()) - .connect_lite( + .connect( moq_net::time::now(), moq_tokio::transport::Session::new(session, moq_net::transport::Binding::Uri), ) .await - .expect("connect_lite"); + .expect("connect"); tokio::spawn(moq_net::time::run(session_driver)); let bc = { @@ -335,12 +335,15 @@ fn lite_session_over_webtransport() { let (session, driver) = moq_net::Server::new() .with_publisher(&serve_origin) - .accept_lite( + .accept( moq_net::time::now(), - moq_uring::transport::Session::new(session, moq_net::transport::Binding::Uri), + moq_uring::transport::Local::new(moq_uring::transport::Session::new( + session, + moq_net::transport::Binding::Uri, + )), ) .await - .expect("accept_lite"); + .expect("accept"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await;