diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index b3afcc93296..1961deea5b5 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -22,7 +22,7 @@ What you use it for, beyond what the concept page already describes: - **Discovery follows what you watch.** A client session asks the relay for announcements only under the prefixes its subscribe origin's readers watch, with `announced()`, `routed_broadcast`, or `announced_broadcast`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A broadcast resolved through a discovered route keeps its path watched until dropped. `request_broadcast` watches nothing, so a path nobody watches is unroutable. A server session, or a client fed into an origin handle marked `with_link(origin::Link::Peer)` or `Link::Upstream`, asks for everything it may, so a relay discovers everything. An HTTP gateway (the moq-hls `Server`, the moq-rtc WHEP router) watches its whole scope while it lives, so the paths its callers name never open a request each. moq-lite-03/04 sessions, whose announcements mark no initial set, still ask for everything. - **One track per name.** Concurrent subscriptions share one request and producer. The application owns sequence numbers: the broadcast keeps nothing of an ended track, so a producer that replaces it starts at 0. A name always means the same content, so the replacement continues past its predecessor's groups with `create_group` at an explicit sequence. Readers of an aborted track resume onto the replacement only when the broadcast announces an epoch; otherwise their subscription ends with the error. - **End content by finishing tracks.** A broadcast has no finished state. When the input ends, finish each track; the broadcast stays announced and keeps serving them until you close it or drop its last producer. How long to keep it up for late readers is yours to decide: `moq_mux::Linger` holds a hang broadcast for a fixed window. -- **Publish only while someone is watching.** `demand()` on a track, group, or broadcast says whether a subscriber is attached, which is how capture and transcode skip work nobody asked for. Holding a broadcast or track consumer is not demand: a track consumer requests nothing until it queries, subscribes, or fetches, and those count until dropped. A shared fetch stays up until its last reader leaves. +- **Publish only while someone is watching.** `demand()` on a track, group, or broadcast says whether a subscriber is attached, which is how capture and transcode skip work nobody asked for. Holding a broadcast or track consumer is not demand: a track consumer requests nothing until it queries, subscribes, or fetches, and those count until dropped. A shared fetch stays up until its last reader leaves. Demand rolls up: a frame being read keeps its group wanted, a group a subscription or fetch handed out keeps its track wanted until read to its end, and any wanted track keeps its broadcast wanted. - **Use the model without a runtime.** Disable default features on `moq-net` to use its `poll_*` APIs and origin driver with caller-supplied time. The default `async` feature adds async helpers, `time::run`, and both wire sessions. - **You drive the session, or `moq-tokio` does.** `connect` and `accept` return a session plus a driver that never reads the clock itself. `moq_net::time::run` polls it on tokio or in the browser. `moq-tokio` and `moq-wasm` hide that. A custom transport implements `moq_net::transport::poll`, and `moq_net::transport::memory::pair` links two sessions in one process with no sockets, for tests or in-process links. - **A session bounds what its peer can hold.** Each request holds a stream, so the transport's stream credit bounds subscriptions and FETCHes. `session::Limits` (via `Client::with_limits` and `Server::with_limits`) bounds the announcements per session. The default suits a relay mesh, so lower it for untrusted peers. Past it, the session closes with `TOO_MANY_REQUESTS`. Peer-declared lengths are capped before they are buffered. diff --git a/quest/m1/api-audit/wanted.md b/quest/m1/api-audit/wanted.md index 728092c4d1a..4c6825b7a05 100644 --- a/quest/m1/api-audit/wanted.md +++ b/quest/m1/api-audit/wanted.md @@ -15,11 +15,12 @@ Decided 2026-10-25 (maintainer), extending the `wanted` shape that - Yes/no demand returns `bool`; `track::Demand::wanted() -> Option` keeps its meaning. -- Broadcast demand is deleted: Rust's `broadcast::Demand` went in - [#6744](https://github.com/moq-dev/moq/pull/6744), and the `@internal` JS - `Broadcast.Demand` (read only by tests and a bench) goes too. The origin - closes a superseded producer explicitly instead of a publisher watching - demand. +- Rust's `broadcast::Demand`, deleted in + [#6744](https://github.com/moq-dev/moq/pull/6744), is back in the `wanted` + shape for publishers that steer on whether anyone reads a broadcast. The + `@internal` JS `Broadcast.Demand` (read only by tests and a bench) goes. The + origin closes a superseded producer explicitly instead of a publisher + watching demand. - `group::Demand` already says `wanted`. The JS group demand stays internal. - `bandwidth::Producer` gets a `demand()` handle with `wanted() -> bool` and `wanted_changed()`, like track. diff --git a/rs/moq-e2ee/src/tests.rs b/rs/moq-e2ee/src/tests.rs index fba81a5fc01..04525cd1c4c 100644 --- a/rs/moq-e2ee/src/tests.rs +++ b/rs/moq-e2ee/src/tests.rs @@ -836,10 +836,6 @@ fn grouped_failure_wakes_pending_reads_and_releases_demand() { assert!(frame.woken(), "parked frame read was not woken"); assert!(track.woken(), "parked group and datagram reads were not woken"); - assert!( - !demand.wanted().is_some(), - "track demand stayed live after authentication failure" - ); assert!(!bad_demand.wanted(), "failed group demand stayed live"); assert!(!pair.net.is_closed(), "publisher must stay open"); @@ -853,6 +849,11 @@ fn grouped_failure_wakes_pending_reads_and_releases_demand() { !open_demand.wanted(), "sibling group demand stayed live after its woken read" ); + // The sibling's reader held the track wanted until that read released it. + assert!( + !demand.wanted().is_some(), + "track demand stayed live after authentication failure" + ); assert_eq!( track .poll(|waiter| pair.consumer.poll_recv_group(waiter)) @@ -873,7 +874,7 @@ fn grouped_failure_wakes_pending_reads_and_releases_demand() { } #[test] -fn dropping_track_releases_demand_while_group_is_held() { +fn a_held_group_keeps_demand_after_the_track_drops() { let mut pair = pair("video"); let demand = pair.net.demand(); let mut open = pair.producer.append_group().unwrap(); @@ -881,12 +882,15 @@ fn dropping_track_releases_demand_while_group_is_held() { let mut group = recv_group(&mut pair.consumer); drop(pair.consumer); - assert!(!demand.wanted().is_some(), "a held group kept the track subscription"); + assert!(demand.wanted().is_some(), "a held group keeps its track wanted"); // The held group is not failed by the unsubscribe. assert_eq!(read_frame(&mut group).plaintext, &b"ok"[..]); open.write_frame(ms(2), b"next").unwrap(); assert_eq!(read_frame(&mut group).plaintext, &b"next"[..]); + + drop(group); + assert!(demand.wanted().is_none(), "the last reader leaving releases the track"); } #[test] diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index 869f2527551..c1f3db3b832 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -143,6 +143,10 @@ required-features = ["async"] name = "tracks" harness = false +[[bench]] +name = "demand" +harness = false + [[bench]] name = "tail" harness = false diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index b110770ddcc..6bcfb713078 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,22 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `broadcast::Demand` reads like `track::Demand` + +Replace `is_used()` with `wanted()`, and `used()`, `unused()`, `poll_used`, and `poll_unused` with a loop over `wanted_changed()` (or `poll_wanted_changed`), which returns the new value each time it changes and errors once the broadcast closes. +`broadcast::Consumer::demand()` is gone: call `demand()` on the broadcast's `Producer`. + +```rust +// Before +demand.used().await?; +demand.unused().await?; +// After +while !demand.wanted_changed().await? {} +while demand.wanted_changed().await? {} +``` + +A group a subscription or fetch handed out now keeps its track wanted until the group finishes and a reader reads its end (a range cap is not the end): until then `track::Demand::wanted` stays `Some` and `track::Producer::abort_unused` declines. Drop the group, or read it to its end, to let the demand go. + ### Paths fit moq-transport's 4096-byte namespace ([#6927](https://github.com/moq-dev/moq/pull/6927)) `Path::MAX_BYTES` is 4096, moq-transport's namespace limit, so every path fits every wire and building a longer one fails with `path::Invalid::TooLong`. Keep broadcast paths within 4096 bytes. @@ -78,7 +94,6 @@ A sender that relied on `None` to hold its rate while unwatched should check its These had no caller outside tests; replace them as follows. -- `broadcast::Demand` and both `broadcast::{Producer, Consumer}::demand()`: watch each track's `track::Demand` instead. - `goaway::Consumer::{peek, recv}`: poll it, `goaway.poll(&kio::Waiter::noop())` to peek or `kio::wait(|waiter| goaway.poll(waiter)).await` to wait. - `bandwidth::Producer::{abort, closed}`: drop the last producer to close it. - `Version::{names, from_alpn, is_ietf}`: use `Versions::all().iter()` or `Version::from_code`, match `alpn()` yourself, and `!is_lite()`. diff --git a/rs/moq-net/benches/demand.rs b/rs/moq-net/benches/demand.rs new file mode 100644 index 00000000000..6b5d7c6f45f --- /dev/null +++ b/rs/moq-net/benches/demand.rs @@ -0,0 +1,78 @@ +//! What a broadcast's demand costs as it fans out. +//! +//! - edge: a reader subscribes to one track through the broadcast and leaves, and every +//! [`broadcast::Demand`] handle reads both edges. Swept over the tracks the broadcast +//! holds and the handles watching it, so a cost that grows with the table rather than +//! the touched track shows up as a slope. +//! - churn: a short-lived track comes and goes while handles that never poll watch the +//! broadcast, the insert each handle's inbox absorbs. Swept the same way, with the live +//! tracks one short of a power of two so they fill an inbox's capacity but one slot. +//! +//! Run with `cargo bench -p moq-net --bench demand`. + +use std::task::Poll; + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; +use moq_net::broadcast; + +/// `(tracks, handles)` shapes: the broadcast's tracks and the demand handles watching it. +const SHAPES: [(usize, usize); 5] = [(1, 1), (100, 1), (10_000, 1), (1, 100), (10_000, 100)]; + +/// `(tracks, handles)` shapes for churn: each `tracks` is one under the inbox capacity it grows to. +const CHURN_SHAPES: [(usize, usize); 5] = [(3, 1), (127, 1), (16_383, 1), (3, 100), (16_383, 100)]; + +fn changed(demand: &mut broadcast::Demand, waiter: &kio::Waiter) -> Option { + match demand.poll_wanted_changed(waiter) { + Poll::Ready(wanted) => Some(wanted.expect("broadcast closed")), + Poll::Pending => None, + } +} + +fn bench_edge(c: &mut Criterion) { + let mut group = c.benchmark_group("demand/edge"); + for (tracks, handles) in SHAPES { + let id = BenchmarkId::from_parameter(format!("{tracks}t_{handles}h")); + group.bench_function(id, |b| { + let producer = broadcast::Info::new().produce(); + let _tracks: Vec<_> = (0..tracks) + .map(|i| producer.create_track(format!("t{i}"), None).unwrap()) + .collect(); + let consumer = producer.consume(); + let waiter = kio::Waiter::noop(); + let mut demands: Vec<_> = (0..handles).map(|_| producer.demand()).collect(); + for demand in &mut demands { + assert_eq!(changed(demand, &waiter), Some(false)); + } + b.iter(|| { + let reader = consumer.track("t0").unwrap().subscribe(None); + for demand in &mut demands { + assert_eq!(changed(demand, &waiter), Some(true)); + } + drop(reader); + for demand in &mut demands { + assert_eq!(changed(demand, &waiter), Some(false)); + } + }); + }); + } + group.finish(); +} + +fn bench_churn(c: &mut Criterion) { + let mut group = c.benchmark_group("demand/churn"); + for (tracks, handles) in CHURN_SHAPES { + let id = BenchmarkId::from_parameter(format!("{tracks}t_{handles}h")); + group.bench_function(id, |b| { + let producer = broadcast::Info::new().produce(); + let _demands: Vec<_> = (0..handles).map(|_| producer.demand()).collect(); + let _tracks: Vec<_> = (0..tracks) + .map(|i| producer.create_track(format!("t{i}"), None).unwrap()) + .collect(); + b.iter(|| drop(producer.create_track("churn", None).unwrap())); + }); + } + group.finish(); +} + +criterion_group!(benches, bench_edge, bench_churn); +criterion_main!(benches); diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 123d8f294c3..6f4a24c2cbc 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -10,7 +10,10 @@ use crate::{cache, stats, track}; use std::{ collections::HashSet, - sync::Arc, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, task::{Poll, ready}, }; @@ -18,7 +21,7 @@ use crate::Error; use crate::origin::Route; use super::origin_impl::Announcer; -use super::{RequestQueue, WeakCache}; +use super::{RequestQueue, WeakCache, WeakEntry}; #[path = "listing.rs"] mod listing; @@ -120,18 +123,41 @@ struct BroadcastState { // Set once the broadcast ends: `Producer::close()` or the last producer-side // handle dropping. Every lookup after it answers `Unroutable`. closing: bool, + + // The tracks cached since each [`Demand`] handle last looked, by handle, so a + // handle watches each track once rather than rescanning the table. + demands: slab::Slab>, } impl BroadcastState { /// Insert a track weak handle into the lookup, returning an error if a live /// track already holds the name. A closed entry under the name is reclaimed. fn insert_track(&mut self, weak: track::TrackWeak) -> Result<(), Error> { - match self.tracks.insert(weak.name().clone(), weak) { + match self.cache_track(weak) { Some(_) => Err(Error::Duplicate), None => Ok(()), } } + /// Cache a track under its name and hand it to every [`Demand`], unless a live track + /// already holds the name, which is returned instead. + fn cache_track(&mut self, weak: track::TrackWeak) -> Option { + if let Some(live) = self.tracks.insert(weak.name().clone(), weak.clone()) { + return Some(live); + } + for (_, cached) in &mut self.demands { + // A handle that stops polling must not pin every track it was handed. Headroom + // for as many tracks as survive defers the next scan until the inbox doubles, + // so a full inbox losing one track doesn't rescan it on every insert. + if cached.len() == cached.capacity() { + cached.retain(|track| !track.is_closed()); + cached.reserve(cached.len()); + } + cached.push(weak.clone()); + } + None + } + /// Resolve every name the broadcast never filled, so subscribers waiting on a /// [`track::Info`] that can no longer arrive fail with `err` instead of parking. /// @@ -311,6 +337,11 @@ impl Producer { &self.info } + /// A watch-only handle to whether anyone reads the broadcast. See [`Demand`]. + pub fn demand(&self) -> Demand { + Demand::new(self.alive.token.consume().weak(), self.state.clone()) + } + /// Produce a new track and insert it into the broadcast. /// /// Pass a name and an optional [`track::Info`], so a bare name works: @@ -334,7 +365,7 @@ impl Producer { // Cache it like a served request so concurrent lookups coalesce; a // live same-name entry cannot exist (its presence would have kept // the request from queuing). - let _ = state.tracks.insert(name.clone(), track.weak()); + let _ = state.cache_track(track.weak()); let unlist = Unlist::list(&self.state, &mut state.listing, name, &info); return Ok(track.listed(unlist)); } @@ -772,7 +803,7 @@ impl Requests { let pending = state.requests.remove(&name).expect("popped key must be pending"); // Cache the served track so concurrent lookups coalesce onto it. If a live track already // holds the name (a publish raced the request), `insert` keeps it rather than shadowing it. - let _ = state.tracks.insert(name, pending.weak()); + let _ = state.cache_track(pending.weak()); // Attribute the served track to this broadcast's ingress scope (no-op untagged). Poll::Ready(Ok(pending.claim().with_stats(self.stats.clone()))) } @@ -1093,6 +1124,151 @@ impl super::WeakEntry for WeakConsumer { } } +/// A cloneable, watch-only handle to whether anyone reads a broadcast. +/// +/// Obtained from [`Producer::demand`]. The broadcast is wanted while any of its tracks is +/// ([`track::Demand::wanted`] is `Some`) or a track request waits for a [`Requests`] +/// handler, so a publisher can follow its readers without holding every track's producer. +/// It's a weak handle: it neither keeps the broadcast alive nor counts as demand. +/// Each clone watches on its own, so its first +/// [`poll_wanted_changed`](Self::poll_wanted_changed) reports the current value. +pub struct Demand { + alive: kio::ConsumerWeak<()>, + state: kio::Shared, + // This handle's inbox in [`BroadcastState::demands`]. + inbox: usize, + // One watcher per track, each counting its track in `wanted` while the track is wanted, + // so a reader coming or going re-polls only its own track. + watchers: kio::Tasks, + wanted: Arc, + // The value `poll_wanted_changed` last returned, or `None` before it returned any. + prev: Option, +} + +impl Clone for Demand { + fn clone(&self) -> Self { + Self::new(self.alive.clone(), self.state.clone()) + } +} + +impl Demand { + fn new(alive: kio::ConsumerWeak<()>, state: kio::Shared) -> Self { + let inbox = { + let mut state = state.lock(); + let open = state + .tracks + .iter() + .filter(|track| !track.is_closed()) + .cloned() + .collect(); + state.demands.insert(open) + }; + Self { + alive, + state, + inbox, + watchers: kio::Tasks::new(), + wanted: Arc::default(), + prev: None, + } + } + + /// Whether the broadcast is open and any of its tracks is wanted or requested right now. + pub fn wanted(&self) -> bool { + if self.alive.is_closed() { + return false; + } + let state = self.state.read(); + !state.requests.is_empty() || state.tracks.iter().any(|track| track.is_used()) + } + + /// Return [`wanted`](Self::wanted) once it differs from what this handle last returned: + /// immediately on the first call, then on each change. Errors once the broadcast closes. + /// + /// So `while !demand.wanted_changed().await? {}` waits for a reader, and + /// `while demand.wanted_changed().await? {}` waits for the last one to leave. + #[cfg(feature = "async")] + pub async fn wanted_changed(&mut self) -> Result { + kio::wait(|waiter| self.poll_wanted_changed(waiter)).await + } + + /// Poll for [`wanted`](Self::wanted) to differ from what this handle last returned. + pub fn poll_wanted_changed(&mut self, waiter: &kio::Waiter) -> Poll> { + if self.alive.poll_closed(waiter).is_ready() { + return Poll::Ready(Err(Error::Dropped)); + } + let requested = self.poll_cached(waiter); + // Ready only while no track is watched; the parent waker is registered either way. + let _ = self.watchers.poll(waiter); + let wanted = requested || self.wanted.load(Ordering::Relaxed) > 0; + if self.prev == Some(wanted) { + return Poll::Pending; + } + self.prev = Some(wanted); + Poll::Ready(Ok(wanted)) + } + + /// Watch the tracks cached since the last look, returning whether a track request is + /// pending. Parks `waiter` on either changing. + fn poll_cached(&mut self, waiter: &kio::Waiter) -> bool { + loop { + let mut requested = false; + let polled = self.state.poll(waiter, |state| { + requested = !state.requests.is_empty(); + match state.demands[self.inbox].is_empty() { + true => Poll::Pending, + false => Poll::Ready(()), + } + }); + let Poll::Ready(mut state) = polled else { + return requested; + }; + let cached = std::mem::take(&mut state.demands[self.inbox]); + drop(state); + for track in cached { + self.watchers.push(Watcher { + track: track.demand(), + counted: false, + wanted: self.wanted.clone(), + }); + } + } + } +} + +impl Drop for Demand { + fn drop(&mut self) { + self.state.lock().demands.remove(self.inbox); + } +} + +/// Counts one track in its [`Demand`] while the track is wanted. +struct Watcher { + track: track::Demand, + counted: bool, + wanted: Arc, +} + +impl kio::Task for Watcher { + type Output = (); + + fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { + let state = self.track.poll_state(waiter); + let wanted = state == track::DemandState::Active; + if wanted != self.counted { + self.counted = wanted; + match wanted { + true => self.wanted.fetch_add(1, Ordering::Relaxed), + false => self.wanted.fetch_sub(1, Ordering::Relaxed), + }; + } + match state { + track::DemandState::Closed => Poll::Ready(()), + _ => Poll::Pending, + } + } +} + #[cfg(test)] #[expect(missing_docs, reason = "test-only assertion helpers")] impl Consumer { @@ -1112,7 +1288,6 @@ impl Consumer { /// The tracks still open on the broadcast, whether or not anyone reads them. pub fn open_tracks(&self) -> usize { - use super::WeakEntry; self.state .read() .tracks @@ -1325,6 +1500,165 @@ mod test { assert_eq!(producer.unique_name(""), "12"); } + /// Records whether it was woken. + #[derive(Default)] + struct Flag(std::sync::atomic::AtomicBool); + + impl std::task::Wake for Flag { + fn wake(self: Arc) { + self.0.store(true, Ordering::Relaxed); + } + } + + impl Flag { + fn waiter(self: &Arc) -> kio::Waiter { + kio::Waiter::new(self.clone().into()) + } + + fn take(&self) -> bool { + self.0.swap(false, Ordering::Relaxed) + } + } + + /// Poll `demand` for a change with `waiter`, without waiting. + fn wanted_changed(demand: &mut Demand, waiter: &kio::Waiter) -> Option { + match demand.poll_wanted_changed(waiter) { + Poll::Ready(res) => Some(res.unwrap()), + Poll::Pending => None, + } + } + + /// Any track's demand is the broadcast's, whether the track existed before the handle or + /// came after it, and a reader arriving straight through a track producer, which never + /// touches the broadcast's state, still wakes it. It stays wanted until the last wanted + /// track goes. + #[test] + fn demand_follows_every_track() { + let producer = Info::new().produce(); + let video = producer.create_track("video", None).unwrap(); + let mut demand = producer.demand(); + let audio = producer.create_track("audio", None).unwrap(); + let flag = Arc::new(Flag::default()); + let waiter = flag.waiter(); + assert_eq!( + wanted_changed(&mut demand, &waiter), + Some(false), + "the first call reports" + ); + + let _bare = producer.consume().track("audio").unwrap(); + assert_eq!( + wanted_changed(&mut demand, &waiter), + None, + "a bare handle is not demand" + ); + + flag.take(); + let reading_audio = audio.subscribe(None); + assert!(flag.take(), "a reader wakes the parked poll"); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(true)); + assert!(demand.wanted()); + + let reading_video = video.subscribe(None); + assert_eq!(wanted_changed(&mut demand, &waiter), None, "already wanted"); + drop(reading_audio); + assert_eq!(wanted_changed(&mut demand, &waiter), None, "video is still read"); + flag.take(); + drop(reading_video); + assert!(flag.take(), "the last reader leaving wakes the parked poll"); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(false)); + assert!(!demand.wanted()); + + // A clone reports the current value first, whatever the original last returned. + let _reading = video.subscribe(None); + assert_eq!(wanted_changed(&mut demand.clone(), &waiter), Some(true)); + } + + /// A request waiting for a handler is demand, and handing it out moves that demand onto + /// the track it became without a gap, until its subscriber leaves. + #[test] + fn a_pending_request_is_demand_until_its_reader_leaves() { + let producer = Info::new().produce(); + let mut requests = producer.requests(); + let mut demand = producer.demand(); + let waiter = kio::Waiter::noop(); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(false)); + + let subscribing = producer.consume().track("video").unwrap().subscribe(None); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(true), "queued"); + + let Poll::Ready(Ok(request)) = requests.poll_requested_track(&waiter) else { + panic!("the request was not queued"); + }; + assert_eq!(wanted_changed(&mut demand, &waiter), None, "handed out, still read"); + let _track = request.accept(None); + assert_eq!(wanted_changed(&mut demand, &waiter), None, "accepted, still read"); + + drop(subscribing); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(false)); + } + + /// A closed track is never demand, even while a reader lingers to drain it. + #[test] + fn a_closed_track_is_not_demand() { + let producer = Info::new().produce(); + let video = producer.create_track("video", None).unwrap(); + let mut demand = producer.demand(); + let _reading = video.subscribe(None); + let waiter = kio::Waiter::noop(); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(true)); + + video.abort(Error::Cancel).unwrap(); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(false)); + assert!(!demand.wanted()); + } + + /// The handle reports the broadcast closing, wanted or not. + #[test] + fn demand_errors_once_closed() { + let producer = Info::new().produce(); + let video = producer.create_track("video", None).unwrap(); + let _reading = video.subscribe(None); + let mut demand = producer.demand(); + producer.close(); + assert!(!demand.wanted()); + assert!(matches!( + demand.poll_wanted_changed(&kio::Waiter::noop()), + Poll::Ready(Err(Error::Dropped)) + )); + } + + /// A reader of a frame keeps its group wanted, the group its track, and the track the + /// broadcast, after every handle above it is gone. + #[test] + fn a_frame_read_keeps_the_broadcast_wanted() { + let producer = Info::new().produce(); + let video = producer.create_track("video", track::Info::timed()).unwrap(); + let mut demand = producer.demand(); + let waiter = kio::Waiter::noop(); + let mut subscriber = video.subscribe(None); + let mut writing = video.append_group().unwrap(); + writing.write_frame(crate::Timestamp::ZERO, &b"key"[..]).unwrap(); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(true)); + + let mut group = subscriber.assert_group(); + drop(subscriber); + assert_eq!(wanted_changed(&mut demand, &waiter), None, "the group is read"); + assert!(video.demand().wanted().is_some()); + + let Poll::Ready(Ok(Some(frame))) = group.poll_next_frame(&waiter) else { + panic!("the frame was written"); + }; + drop(group); + assert!(writing.demand().wanted(), "the frame's reader reads the group"); + assert_eq!(wanted_changed(&mut demand, &waiter), None, "the frame is read"); + + drop(frame); + assert!(!writing.demand().wanted()); + assert_eq!(video.demand().wanted(), None); + assert_eq!(wanted_changed(&mut demand, &waiter), Some(false)); + } + /// Subscribe and assert the result hasn't resolved yet (it stays pending until /// a publisher accepts). Returns the pending subscription to resolve after accepting. macro_rules! subscribe_pending { diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 96bae34053b..f524af12831 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -2584,6 +2584,9 @@ fn settle(poll: Poll, kio::Ref<'_, GroupState>>>) /// A cloneable, watch-only handle to a group's readers. /// +/// A frame being read counts as a reader of its group, and a group a track subscription +/// or fetch handed out keeps that track wanted until a reader reads its end +/// ([`track::Demand`]). /// Once every handle to a finished group is gone, demand waits end with /// [`Error::Dropped`]; an aborted group's keep its abort reason. Each clone watches on /// its own, so its first [`poll_wanted_changed`](Self::poll_wanted_changed) reports the @@ -2921,11 +2924,15 @@ pub struct Consumer { recover: Option>, } -/// Subscriber-specific policy for expiring a group after it was handed out. +/// Subscriber-specific policy for a group after it was handed out: when it expires, and +/// the track demand its readers hold until they read its end. /// /// Unwind safe so the [`Consumer`] holding it is, and so the published /// [`track::Fetching`] that holds a consumer stays unwind safe too. pub(super) trait Expiry: Send + Sync + std::panic::UnwindSafe + std::panic::RefUnwindSafe { + /// A reader read the group's end, so it no longer keeps the track wanted. + fn ended(&self) {} + /// Return whether the group is stale, registering `waiter` for anything that /// could change the answer while the group remains live. /// A logical reader supplies its current budget after the original copy is gone. @@ -3329,7 +3336,8 @@ impl Consumer { /// passed the [`Self::with_range`] cap. pub fn poll_next_frame(&mut self, waiter: &kio::Waiter) -> Poll>> { if self.recover.is_none() { - return self.poll_next_frame_once(waiter); + let res = self.poll_next_frame_once(waiter); + return self.ended_on(res); } let res = self.poll_resumed(waiter, Self::poll_next_frame_once); match (res, &self.recover) { @@ -3337,7 +3345,25 @@ impl Consumer { (Poll::Ready(Ok(Some(frame))), Some(recover)) => Poll::Ready(Ok(Some( frame.with_recover((**recover).clone(), self.cursor.index as u64 - 1), ))), - (res, _) => res, + (res, _) => self.ended_on(res), + } + } + + /// Pass a read through, releasing the track demand once it reached the group's end. + fn ended_on(&self, res: Poll>>) -> Poll>> { + if let Poll::Ready(Ok(None)) = res { + self.release_if_finished(); + } + res + } + + /// Release the track demand this group holds once it finished, since the track owes it + /// nothing more. A range cap ends only this reader, and can be raised, so it keeps it. + fn release_if_finished(&self) { + if let Some(expiry) = &self.expiry + && self.cursor.state.read().fin.is_some() + { + expiry.ended(); } } @@ -3371,10 +3397,11 @@ impl Consumer { waiter: &kio::Waiter, take: impl Fn(&mut Prefetch) -> Option + Copy, ) -> Poll>> { - if self.recover.is_none() { - return self.poll_read_frame_once(waiter, take); - } - self.poll_resumed(waiter, |this, waiter| this.poll_read_frame_once(waiter, take)) + let res = match self.recover.is_none() { + true => self.poll_read_frame_once(waiter, take), + false => self.poll_resumed(waiter, |this, waiter| this.poll_read_frame_once(waiter, take)), + }; + self.ended_on(res) } /// Read the next frame with its relay metadata, for wire egress. @@ -3468,13 +3495,18 @@ impl Consumer { /// An error means the frames this cursor has yet to read are gone (evicted, cut, or /// failed upstream) and no route will continue them. pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll> { - if self.recover.is_none() { - return self.poll_finished_once(waiter); + let res = match self.recover.is_none() { + true => self.poll_finished_once(waiter), + false => self + .poll_resumed(waiter, |this, waiter| { + this.poll_finished_once(waiter).map(|res| res.map(Some)) + }) + .map(|res| res.map(|index| index.expect("finished with an index"))), + }; + if let Poll::Ready(Ok(_)) = res { + self.release_if_finished(); } - let res = self.poll_resumed(waiter, |this, waiter| { - this.poll_finished_once(waiter).map(|res| res.map(Some)) - }); - res.map(|res| res.map(|index| index.expect("finished with an index"))) + res } fn poll_finished_once(&mut self, waiter: &kio::Waiter) -> Poll> { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index ceb7479ea23..c583307886e 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3447,8 +3447,8 @@ impl Producer { /// may still observe its final state. `Err(producer)` leaves the track unchanged /// so the caller can continue serving and wait for the next unused wake. /// - /// The abort is the same as [`Self::abort`]: a group consumer that outlived the track's - /// consumers ends with `err`, and the group's writer fails. + /// The abort is the same as [`Self::abort`]. A reader still in a group it was handed + /// counts as a consumer until it reads the group's end. #[cfg_attr( target_pointer_width = "64", expect( @@ -3974,6 +3974,11 @@ impl TrackWeak { !self.state.is_closed() && self.state.is_used() } + /// A [`Demand`] on the cached track, for a broadcast to watch. + pub(super) fn demand(&self) -> Demand { + Demand::new(self.name.clone(), self.state.clone()) + } + /// The readers' aggregate demand and contribution count, or `None` without one. pub(super) fn subscription(&self) -> Option<(Subscription, usize)> { let state = self.state.read(); @@ -4030,7 +4035,9 @@ impl super::WeakEntry for TrackWeak { /// Obtained from [`Producer::demand`], [`Request::demand`], or [`Requests::demand`]. /// A publisher uses it to learn which group anyone subscribed, querying, or fetching /// wants (on-demand capture / encoding) without being able to publish frames or close the -/// track. A bare [`Consumer`] is not demand. It's a weak handle, so it neither keeps +/// track. A bare [`Consumer`] is not demand. A group a subscription or fetch handed out +/// keeps its track wanted until a reader reads its end, and a wanted track makes its +/// broadcast wanted ([`broadcast::Demand`]). It's a weak handle, so it neither keeps /// the track alive nor pins its cached groups; once the owning [`Producer`] /// goes away, [`poll_wanted_changed`](Self::poll_wanted_changed) reports the track's /// closure. Each clone watches on its own, so its first @@ -5433,6 +5440,7 @@ impl kio::Task for Fetching { // own start. Coverage was checked first, so this can only skip frames they excluded. let expiry: Arc = Arc::new(FetchExpiry { state: state.weak(), + reading: Reading::new(state.clone()), sequence, max_delay: self.max_delay, superseded: Superseded::default(), @@ -5627,12 +5635,8 @@ struct Drift { /// Keeps one handed-out group tied to the subscription whose cursor selected it. struct GroupExpiry { - /// Weak so a handed-out group never joins the track's consumer count, which is what - /// [`Demand`] reads to decide the track still has readers. A group - /// outlives the cursor that produced it (a relay serves one for the life of its - /// stream), and pinning demand on that would hold an upstream subscription open past - /// the last real subscriber. state: kio::ConsumerWeak, + reading: Reading, subscription: kio::Consumer, /// What the cursor can be served; see [`Cursor::published`]. servable: kio::Consumer, @@ -5640,14 +5644,37 @@ struct GroupExpiry { superseded: Superseded, } -/// Keeps one fetched group tied to this caller's age budget without adding live demand. +/// Keeps one fetched group tied to this caller's age budget. struct FetchExpiry { state: kio::ConsumerWeak, + reading: Reading, sequence: u64, max_delay: Option, superseded: Superseded, } +/// The track demand a handed-out group holds for its readers, and the frames they take, +/// until one of them reads the group's end: from then on the track owes it nothing. +struct Reading(Mutex>>); + +impl Reading { + fn new(track: kio::Consumer) -> Self { + Self(Mutex::new(Some(track))) + } + + /// The same hold on the copy a reader moved to, unless the group already ended. + fn moved(&self, track: &kio::ConsumerWeak) -> Self { + let held = self.0.lock().unwrap_or_else(PoisonError::into_inner).is_some(); + Self(Mutex::new(held.then(|| track.consume()))) + } + + fn release(&self) { + let track = self.0.lock().unwrap_or_else(PoisonError::into_inner).take(); + // Dropped outside the lock: the last reader leaving wakes the track's watchers. + drop(track); + } +} + /// The successor a handed-out group was last judged by, so the group stays judgeable once /// the cache evicts both; see [`TrackState::superseded_at`]. Each reader keeps its own, so /// the memory is bounded by the groups readers hold. @@ -5665,14 +5692,20 @@ impl group::Expiry for FetchExpiry { fn for_track(&self, track: &Consumer) -> Arc { // A replacement copy is always one track instance; the fallback is never read. let state = track.with_state(|state| state.cloned()); + let state = state.unwrap_or_else(|| self.state.clone()); Arc::new(Self { - state: state.unwrap_or_else(|| self.state.clone()), + reading: self.reading.moved(&state), + state, sequence: self.sequence, max_delay: self.max_delay, superseded: Superseded::default(), }) } + fn ended(&self) { + self.reading.release(); + } + fn is_expired(&self, max_delay: Option, waiter: &kio::Waiter) -> bool { let state = self.state.read(); let budget = match (self.max_delay, max_delay) { @@ -5691,8 +5724,10 @@ impl group::Expiry for GroupExpiry { fn for_track(&self, track: &Consumer) -> Arc { // A replacement copy is always one track instance; the fallback is never read. let state = track.with_state(|state| state.cloned()); + let state = state.unwrap_or_else(|| self.state.clone()); Arc::new(Self { - state: state.unwrap_or_else(|| self.state.clone()), + reading: self.reading.moved(&state), + state, subscription: self.subscription.clone(), servable: self.servable.clone(), sequence: self.sequence, @@ -5700,6 +5735,10 @@ impl group::Expiry for GroupExpiry { }) } + fn ended(&self) { + self.reading.release(); + } + fn is_expired(&self, max_delay: Option, waiter: &kio::Waiter) -> bool { let max_delay = max_delay.unwrap_or_else(|| { let mut max_delay = Duration::default(); @@ -6565,6 +6604,7 @@ impl Cursor { let sequence = group.sequence; group.with_expiry(Arc::new(GroupExpiry { state: self.state.weak(), + reading: Reading::new(self.state.clone()), subscription: self.subscription.consume(), servable: self.published.consume(), sequence, @@ -13074,11 +13114,11 @@ mod test { assert!(matches!(write, Err(Error::Cancel)), "{write:?}"); } - /// Aborting an unused track aborts an open group whose reader outlived the track's - /// consumers. + /// A reader still in an open group keeps the track used after its subscription leaves, + /// so only its leaving lets an unused abort end the group's writer. #[test] #[cfg(feature = "async")] - fn abort_unused_ends_a_pulled_open_group() { + fn abort_unused_waits_for_a_pulled_open_group() { let producer = track_producer("test", Info::timed()); let mut consumer = producer.subscribe(None); let mut writer = producer.append_group().unwrap(); @@ -13089,9 +13129,11 @@ mod test { assert!(group.read_frame().now_or_never().unwrap().unwrap().is_some()); drop(consumer); + let producer = producer + .abort_unused(Error::Cancel) + .expect_err("the group's reader keeps the track"); + drop(group); assert!(producer.abort_unused(Error::Cancel).is_ok()); - let read = group.read_frame().now_or_never().expect("should not block"); - assert!(matches!(read, Err(Error::Cancel)), "{read:?}"); let write = writer.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"y")); assert!(matches!(write, Err(Error::Cancel)), "{write:?}"); } @@ -17596,6 +17638,80 @@ mod test { } } + /// A handed-out group keeps its track wanted once its subscription leaves, until a + /// reader reads its end: a finished group owes nothing more, even while still held. + #[test] + fn a_group_read_keeps_its_track_wanted_until_its_end() { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(None); + let mut writing = producer.append_group().unwrap(); + writing.write_frame(Timestamp::ZERO, &b"key"[..]).unwrap(); + let mut group = subscriber.assert_group(); + let other = group.clone(); + drop(subscriber); + assert_eq!(producer.demand().wanted(), Some(0), "the group is read"); + + writing.finish().unwrap(); + assert_eq!(producer.demand().wanted(), Some(0), "not read to its end yet"); + let waiter = kio::Waiter::noop(); + assert!(matches!(group.poll_read_frame(&waiter), Poll::Ready(Ok(Some(_))))); + assert!(matches!(group.poll_read_frame(&waiter), Poll::Ready(Ok(None)))); + assert_eq!( + producer.demand().wanted(), + None, + "a clone reading the end ends it for all" + ); + drop(other); + } + + /// A reader stopping at its range cap has not reached the group's end: the group is + /// still open, so it keeps its track wanted for the others and for a raised cap. + #[test] + fn a_range_cap_is_not_the_groups_end() { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(None); + let mut writing = producer.append_group().unwrap(); + writing.write_frame(Timestamp::ZERO, &b"key"[..]).unwrap(); + let group = subscriber.assert_group(); + let mut capped = group.clone().with_range(..1); + drop(subscriber); + + let waiter = kio::Waiter::noop(); + assert!(matches!(capped.poll_read_frame(&waiter), Poll::Ready(Ok(Some(_))))); + assert!(matches!(capped.poll_read_frame(&waiter), Poll::Ready(Ok(None)))); + assert!(matches!(capped.poll_finished(&waiter), Poll::Ready(Ok(1)))); + assert_eq!(producer.demand().wanted(), Some(0), "the cap is not the group's end"); + drop(group); + let producer = producer + .abort_unused(Error::Cancel) + .expect_err("the capped reader can raise its cap"); + + writing.write_frame(Timestamp::ZERO, &b"delta"[..]).unwrap(); + let mut capped = capped.with_range(..); + assert!(matches!(capped.poll_read_frame(&waiter), Poll::Ready(Ok(Some(_))))); + writing.finish().unwrap(); + assert!(matches!(capped.poll_read_frame(&waiter), Poll::Ready(Ok(None)))); + assert_eq!(producer.demand().wanted(), None, "read to the group's end"); + } + + /// A fetched group keeps its track wanted after the fetch resolved, while it is read. + #[test] + #[cfg(feature = "async")] + fn a_fetched_group_keeps_its_track_wanted() { + let producer = track_producer("test", Info::timed()); + let mut writing = producer.append_group().unwrap(); + writing.write_frame(Timestamp::ZERO, &b"key"[..]).unwrap(); + let group = producer + .consume() + .fetch_group(0, None) + .now_or_never() + .expect("cached") + .unwrap(); + assert_eq!(producer.demand().wanted(), Some(0)); + drop(group); + assert_eq!(producer.demand().wanted(), None); + } + /// A datagram past the newest group took the sequence after it, so a join over the /// ratio wants the next free one rather than one already written. #[test] diff --git a/rs/moq-net/tests/loom.rs b/rs/moq-net/tests/loom.rs index ad1121a64c4..f74cf58299f 100644 --- a/rs/moq-net/tests/loom.rs +++ b/rs/moq-net/tests/loom.rs @@ -206,6 +206,52 @@ fn subscriber_wakes_parked_demand() { }); } +/// A publisher waiting on `broadcast::Demand` while one reader hands over to the next: +/// the last reader leaves and a new one arrives straight from the track producer, which +/// never touches the broadcast's own state. Whatever the publisher's poll saw, it must be +/// woken for the newcomer. +#[test] +fn broadcast_demand_sees_a_reader_return() { + model(|| { + let broadcast = broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let video = broadcast.create_track("video", None).expect("create track"); + let mut demand = broadcast.demand(); + let first = block_on(consumer.track("video").expect("track").subscribe(None)).expect("subscribe"); + + let reader = thread::spawn(move || { + drop(first); + let next = video.subscribe(None); + // Returned, so neither the reader nor the track goes before the join. + (video, next) + }); + + // The first call reports whatever the swap left; a reader gone must not stay unseen. + while !block_on(demand.wanted_changed()).expect("the returning reader was missed") {} + drop(reader.join().unwrap()); + }); +} + +/// A publisher waiting on `broadcast::Demand` while a reader comes and goes straight +/// through the track producer: its leaving must wake the publisher. +#[test] +fn broadcast_demand_sees_a_reader_leave() { + model(|| { + let broadcast = broadcast::Info::new().produce(); + let video = broadcast.create_track("video", None).expect("create track"); + let mut demand = broadcast.demand(); + let reader = video.subscribe(None); + + let leaver = thread::spawn(move || { + drop(reader); + video + }); + + while block_on(demand.wanted_changed()).expect("the reader leaving was missed") {} + drop(leaver.join().unwrap()); + }); +} + /// A reader taking and dropping a track around the front's demand poll. The front must /// see it go: one that missed the edge keeps its source subscribed for nobody, which /// shows up here as a deadlock waiting for the source to go unread. diff --git a/rs/moq-transcode/src/lib.rs b/rs/moq-transcode/src/lib.rs index 34a8eb666ea..1b1323dc394 100644 --- a/rs/moq-transcode/src/lib.rs +++ b/rs/moq-transcode/src/lib.rs @@ -1515,11 +1515,10 @@ mod tests { } /// A fetch is demand, so it starts the live path, which claims the group the - /// source has open. A reader that picked that group up from the cache is not - /// demand, so when the fetch ends the live path is left mid-group with + /// source has open. When the fetch ends, the live path is left mid-group with /// nobody asking. It finishes the group before leaving its source: a group - /// name always means the same content, so the reader gets the whole group - /// rather than one aborted under it. + /// name always means the same content, so a later reader of that sequence + /// gets the whole group rather than one aborted under it. /// /// Paused so the runtime settles before the source group ends: the live path /// has seen the demand go away by then, which is the moment that used to @@ -1585,9 +1584,6 @@ mod tests { let mut edge = rung.clone(); while edge.largest().await.map(|position| (position.group, position.frame)) != Some((0, 1)) {} - // Read straight from the cache while the group is still being written. - let mut fetched = rung.fetch_group(0, None).await.unwrap(); - // The only demand ends with output group 0 still open. drop(fetching); tokio::time::sleep(std::time::Duration::from_secs(1)).await; @@ -1600,6 +1596,8 @@ mod tests { group.finish().unwrap(); while source_demand.wanted_changed().await.unwrap().is_some() {} + // Read from the cache after the live path left. + let mut fetched = rung.fetch_group(0, None).await.unwrap(); let mut frames = 0; while fetched.read_frame().await.expect("group 0 was aborted").is_some() { frames += 1;