diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index 1a37f827737..5d7dac2a1ac 100644 --- a/dart/moq/MIGRATE.md +++ b/dart/moq/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `Fill.wantedChanged()` waits for a reader to pick up the group served before + +`await fill.wantedChanged()` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wantedChanged()` until it returns null, accepting one group per call, rather than looping over `wanted()`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take concurrently, since each now waits on its own readers. + ### Socket sessions package:moq/socket.dart now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change. diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index e157887146f..42518706383 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -24711,10 +24711,10 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqfill_unwanted() != 13637) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqfill_wanted() != 65141) { + if (uniffi_moq_ffi_checksum_method_moqfill_wanted() != 16564) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqfill_wanted_changed() != 42560) { + if (uniffi_moq_ffi_checksum_method_moqfill_wanted_changed() != 7000) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqgroupproducer_abort() != 59787) { diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index c43be664ec9..a1fd89df012 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `Fill.WantedChanged` waits for a reader to pick up the group served before + +`Fill.WantedChanged(ctx)` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `WantedChanged` until it returns nil, accepting one group per call, rather than looping over `Wanted`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take in their own goroutines, since each now waits on its own readers. + ### Socket sessions moq.dev/moq/socket now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change. diff --git a/go/wrapper/publish.go b/go/wrapper/publish.go index 013ac28ad76..07282c691d9 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -317,11 +317,13 @@ func (d *TrackRequests) Cancel() { // Fill is a run of groups readers miss, for a TrackRequests to serve. // -// Accept each group you have and DropGroups the rest; Wanted names what is still -// owed, in the fill's order. Releasing the fill names whatever is left as dropped +// Accept each group you have and DropGroups the rest, one group at a time as +// WantedChanged returns them; Wanted names what is still owed, in the fill's +// order. Releasing the fill names whatever is left as dropped // while a reader still wants it; once nobody does, it is withdrawn, so stop // serving it once Wanted is nil. Stop serving a single group when Unwanted -// returns for it, and move on to Wanted. +// returns for it, and move on to Wanted. Serve the fills you take +// concurrently: each waits on its own readers. type Fill struct { inner *ffi.MoqFill } @@ -358,12 +360,16 @@ func (f *Fill) Order() Order { } // Wanted is the next group still owed in the fill's order, or nil once every -// group is answered or no reader waits on the fill. +// group is answered or no reader waits on the fill. It does not wait for the +// readers; WantedChanged does. func (f *Fill) Wanted() *uint64 { return f.inner.Wanted() } -// WantedChanged blocks until Wanted changes and returns it. The first call returns at once. +// WantedChanged blocks until Wanted changed and every group this fill accepted +// reached a reader, which picks it up when it next reads, and returns it. Nil, +// once the fill is done, never waits; the first call returns at once unless an +// accepted group has yet to reach a reader. func (f *Fill) WantedChanged(ctx context.Context) (*uint64, error) { return f.inner.WantedChanged(ctx) } diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index 04008a715c3..ba7286c5ffe 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `Fill.wantedChanged()` waits for a reader to pick up the group served before + +`fill.wantedChanged()` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wantedChanged()` until it returns null, accepting one group per call, rather than looping over `wanted()`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take in their own coroutines, since each now waits on its own readers. + ### Socket sessions dev.moq.socket now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change. Like broadcasts, endpoint `end()` forces shutdown while AutoCloseable `close()` releases the handle. diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index deb7664f508..e91e7200cb3 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `Fill.wanted_changed()` waits for a reader to pick up the group served before + +`await fill.wanted_changed()` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wanted_changed()` until it returns `None`, accepting one group per call, rather than looping over `wanted()`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take in their own tasks, since each now waits on its own readers. + ### Socket sessions moq.socket now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change. diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index ef3530a1799..b1498a592da 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -208,11 +208,13 @@ def abort(self, error_code: int) -> None: class Fill: """A run of groups readers miss, for a :class:`TrackRequests` to serve. - Call :meth:`accept` for each group you have and :meth:`drop_groups` for the rest; - :meth:`wanted` names what is still owed, in the fill's order. Dropping the fill names + Call :meth:`accept` for each group you have and :meth:`drop_groups` for the rest, one + group at a time as :meth:`wanted_changed` yields them; :meth:`wanted` names what is still + owed, in the fill's order. Dropping the fill names whatever is left as dropped while a reader still wants it; once nobody does, it is withdrawn, so stop serving it once :meth:`wanted` is ``None``. Stop serving a single group - when :meth:`unwanted` returns for it, and move on to :meth:`wanted`. + when :meth:`unwanted` returns for it, and move on to :meth:`wanted`. Serve the fills you + take concurrently: each waits on its own readers. """ def __init__(self, inner: MoqFill) -> None: @@ -251,12 +253,18 @@ def order(self) -> MoqOrder: def wanted(self) -> int | None: """The next group still owed in the fill's order. - ``None`` once every group is answered or no reader waits on the fill. + ``None`` once every group is answered or no reader waits on the fill. It does not wait + for the readers; :meth:`wanted_changed` does. """ return self._inner.wanted() async def wanted_changed(self) -> int | None: - """Wait for ``wanted`` to change and return it. The first call returns at once.""" + """The next group to serve, once every group this fill accepted reached a reader. + + Returns ``wanted`` once it changed and a reader picked up what was served, as it next + reads. ``None``, once the fill is done, never waits; the first call returns at once + unless an accepted group has yet to reach a reader. + """ return await self._inner.wanted_changed() def accept(self, sequence: int) -> GroupProducer: diff --git a/quest/m2/unbounded-groups/README.md b/quest/m2/unbounded-groups/README.md index f834d61ce5e..1157a11f5b9 100644 --- a/quest/m2/unbounded-groups/README.md +++ b/quest/m2/unbounded-groups/README.md @@ -33,7 +33,9 @@ flights: prefix is transferred once per flight. Carrying the inflater's state from the cached copy, and keeping refresh or `Evicted` for compressed groups, were rejected. -- Paced batch fills are [their own quest](/quest/m2/unbounded-groups/paced-batch-fills.md). +- Paced batch fills were their own quest, landed in + [moq-dev/moq#8017](https://github.com/moq-dev/moq/pull/8017): `Fill::wanted_changed` + yields the next group only once a reader picked up the one before. - No cap on concurrent flights per group: memory is bounded by each flight's tail and pool debt, and the cost of more flights is upstream transfer. Revisit with relay evidence. @@ -178,7 +180,6 @@ the model layer (maintainer, 2026-10-30). ## Required -- [Paced batch fills](/quest/m2/unbounded-groups/paced-batch-fills.md) - fills that write a batch of groups stop evicting their own earlier groups before the reader takes them - [Byte holes](/quest/m2/unbounded-groups/byte-holes.md) - segments inside a frame shed like front frames and flights start at `(frame, byte)` on the current lite wip; the frame cap goes - [JS byte holes](/quest/m2/unbounded-groups/js-byte-holes.md) - `@moq/net` mirrors the byte-offset wire and continues a frame cut off mid-way - [Compressed frames have no size cap and inflate as a bounded stream](/quest/m2/unbounded-groups/compressed-stream.md) - A compressed frame of any size publishes and is received in Rust and `@moq/net`, like a plain one diff --git a/quest/m2/unbounded-groups/paced-batch-fills.md b/quest/m2/unbounded-groups/paced-batch-fills.md deleted file mode 100644 index 91f77a38ca4..00000000000 --- a/quest/m2/unbounded-groups/paced-batch-fills.md +++ /dev/null @@ -1,26 +0,0 @@ -# [M] Paced batch fills - -## Goal - -A `track::Requests` fill that writes a whole batch of groups at once no longer -evicts its own earlier groups before the reader takes them. A reader of a -batch fill reads every group it asked for, at its own pace, without the cache -holding the whole batch. - -## Plan - -Decided 2026-11-05 (maintainer, /quest-plan): split from -[moq-dev/moq#7944](https://github.com/moq-dev/moq/pull/7944), which landed -flights, into a quest of its own. The failure crosses groups, while a flight -paces one group copy by its riders, and pacing the batch changes what -`Fill::wanted` promises every handler. - -The follow-up from [moq-dev/moq#7339](https://github.com/moq-dev/moq/pull/7339), -moved to flights from -[fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664) on -2026-11-02 (maintainer). Recommendation from #7944: `Fill::wanted` yields the -next group only once the readers took the one before, so a handler serving a -batch writes it a group at a time. - -Public API: `Fill::wanted`'s contract for handlers, with a `MIGRATE.md` note. -Wire: none. diff --git a/rs/moq-archive/src/consumer/track.rs b/rs/moq-archive/src/consumer/track.rs index 23c7ccdc160..2b3b6db58dc 100644 --- a/rs/moq-archive/src/consumer/track.rs +++ b/rs/moq-archive/src/consumer/track.rs @@ -154,17 +154,13 @@ impl Live { } } -/// Serve `fill` from the archive, a group at a time in the fill's order, until every group is -/// answered or nobody waits on it. -pub(super) async fn serve(shared: Arc>, index: Arc>, fill: Fill) { +/// Serve `fill` from the archive, a group at a time in the fill's order as its readers pick +/// them up, until every group is answered or nobody waits on it. +pub(super) async fn serve(shared: Arc>, index: Arc>, mut fill: Fill) { if let Some(pts) = fill.time() { return look_up(&shared, &index, &fill, pts).await; } - let mut last = None; - while let Some(sequence) = fill.wanted() - && last != Some(sequence) - { - last = Some(sequence); + while let Some(sequence) = fill.wanted_changed().await { serve_group(&shared, &index, &fill, sequence).await; } } diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index b399493045b..f4975f26a96 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -957,14 +957,11 @@ impl Publish { } } - /// Hand `fill` to C a group at a time, each once the one before it is answered or nobody - /// reads it any more, until every group is answered or nobody waits on the fill. + /// Hand `fill` to C a group at a time, each once the one before it reached a reader or + /// nobody reads it any more, until every group is answered or nobody waits on the fill. async fn serve_fill(callback: OnStatus, fill: Arc) { - let mut last = None; - while let Some(sequence) = fill.wanted() - && last != Some(sequence) - { - last = Some(sequence); + let mut changes = (*fill).clone(); + while let Some(sequence) = changes.wanted_changed().await { let (pending, answered) = oneshot::channel(); let request = GroupRequest { fill: fill.clone(), diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index 09a6cc2c73c..3f8e68f7198 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -5,6 +5,10 @@ These are the UniFFI names; each language wrapper has its own `MIGRATE.md` in it ## Unreleased +### `MoqFill::wanted_changed` waits for a reader to pick up the group served before + +`MoqFill::wanted_changed` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wanted_changed` until it returns null, accepting one group per call, rather than looping over `wanted`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take concurrently, since each now waits on its own readers. + ### Socket sessions MoqSocketEndpoint, MoqSocketIncomingTcp, MoqSocketIncomingTrack, MoqSocketTcpStream, and MoqSocketTrackStream now expose TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change. Dropping an unfinished group producer now resets it with `Unroutable` instead of leaking Rust's programming-error `Dropped`; finished groups still end cleanly. diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index a6bc644d0d5..acf270ced14 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -478,10 +478,12 @@ impl MoqTrackRequests { /// A run of groups readers miss, mirroring [`moq_net::track::Fill`]. /// /// Call [`accept`](Self::accept) for each group you have and [`drop_groups`](Self::drop_groups) -/// for the rest; [`wanted`](Self::wanted) names what is still owed, in the fill's order. Releasing +/// for the rest, one group at a time as [`wanted_changed`](Self::wanted_changed) yields them; +/// [`wanted`](Self::wanted) names what is still owed, in the fill's order. Releasing /// it names whatever is left as dropped while a reader still wants it; once nobody does, it is /// withdrawn, so stop serving it once [`wanted`](Self::wanted) is null. Stop serving a single -/// group when [`unwanted`](Self::unwanted) resolves for it, and move on to the next. +/// group when [`unwanted`](Self::unwanted) resolves for it, and move on to the next. Serve the +/// fills you take concurrently: each waits on its own readers. #[derive(uniffi::Object)] pub struct MoqFill { inner: moq_net::track::Fill, @@ -530,12 +532,15 @@ impl MoqFill { } /// The next group still owed in the fill's order, or null once every group is answered or - /// no reader waits on the fill. + /// no reader waits on the fill. It does not wait for the readers; `wanted_changed` does. pub fn wanted(&self) -> Option { self.inner.wanted() } - /// Wait for `wanted` to change and return the new value. The first call returns at once. + /// The next group to serve: `wanted` once it changed and every group this fill accepted + /// reached a reader, which picks it up when it next reads. Null, once the fill is done, + /// never waits; the first call returns at once unless an accepted group has yet to reach a + /// reader. Serve one group per call, so the fill writes no faster than its readers read. pub async fn wanted_changed(&self) -> Result, MoqError> { // Polled in place: a cancelled call left unfreed must not advance the // watcher and swallow the change the next call waits for. diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index c9a8ccd87ec..ae43f831715 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,13 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `track::Fill::wanted_changed` waits for a reader to pick up the group served before + +`Fill::wanted_changed` (and `poll_wanted_changed`) now yields the next group only once every group the fill accepted reached a reader, which picks it up as it asks for its next group, so a handler writes a batch one group at a time at its fastest reader's pace. +Serve a fill with `while let Some(sequence) = fill.wanted_changed().await` rather than looping over `Fill::wanted`, which still names the next group owed without waiting: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. +Serve the fills you take concurrently: each waits on its own readers, so serving them in turn holds every fill behind the readers of the one before. +Don't block that loop on `group::Producer::ready` while writing a group: a reader reading in another order than the fill's picks the group up but reads none of it until its older groups arrive. + ### Snapshots split connected sessions into idle and active `stats::Snapshot::activity()` returns each tier's `stats::Activity`: connected sessions with diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 0ac030e46ae..6546a6a9e6a 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -11113,6 +11113,8 @@ struct FillServeRun { fill: track::Fill, info: FetchInfo, run: Option<(u64, FetchServeRun)>, + /// The group the last run answered. + last: Option, } impl Drop for FillServeRun { @@ -11132,6 +11134,7 @@ impl FillServeRun { fill, info, run: None, + last: None, } } } @@ -11141,25 +11144,24 @@ impl kio::Task for FillServeRun { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { loop { - let last = match &mut self.run { - Some((sequence, run)) => { - ready!(run.poll(waiter)); - let sequence = *sequence; - self.run = None; - Some(sequence) - } - None => None, - }; + if let Some((sequence, run)) = &mut self.run { + ready!(run.poll(waiter)); + self.last = Some(*sequence); + self.run = None; + } // Withdrawn: nobody waits on what is left, so it records nothing. if self.fill.poll_withdrawn(waiter).is_ready() { return Poll::Ready(()); } + // The next fetch waits for a reader to pick up the group the last one brought, so a + // fill wider than the cache never evicts its own groups before they are read. + ready!(self.fill.poll_paced(waiter)); let Some(sequence) = self.fill.wanted() else { return Poll::Ready(()); }; // Every run answers its group, so the same one coming back is a bug; ending the // fill names it dropped rather than fetching it forever. - if last == Some(sequence) { + if self.last == Some(sequence) { debug_assert!(false, "fill group {sequence} left unanswered"); return Poll::Ready(()); } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index ccc51749734..8e5e90c8302 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -5746,6 +5746,20 @@ impl kio::Task for Fetching { type Output = Result; fn poll(&mut self, waiter: &kio::Waiter) -> Poll { + let res = ready!(self.poll_answer(waiter)); + // Picked up, so a batch fill this fetch joined may serve its next group, though the + // caller keeps the fetch and its join. + if let Some(join) = &self.result { + let mut picked = RangeSet::new(); + picked.insert(self.sequence..self.sequence.saturating_add(1)); + join.saw(&picked); + } + Poll::Ready(res) + } +} + +impl Fetching { + fn poll_answer(&mut self, waiter: &kio::Waiter) -> Poll> { // Made before a front took its logical track, which caches nothing: the serving // route answers instead. if self.resume.is_none() @@ -5887,7 +5901,7 @@ impl kio::Task for Fetching { end: self.frame_end, }, ); - return self.poll(waiter); + return self.poll_answer(waiter); } Poll::Ready(Err(err)) } @@ -7707,6 +7721,9 @@ impl Cursor { let drift = ready!(self.poll_drift(&self.servable, waiter))?; self.reoffer(drift.budget); self.probe(self.floor(), &drift, taken, waiter); + if let Some(misses) = &mut self.misses { + misses.pick_up(); + } self.poll_recv_with(drift, waiter) } @@ -8006,6 +8023,9 @@ impl Cursor { // One scan for the whole poll, so walking a backlog off stays linear in its size. let drift = ready!(self.poll_drift(&self.servable, waiter))?; self.probe(floor, &drift, taken, waiter); + if let Some(misses) = &mut self.misses { + misses.pick_up(); + } self.poll_next_with(floor, drift, waiter) } @@ -19951,7 +19971,8 @@ mod test { } /// A parked [`Fill::poll_wanted_changed`] wakes on every move of the group owed: a - /// reader narrowing past it, an answer, and the last reader leaving, which ends at `None`. + /// reader narrowing past it, a reader picking up an answer, and the last reader leaving, + /// which ends at `None`. #[test] fn fill_wanted_changed_follows_the_group_owed() { let mut producer = track_producer("test", Info::timed()); @@ -19980,7 +20001,13 @@ mod test { assert!(fill.poll_wanted_changed(&waiter).is_pending()); write_filled(fill.accept(8, None), 8); - assert!(flag.take(), "an answer woke the parked poll"); + flag.take(); + assert!( + fill.poll_wanted_changed(&waiter).is_pending(), + "group 9 waits for the reader to pick up 8" + ); + assert!(matches!(subscriber.poll_recv_group(&noop), Poll::Ready(Ok(Some(_))))); + assert!(flag.take(), "the reader picking it up woke the parked poll"); assert_eq!(fill.poll_wanted_changed(&waiter), Poll::Ready(Some(9))); assert!(fill.poll_wanted_changed(&waiter).is_pending()); @@ -20653,13 +20680,23 @@ mod test { }; let waiter = kio::Waiter::noop(); let mut refilled = 0; + let mut fills = Vec::new(); let mut serve = || { while let Poll::Ready(Ok(fill)) = dynamic.poll_requested_fill(&waiter) { - while let Some(sequence) = fill.wanted() { - write(fill.accept(sequence, None), sequence); - refilled += 1; - } + fills.push(fill); } + fills.retain_mut(|fill| { + loop { + match fill.poll_wanted_changed(&waiter) { + Poll::Ready(Some(sequence)) => { + write(fill.accept(sequence, None), sequence); + refilled += 1; + } + Poll::Ready(None) => break false, + Poll::Pending => break true, + } + } + }); }; let (mut read, mut drops) = (Vec::new(), Vec::new()); // A wire publisher forwards what the reader gives up as it reads. @@ -20726,6 +20763,153 @@ mod test { assert_eq!(lost, named, "every group lost is named dropped"); } + /// A batch far past the cache, asked by one slow reader: served through `wanted_changed`, + /// the fill writes one group at a time as the reader takes them, so nothing it wrote is + /// evicted before the reader gets it. + #[test] + #[cfg(feature = "async")] + fn a_batch_fill_waits_for_its_reader() { + let size = 10_000; + let (mut producer, _pool) = pooled_producer(4 * size as u64); + let dynamic = producer.requests(); + let mut subscriber = producer.subscribe(replay().with_range(0..).with_order(Order::Ascending)); + let write = |group: Result, sequence: u64| { + let mut group = group.unwrap(); + group + .write_frame(Timestamp::from_millis(sequence).unwrap(), vec![0u8; size]) + .unwrap(); + group.finish().unwrap(); + }; + producer.start_at(GROUPS).unwrap(); + write(producer.create_group(group::Info { sequence: GROUPS }), GROUPS); + let waiter = kio::Waiter::noop(); + assert!(subscriber.poll_next_group(&waiter).is_pending(), "group 0 is a miss"); + let Poll::Ready(Ok(mut fill)) = dynamic.poll_requested_fill(&waiter) else { + panic!("the reader asked for the batch"); + }; + assert_eq!((fill.start().group, fill.end()), (0, GROUPS)); + let (mut read, mut ahead) = (Vec::new(), 0); + loop { + let mut served = 0; + while let Poll::Ready(Some(sequence)) = fill.poll_wanted_changed(&waiter) { + write(fill.accept(sequence, None), sequence); + served += 1; + } + ahead = ahead.max(served); + match subscriber.poll_next_group(&waiter) { + Poll::Ready(Ok(Some(group))) => read.push(group.sequence), + _ => break, + } + } + assert_eq!(read, (0..=GROUPS).collect::>(), "every group once, in order"); + assert!(drained_drops(&mut subscriber).is_empty()); + assert_eq!(ahead, 1, "one group served at a time"); + } + + /// A reader asking newest first but reading in sequence picks up each group the fill writes + /// while it waits on the oldest, so the fill never waits on a group the reader cannot take + /// yet. + #[test] + #[cfg(feature = "async")] + fn a_fill_against_its_readers_order_still_finishes() { + let mut producer = track_producer("test", Info::timed()); + let dynamic = producer.requests(); + let subscriber = producer.subscribe(replay().with_range(0..).with_order(Order::Descending)); + let mut ordered = subscriber.ordered(); + producer.start_at(4).unwrap(); + write_group(&producer, 4); + let waiter = kio::Waiter::noop(); + assert!(ordered.poll_next_group(&waiter).is_pending(), "group 0 is a miss"); + let Poll::Ready(Ok(mut fill)) = dynamic.poll_requested_fill(&waiter) else { + panic!("the reader asked for the batch"); + }; + assert_eq!(fill.order(), Order::Descending); + let (mut served, mut read) = (Vec::new(), Vec::new()); + while let Poll::Ready(Some(sequence)) = fill.poll_wanted_changed(&waiter) { + write_filled(fill.accept(sequence, None), sequence); + served.push(sequence); + while let Poll::Ready(Ok(Some(group))) = ordered.poll_next_group(&waiter) { + read.push(group.sequence); + } + } + assert_eq!(served, [3, 2, 1, 0]); + assert_eq!(read, [0, 1, 2, 3, 4]); + } + + /// A publisher out of stream credit still forwards drops, whose probe takes in a fill's + /// answers: only a read picks a group up, so the fill never runs ahead of a publisher + /// that cannot send what it wrote. + #[test] + #[cfg(feature = "async")] + fn a_drop_poll_does_not_pick_up_a_filled_group() { + let mut producer = track_producer("test", Info::timed()); + let dynamic = producer.requests(); + let mut subscriber = producer.subscribe(replay().with_range(0..).with_order(Order::Ascending)); + producer.start_at(4).unwrap(); + write_group(&producer, 4); + let waiter = kio::Waiter::noop(); + assert!(subscriber.poll_next_group(&waiter).is_pending(), "group 0 is a miss"); + let Poll::Ready(Ok(mut fill)) = dynamic.poll_requested_fill(&waiter) else { + panic!("the reader asked for the batch"); + }; + assert_eq!(fill.poll_wanted_changed(&waiter), Poll::Ready(Some(0))); + write_filled(fill.accept(0, None), 0); + for _ in 0..3 { + assert!(subscriber.poll_drop(&waiter).is_pending()); + assert!( + fill.poll_wanted_changed(&waiter).is_pending(), + "group 1 waits for a read" + ); + } + assert!(matches!(subscriber.poll_next_group(&waiter), Poll::Ready(Ok(Some(group))) if group.sequence == 0)); + assert_eq!(fill.poll_wanted_changed(&waiter), Poll::Ready(Some(1))); + } + + /// A fetch that joined a batch fill picks up its group as it resolves, so the batch moves + /// on though the caller keeps the fetch and the reader that asked the batch narrowed past + /// that group. + #[test] + #[cfg(feature = "async")] + fn a_fetch_picks_up_its_group_in_a_batch_fill() { + let mut producer = track_producer("test", Info::timed()); + let dynamic = producer.requests(); + let ascending = |groups| replay().with_range(groups).with_order(Order::Ascending); + let mut subscriber = producer.subscribe(ascending(0..)); + producer.start_at(4).unwrap(); + write_group(&producer, 4); + let waiter = kio::Waiter::noop(); + assert!(subscriber.poll_recv_group(&waiter).is_ready(), "the live group"); + assert!( + subscriber.poll_recv_group(&waiter).is_pending(), + "groups 0..4 are a miss" + ); + let Poll::Ready(Ok(mut fill)) = dynamic.poll_requested_fill(&waiter) else { + panic!("the reader asked for the batch"); + }; + assert_eq!((fill.start().group, fill.end()), (0, 4)); + let mut fetch = producer.consume().fetch_group(2, None); + assert!(kio::Task::poll(&mut *fetch, &waiter).is_pending()); + assert!( + dynamic.poll_requested_fill(&waiter).is_pending(), + "the fetch joined the batch" + ); + + subscriber.update(ascending(3..)).unwrap(); + assert!(subscriber.poll_recv_group(&waiter).is_pending()); + assert_eq!( + fill.poll_wanted_changed(&waiter), + Poll::Ready(Some(2)), + "only the fetch wants 2" + ); + write_filled(fill.accept(2, None), 2); + assert!( + fill.poll_wanted_changed(&waiter).is_pending(), + "3 waits for 2 to be picked up" + ); + assert!(matches!(kio::Task::poll(&mut *fetch, &waiter), Poll::Ready(Ok(_)))); + assert_eq!(fill.poll_wanted_changed(&waiter), Poll::Ready(Some(3))); + } + /// A reader forgets the misses it gave up once its window moves past them, as it forgets /// what it judged, so a feed skipping every other group costs it only the gaps it can /// still read. diff --git a/rs/moq-net/src/model/track/fill.rs b/rs/moq-net/src/model/track/fill.rs index 24bef42df54..39b8b64a720 100644 --- a/rs/moq-net/src/model/track/fill.rs +++ b/rs/moq-net/src/model/track/fill.rs @@ -54,6 +54,9 @@ struct Pending { flight: bool, // How many joins still read each sequence: one a reader narrowed away is no longer owed. wants: RangeMap, + // What its handler accepted that no join has picked up yet, which holds back the next + // group: see [`Fill::poll_wanted_changed`]. + unseen: RangeSet, // Readers hold consumers, so it counts the fill's demand; the handler's copy says what // became of each sequence. result: kio::Producer, @@ -76,6 +79,7 @@ impl Pending { /// Count one join out of reading `groups`, which it counted in. Returns the runs nobody /// reads any more. fn unwant(&mut self, groups: ops::Range) -> Vec> { + self.unseen.remove(groups.clone()); let counts: Vec<(ops::Range, u32)> = self .wants .overlapping(&groups) @@ -349,6 +353,23 @@ impl Join { self.held.is_empty() } + /// The reader picked up `groups` the fill accepted, so its handler may serve the next one. + pub(super) fn saw(&self, groups: &RangeSet) { + let mut fills = self.fills.lock(); + // Read first: a mutable borrow wakes every poll parked on the fills. + if !fills + .pending + .get(&self.id) + .is_some_and(|pending| self.is_ours(pending) && groups.iter().any(|run| pending.unseen.overlaps(run))) + { + return; + } + let unseen = &mut fills.pending.get_mut(&self.id).expect("checked above").unseen; + for run in groups.iter() { + unseen.remove(run.clone()); + } + } + /// Stop reading `lost`, so the fill no longer owes this reader any of it. pub(super) fn release(&mut self, lost: &RangeSet) { let released: Vec> = lost @@ -524,6 +545,7 @@ impl FillState { queued: true, flight: false, wants: RangeMap::from_iter([(groups.clone(), 1)]), + unseen: RangeSet::new(), result, }, ); @@ -574,6 +596,7 @@ impl FillState { queued: true, flight: true, wants: RangeMap::from_iter([(groups.clone(), 1)]), + unseen: RangeSet::new(), result: Default::default(), }, ); @@ -623,6 +646,7 @@ impl FillState { queued: true, flight: false, wants: RangeMap::new(), + unseen: RangeSet::new(), result, }, ); @@ -790,7 +814,10 @@ pub(super) fn poll_requested( /// [`start`](Self::start)'s frame. Its [`priority`](Self::priority), /// [`max_delay`](Self::max_delay), and [`order`](Self::order) follow its readers as they /// change, so read them as each group is served. The handler calls [`accept`](Self::accept) for each group -/// it has and [`drop_groups`](Self::drop_groups) for the rest. Dropping the fill names +/// it has and [`drop_groups`](Self::drop_groups) for the rest, one group at a time as +/// [`wanted_changed`](Self::wanted_changed) yields them, so it writes no faster than its +/// readers read. Serve the fills you take concurrently: each waits on its own readers, so +/// serving them in turn holds every fill behind the readers of the one before. Dropping the fill names /// whatever is left as dropped while a reader still wants it; once nobody does, the fill is /// withdrawn and records nothing, so stop serving it once [`wanted`](Self::wanted) is `None`. /// A reader narrowing its ranges may leave a single group, so stop serving that one when @@ -944,7 +971,8 @@ impl Fill { /// The next group still owed in the fill's order: a reader still reads it, and it was /// neither answered nor delivered live since it was asked. `None` once none is owed or - /// no reader waits on the fill, so stop serving it. + /// no reader waits on the fill, so stop serving it. It does not wait for the readers to + /// pick up what was served before it; [`wanted_changed`](Self::wanted_changed) does. pub fn wanted(&self) -> Option { self.wanted_except(None) } @@ -988,11 +1016,15 @@ impl Fill { } } - /// [`wanted`](Self::wanted) once it differs from what this handle last returned: - /// immediately on the first call, then on each change. + /// The next group to serve: [`wanted`](Self::wanted) once it differs from what this handle + /// last returned and every group this fill accepted reached a reader, which picks it up + /// when it next reads. `None`, once the fill is withdrawn or answered, never waits. The + /// first call returns at once, unless a group already accepted has yet to reach a reader. /// - /// So `while let Some(sequence) = fill.wanted_changed().await` serves each group as it - /// becomes owed, and ends once the fill is withdrawn or answered. + /// So `while let Some(sequence) = fill.wanted_changed().await` writes one group at a time + /// as the fastest reader reads on, rather than a batch the cache may evict before it is + /// read. A reader reading in another order than the fill's picks up groups it cannot take + /// yet, so those wait in the cache. #[cfg_attr(feature = "async", kio::wrapper)] pub fn poll_wanted_changed(&mut self, waiter: &kio::Waiter) -> Poll> { // Arm every input of `wanted` before reading it, so a change in between still wakes: @@ -1004,13 +1036,30 @@ impl Fill { let _ = self.inner.fills.poll(waiter, |_| Poll::Pending); let _ = self.inner.asks.poll(waiter, |_| Poll::Pending); let wanted = self.wanted(); - if self.prev == Some(wanted) { + if self.prev == Some(wanted) || (wanted.is_some() && self.inner.is_unseen()) { return Poll::Pending; } self.prev = Some(wanted); Poll::Ready(wanted) } + /// Poll until every group this fill accepted reached a reader, or nobody waits on the + /// fill, as [`Self::poll_wanted_changed`] paces. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn poll_paced(&self, waiter: &kio::Waiter) -> Poll<()> { + if self.poll_withdrawn(waiter).is_ready() { + return Poll::Ready(()); + } + let id = self.inner.id; + self.inner + .fills + .poll(waiter, |fills| match fills.pending.get(&id) { + Some(pending) if !pending.unseen.is_empty() => Poll::Pending, + _ => Poll::Ready(()), + }) + .map(drop) + } + /// Refuse every group still owed with `err`, which says nothing about whether the source /// has them, so nothing is recorded dropped. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] @@ -1151,6 +1200,15 @@ impl Inner { self.result.wanted() && self.fills.read().wants(self.id, sequence) } + /// Whether a group this fill accepted has yet to reach a reader. + fn is_unseen(&self) -> bool { + self.fills + .read() + .pending + .get(&self.id) + .is_some_and(|pending| !pending.unseen.is_empty()) + } + fn max_delay(&self) -> Option { self.asks.read().max_delay() } @@ -1209,6 +1267,19 @@ impl Inner { if self.result.read().resolved.contains(&sequence) { return Err(Error::Duplicate); } + // Written for a reader that has yet to pick it up, which the next group waits on. + // Recorded before the group is cached, since a reader may take it from the cache and + // release it before the answer goes out, which must clear it. + let unseen = { + let mut fills = self.fills.lock(); + match fills.pending.get_mut(&self.id) { + Some(pending) if pending.wants.contains_key(&sequence) => { + pending.unseen.insert(sequence..end); + true + } + _ => false, + } + }; // Cache the group before answering: readers resolve an accept through the cache. let res = TrackState::modify(&self.state).and_then(|mut state| { let res = match self.flight { @@ -1220,6 +1291,12 @@ impl Inner { deferred.wake(); res }); + if unseen + && res.is_err() + && let Some(pending) = self.fills.lock().pending.get_mut(&self.id) + { + pending.unseen.remove(sequence..end); + } if let Ok(mut outcome) = self.result.write() { outcome.accept(sequence..end); if let Ok(group) = &res diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index 7efe9a9c444..6ef6769d703 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -74,6 +74,9 @@ pub(super) struct Misses { wanted: RangeSet, /// Answered by a fill and cached, not handed out yet. filled: BTreeSet, + /// What each join's fill answered that a drop poll took in, told to the fill only once a + /// read picks it up (see [`Self::pick_up`]). + picked: BTreeMap>, /// Never asked again: named dropped to the reader. given_up: RangeSet, /// What became [`Self::gone`] since the cursor last took it, for its window's parked time @@ -388,6 +391,7 @@ impl Misses { if join.is_empty() { self.joins.remove(&key); self.parked.remove(&key); + self.picked.remove(&key); } } } @@ -538,6 +542,7 @@ impl Misses { // What is answered is no longer owed this reader, except a filled group: held until // taken, so its source keeps writing it while the reader picks it up. let mut taken = RangeSet::new(); + let mut seen = RangeSet::new(); for (groups, answer) in answers { self.wanted.remove(groups.clone()); taken.insert(groups.clone()); @@ -545,6 +550,7 @@ impl Misses { // What the cache holds is filled, and the rest settled by run: a fill that // ended unanswered may span far more groups than were ever cached. Ok(()) => { + seen.insert(groups.clone()); self.settle(&state, groups.clone(), given_up); for (&sequence, slot) in state.lookup.range(groups) { if slot.serves_end() { @@ -563,6 +569,9 @@ impl Misses { } } drop(state); + if !seen.is_empty() { + self.picked.entry(key).or_default().extend(seen); + } self.release(&taken); // Judged again: the same join may have more to answer, or park the waiter now. } @@ -576,6 +585,15 @@ impl Misses { retry } + /// Tell each fill this reader joined that a read picked up what it answered, so its + /// handler may serve the next group. Only a read: a drop poll takes in answers too, from + /// a publisher that may be unable to send the groups, which must not pace the fill. + pub fn pick_up(&mut self) { + for (key, groups) in std::mem::take(&mut self.picked) { + self.joins[&key].saw(&groups); + } + } + /// Judge the holes in `target` the last probe has not, returning the runs to ask a /// fill for. Gives up a run already stale under this reader's budget. `named` is the /// lowest group the reader's ranges name, which may reach below the feed's start, diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index b1ac72b9e39..098f608ce7a 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -461,7 +461,8 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result { - retired = true; - break; - }, - _ = finishing.wait_for(|finished| *finished) => break, - permit = limit.clone().acquire_owned() => permit.expect("the semaphore stays open"), - }; - let fill = tokio::select! { biased; () = retire.fired() => { @@ -505,7 +494,7 @@ async fn fetches( }, }; - spawn_fill(&mut tasks, rung.clone(), fill, permit); + spawn_fill(&mut tasks, rung.clone(), fill, limit.clone()); } if retired { @@ -517,16 +506,15 @@ async fn fetches( // Taken in one pass, before waiting on anything, and then the handler is // dropped. A consumer holding the retired track can keep fetching, and the // handler admits a cache miss for as long as it is alive, so a drain that - // waited for a slot with it still open would keep taking those in too and - // the rung would never finish. Dropping it closes admission and releases - // whatever arrived after the pass; the fills already popped are held - // here, so they are unaffected. - let mut queued = Vec::new(); + // waited with it still open would keep taking those in too and the rung + // would never finish. Dropping it closes admission and releases whatever + // arrived after the pass; the fills already popped are served here, so + // they are unaffected. loop { tokio::select! { biased; fill = dynamic.requested_fill() => match fill { - Ok(fill) => queued.push(fill), + Ok(fill) => spawn_fill(&mut tasks, rung.clone(), fill, limit.clone()), // The output track closed; nothing left to serve. Err(_) => break, }, @@ -536,11 +524,6 @@ async fn fetches( } drop(dynamic); - for fill in queued { - let permit = limit.clone().acquire_owned().await.expect("the semaphore stays open"); - spawn_fill(&mut tasks, rung.clone(), fill, permit); - } - // Finish every group the handler accepted, or its producer stays open and // the fetch stalls forever. `serve` ends the track only once this returns, // which is what leaves every one of these groups writable. @@ -557,20 +540,27 @@ async fn fetches( Ok(()) } -/// Spawn one fill, holding its concurrency slot for the life of the task. -fn spawn_fill( - tasks: &mut tokio::task::JoinSet<()>, - rung: Rung, - fill: moq_net::track::Fill, - permit: tokio::sync::OwnedSemaphorePermit, -) { +/// Spawn one fill, taking a slot of `limit` for each group it transcodes. +fn spawn_fill(tasks: &mut tokio::task::JoinSet<()>, rung: Rung, mut fill: moq_net::track::Fill, limit: Arc) { + let mut retire = rung.retire.clone(); tasks.spawn(async move { - let _permit = permit; let mut last = None; - while let Some(sequence) = fill.wanted() - && last != Some(sequence) - { + loop { + // A group at a time as its readers pick them up, so a wide fetch never evicts its + // own earlier groups before they are read. The slot is released before waiting on + // them, so a reader that stops reading never holds an encoder session. + // + // Once retired, the track ends only after this drain, so serve what is still owed + // without waiting on a reader that may never read again. Each group once: one this + // fill left unanswered is not asked for again. + let sequence = tokio::select! { + biased; + () = retire.fired() => fill.wanted().filter(|sequence| last != Some(*sequence)), + sequence = fill.wanted_changed() => sequence, + }; + let Some(sequence) = sequence else { break }; last = Some(sequence); + let _permit = limit.acquire().await.expect("the semaphore stays open"); if let Err(err) = fetch(&rung, &fill, sequence).await { tracing::warn!(%err, sequence, "transcode fetch failed"); } @@ -583,7 +573,7 @@ fn spawn_fill( /// The group is dropped only once the source says it no longer has it. Any other /// failure while someone still wants the group refuses it for those readers alone, /// so a later reader asks again. Once nobody wants the group, it is left unanswered -/// instead, releasing the permit for the groups the fill still owes: its last reader +/// instead, releasing the slot for the groups the fill still owes: its last reader /// already left it, and accepting afterwards would cache the group a fresh fill is /// about to serve. async fn fetch(rung: &Rung, fill: &moq_net::track::Fill, sequence: u64) -> Result<(), Error> { @@ -997,8 +987,7 @@ mod tests { assert_eq!((fill.start().group, fill.end()), (7, 10)); let mut tasks = tokio::task::JoinSet::new(); - let permit = Arc::new(Semaphore::new(1)).try_acquire_owned().unwrap(); - spawn_fill(&mut tasks, rung, fill, permit); + spawn_fill(&mut tasks, rung, fill, Arc::new(Semaphore::new(1))); let wait = std::time::Duration::from_secs(5); let first = kio::clock::timeout(wait, source_fills.requested_fill()) .await @@ -1015,6 +1004,63 @@ mod tests { assert_eq!(next.start().group, 8); } + /// Retirement's drain finishes a fill whose reader stopped reading: the groups still owed + /// are served without waiting for that reader to pick up the last one. + #[tokio::test(start_paused = true)] + async fn retirement_serves_a_fill_past_a_reader_that_stopped() { + use std::task::Poll; + let (rung, source, retire) = idle_rung(); + let source_fills = source.create_track("video", None).unwrap().requests(); + + let output = moq_net::broadcast::Info::default().produce(); + let timed = moq_net::track::Info::default() + .with_timescale(moq_net::Timescale::MILLI) + .with_max_age(hang::container::MAX_AGE); + let mut track = output.create_track("video/120p", timed).unwrap(); + let fills = track.requests(); + let mut reader = track.subscribe( + moq_net::track::Subscription::default() + .with_delay(moq_net::track::Delay::default().with_max(None)) + .with_range(7..) + .with_order(moq_net::track::Order::Ascending), + ); + track.start_at(10).unwrap(); + let mut live = track.create_group(moq_net::group::Info { sequence: 10 }).unwrap(); + live.write_frame(moq_net::Timestamp::from_millis(10).unwrap(), &b"x"[..]) + .unwrap(); + live.finish().unwrap(); + let waiter = kio::Waiter::noop(); + while let Poll::Ready(Ok(Some(_))) = reader.poll_recv_group(&waiter) {} + let fill = fills.requested_fill().await.unwrap(); + assert_eq!((fill.start().group, fill.end()), (7, 10)); + + // Group 7 is served, and the reader never reads again to pick it up. + fill.clone().accept(7, None).unwrap().finish().unwrap(); + let mut tasks = tokio::task::JoinSet::new(); + spawn_fill(&mut tasks, rung, fill, Arc::new(Semaphore::new(1))); + let wait = std::time::Duration::from_secs(5); + assert!( + kio::clock::timeout(wait, source_fills.requested_fill()).await.is_err(), + "group 8 waits for a reader to pick up group 7" + ); + + retire.send(true).unwrap(); + for sequence in 8..10 { + let requested = kio::clock::timeout(wait, source_fills.requested_fill()) + .await + .expect("a retired rung serves what its fill still owes") + .unwrap(); + assert_eq!(requested.start().group, sequence); + // The source no longer has it, so the fill drops it and moves on. + requested.drop_groups(sequence..sequence + 1); + } + kio::clock::timeout(wait, tasks.join_next()) + .await + .expect("the drain finishes") + .unwrap() + .unwrap(); + } + /// A restarted source cancels the serving future even while a rung waits for media. /// Its subscribers must see that cause, rather than an unfinished producer drop. #[tokio::test(start_paused = true)] diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 606337b1b3d..4e8d43289f2 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `Fill.wantedChanged()` waits for a reader to pick up the group served before + +`try await fill.wantedChanged()` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wantedChanged()` until it returns nil, accepting one group per call, rather than looping over `wanted`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take in their own tasks, since each now waits on its own readers. + ### Socket sessions Socket now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change. diff --git a/swift/Sources/Moq/Requests.swift b/swift/Sources/Moq/Requests.swift index 108f8c084b7..6c4bc066d67 100644 --- a/swift/Sources/Moq/Requests.swift +++ b/swift/Sources/Moq/Requests.swift @@ -39,10 +39,12 @@ public final class TrackRequest: Sendable { /// A run of groups readers miss, for a `TrackRequests` to serve. /// -/// Accept each group you have and drop the rest; `wanted` names what is still owed, in the -/// fill's order. Releasing the fill names whatever is left as dropped while a reader still -/// wants it; once nobody does, it is withdrawn, so stop serving it once `wanted` is nil. -/// Stop serving a single group when `unwanted(sequence:)` returns for it, and move on to `wanted`. +/// Accept each group you have and drop the rest, one group at a time as `wantedChanged()` +/// returns them; `wanted` names what is still owed, in the fill's order. Releasing the fill +/// names whatever is left as dropped while a reader still wants it; once nobody does, it is +/// withdrawn, so stop serving it once `wanted` is nil. Stop serving a single group when +/// `unwanted(sequence:)` returns for it, and move on to `wanted`. Serve the fills you take +/// concurrently: each waits on its own readers. public final class Fill: Sendable { let ffi: MoqFill @@ -82,12 +84,14 @@ public final class Fill: Sendable { } /// The next group still owed in the fill's order, or nil once every group is answered or - /// no reader waits on the fill. + /// no reader waits on the fill. It does not wait for the readers; `wantedChanged()` does. public var wanted: UInt64? { ffi.wanted() } - /// Suspend until `wanted` changes and return it. The first call returns at once. + /// Suspend until `wanted` changed and every group this fill accepted reached a reader, + /// which picks it up when it next reads, and return it. Nil, once the fill is done, never + /// waits; the first call returns at once unless an accepted group has yet to reach a reader. public func wantedChanged() async throws -> UInt64? { try await ffi.wantedChanged() }