From 4a41ca9e05e1a743df1445d41c5ed144c45dacab Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 22:29:05 -0800 Subject: [PATCH 1/8] chore(quest): claim quest/m2/unbounded-groups/paced-batch-fills.md Quest-Claim: 3b49c0bf1d88f7d3 From f3f9a82f911e7812aca6b94b52cec47b39f87a8b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 23:23:00 -0800 Subject: [PATCH 2/8] feat(net)!: pace a fill's handler by its fastest reader Fill::wanted_changed yields the next group only once a reader took every group the fill accepted, so a batch fill no longer evicts its own earlier groups before the reader takes them. Co-Authored-By: Claude Opus 5.5 --- go/wrapper/publish.go | 12 +++-- py/moq-rs/moq/publish.py | 14 ++++-- rs/moq-archive/src/consumer/track.rs | 12 ++--- rs/moq-c/src/publish.rs | 11 ++--- rs/moq-ffi/src/producer.rs | 10 ++-- rs/moq-net/MIGRATE.md | 6 +++ rs/moq-net/src/lite/subscriber.rs | 22 +++++---- rs/moq-net/src/model/track.rs | 59 +++++++++++++++++++++-- rs/moq-net/src/model/track/fill.rs | 71 ++++++++++++++++++++++++---- swift/Sources/Moq/Requests.swift | 9 ++-- 10 files changed, 172 insertions(+), 54 deletions(-) diff --git a/go/wrapper/publish.go b/go/wrapper/publish.go index 013ac28ad76..ad0721a2072 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -317,8 +317,9 @@ 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. @@ -358,12 +359,15 @@ 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 a reader took every group this fill accepted and +// Wanted changed, and returns it. The first call returns at once unless an +// accepted group still waits for a reader. func (f *Fill) WantedChanged(ctx context.Context) (*uint64, error) { return f.inner.WantedChanged(ctx) } diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index ef3530a1799..c34e95fd2bd 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -208,8 +208,9 @@ 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`. @@ -251,12 +252,17 @@ 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 a reader took every group this fill accepted. + + Returns ``wanted`` once it changed and no accepted group waits for a reader; the first + call returns at once unless one does. + """ return await self._inner.wanted_changed() def accept(self, sequence: int) -> GroupProducer: diff --git a/rs/moq-archive/src/consumer/track.rs b/rs/moq-archive/src/consumer/track.rs index 23c7ccdc160..00780d1a113 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 take +/// them, 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..8021a86d88c 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 a reader took the one before it 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/src/producer.rs b/rs/moq-ffi/src/producer.rs index a6bc644d0d5..2a6727243cc 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -478,7 +478,8 @@ 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. @@ -530,12 +531,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 a reader took every group this fill accepted, or + /// stopped waiting for it, and once it changed. The first call returns at once unless a + /// group is still untaken. Serve one group per call, so the cache never holds a batch the + /// readers have yet to take. 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 13e37ae6884..45bc2c51d5d 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `track::Fill::wanted_changed` waits for a reader to take the group served before + +`Fill::wanted_changed` (and `poll_wanted_changed`) now yields the next group only once a reader took every group the fill accepted, or stopped waiting for it, 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: an ordered reader may take a fill's group only after another fill answers an earlier one, which a handler serving fills one after another would never reach. + ### Socket readers start at the beginning New `track` socket sessions subscribe from group 0, and `tcp` sessions join their diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 0ac030e46ae..e3d73af8a54 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 take 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_taken(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 1e869a15af6..a57de6c5f76 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -20645,13 +20645,21 @@ 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. @@ -20718,6 +20726,49 @@ 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 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..a374711122b 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 let go of yet, which holds back the next + // group: see [`Fill::poll_wanted_changed`]. + untaken: RangeSet, // Readers hold consumers, so it counts the fill's demand; the handler's copy says what // became of each sequence. result: kio::Producer, @@ -73,9 +76,11 @@ impl Pending { } } - /// Count one join out of reading `groups`, which it counted in. Returns the runs nobody - /// reads any more. + /// Count one join out of reading `groups`, which it counted in: it took them or stopped + /// waiting, either of which lets its handler move on. Returns the runs nobody reads any + /// more. fn unwant(&mut self, groups: ops::Range) -> Vec> { + self.untaken.remove(groups.clone()); let counts: Vec<(ops::Range, u32)> = self .wants .overlapping(&groups) @@ -524,6 +529,7 @@ impl FillState { queued: true, flight: false, wants: RangeMap::from_iter([(groups.clone(), 1)]), + untaken: RangeSet::new(), result, }, ); @@ -574,6 +580,7 @@ impl FillState { queued: true, flight: true, wants: RangeMap::from_iter([(groups.clone(), 1)]), + untaken: RangeSet::new(), result: Default::default(), }, ); @@ -623,6 +630,7 @@ impl FillState { queued: true, flight: false, wants: RangeMap::new(), + untaken: RangeSet::new(), result, }, ); @@ -790,7 +798,9 @@ 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 the cache never holds a batch its +/// readers have yet to take. 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 +954,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 + /// take what was served before it; [`wanted_changed`](Self::wanted_changed) does. pub fn wanted(&self) -> Option { self.wanted_except(None) } @@ -988,11 +999,14 @@ 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 a reader took every group this + /// fill accepted, or stopped waiting for it, and once it differs from what this handle + /// last returned. The first call returns at once unless a group is still untaken. /// - /// 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, + /// at the pace of the fastest reader, and ends once the fill is withdrawn or answered. A + /// batch written ahead of its readers would let the cache evict its earlier groups before + /// they take them. #[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 +1018,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_untaken()) { return Poll::Pending; } self.prev = Some(wanted); Poll::Ready(wanted) } + /// Poll until a reader took every group this fill accepted or stopped waiting for it, 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_taken(&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.untaken.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 +1182,15 @@ impl Inner { self.result.wanted() && self.fills.read().wants(self.id, sequence) } + /// Whether a group this fill accepted still waits for a reader to take it. + fn is_untaken(&self) -> bool { + self.fills + .read() + .pending + .get(&self.id) + .is_some_and(|pending| !pending.untaken.is_empty()) + } + fn max_delay(&self) -> Option { self.asks.read().max_delay() } @@ -1228,7 +1268,18 @@ impl Inner { outcome.partial.insert(sequence, group.handle()); } } - uncover(&self.fills, self.id, sequence..end); + let mut fills = self.fills.lock(); + for range in fills.routed(self.id, &(sequence..end)) { + fills.covers.remove(range); + } + // Written for a reader that has yet to take it, which the next group waits on. + if res.is_ok() + && let Some(pending) = fills.pending.get_mut(&self.id) + && pending.wants.contains_key(&sequence) + { + pending.untaken.insert(sequence..end); + } + drop(fills); res } diff --git a/swift/Sources/Moq/Requests.swift b/swift/Sources/Moq/Requests.swift index 108f8c084b7..d32f40d5209 100644 --- a/swift/Sources/Moq/Requests.swift +++ b/swift/Sources/Moq/Requests.swift @@ -39,8 +39,8 @@ 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 +/// 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`. public final class Fill: Sendable { @@ -82,12 +82,13 @@ 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 a reader took every group this fill accepted and `wanted` changed, and + /// return it. The first call returns at once unless an accepted group still waits. public func wantedChanged() async throws -> UInt64? { try await ffi.wantedChanged() } From f33db01be1b21edd1aa76e393562361be0fd588b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 01:16:54 -0800 Subject: [PATCH 3/8] fix(net): pace a fill on its reader picking up each group, not taking it A reader asking newest first but reading in sequence cannot take the newest group until the oldest arrives, so pacing on the take stalled the fill. Co-Authored-By: Claude Opus 5.5 --- go/wrapper/publish.go | 5 +- py/moq-rs/moq/publish.py | 6 +- quest/m2/unbounded-groups/README.md | 5 +- .../m2/unbounded-groups/paced-batch-fills.md | 26 --------- rs/moq-archive/src/consumer/track.rs | 4 +- rs/moq-c/src/publish.rs | 2 +- rs/moq-ffi/src/producer.rs | 7 +-- rs/moq-net/MIGRATE.md | 6 +- rs/moq-net/src/lite/subscriber.rs | 4 +- rs/moq-net/src/model/track.rs | 57 +++++++++++++++--- rs/moq-net/src/model/track/fill.rs | 58 +++++++++++-------- rs/moq-net/src/model/track/miss.rs | 6 ++ swift/Sources/Moq/Requests.swift | 4 +- 13 files changed, 110 insertions(+), 80 deletions(-) delete mode 100644 quest/m2/unbounded-groups/paced-batch-fills.md diff --git a/go/wrapper/publish.go b/go/wrapper/publish.go index ad0721a2072..ccd838749ac 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -365,9 +365,8 @@ func (f *Fill) Wanted() *uint64 { return f.inner.Wanted() } -// WantedChanged blocks until a reader took every group this fill accepted and -// Wanted changed, and returns it. The first call returns at once unless an -// accepted group still waits for a reader. +// WantedChanged blocks until every group this fill accepted reached a reader +// and Wanted changed, and returns it. The first call returns at once. func (f *Fill) WantedChanged(ctx context.Context) (*uint64, error) { return f.inner.WantedChanged(ctx) } diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index c34e95fd2bd..775240b2745 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -258,10 +258,10 @@ def wanted(self) -> int | None: return self._inner.wanted() async def wanted_changed(self) -> int | None: - """The next group to serve, once a reader took every group this fill accepted. + """The next group to serve, once every group this fill accepted reached a reader. - Returns ``wanted`` once it changed and no accepted group waits for a reader; the first - call returns at once unless one does. + Returns ``wanted`` once it changed and the readers picked up what was served; the first + call returns at once. """ return await self._inner.wanted_changed() 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 00780d1a113..2b3b6db58dc 100644 --- a/rs/moq-archive/src/consumer/track.rs +++ b/rs/moq-archive/src/consumer/track.rs @@ -154,8 +154,8 @@ impl Live { } } -/// Serve `fill` from the archive, a group at a time in the fill's order as its readers take -/// them, until every group is answered or nobody waits on it. +/// 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; diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index 8021a86d88c..f4975f26a96 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -957,7 +957,7 @@ impl Publish { } } - /// Hand `fill` to C a group at a time, each once a reader took the one before it or + /// 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 changes = (*fill).clone(); diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 2a6727243cc..d6ea82dda3b 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -536,10 +536,9 @@ impl MoqFill { self.inner.wanted() } - /// The next group to serve: `wanted` once a reader took every group this fill accepted, or - /// stopped waiting for it, and once it changed. The first call returns at once unless a - /// group is still untaken. Serve one group per call, so the cache never holds a batch the - /// readers have yet to take. + /// The next group to serve: `wanted` once every group this fill accepted reached a reader + /// and it changed. The first call returns at once. Serve one group per call, so the cache + /// never holds a batch the readers have yet to take. 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 45bc2c51d5d..c4943986b7b 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,11 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased -### `track::Fill::wanted_changed` waits for a reader to take the group served before +### `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 a reader took every group the fill accepted, or stopped waiting for it, so a handler writes a batch one group at a time at its fastest reader's pace. +`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: an ordered reader may take a fill's group only after another fill answers an earlier one, which a handler serving fills one after another would never reach. +Serve the fills you take concurrently: a fill waits on readers that may first need another fill's groups. ### Socket readers start at the beginning diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index e3d73af8a54..6546a6a9e6a 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -11153,9 +11153,9 @@ impl kio::Task for FillServeRun { if self.fill.poll_withdrawn(waiter).is_ready() { return Poll::Ready(()); } - // The next fetch waits for a reader to take the group the last one brought, so a + // 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_taken(waiter)); + ready!(self.fill.poll_paced(waiter)); let Some(sequence) = self.fill.wanted() else { return Poll::Ready(()); }; diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index a57de6c5f76..d2d7d2e7621 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -19943,7 +19943,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()); @@ -19972,7 +19973,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()); @@ -20650,14 +20657,16 @@ mod test { while let Poll::Ready(Ok(fill)) = dynamic.poll_requested_fill(&waiter) { 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; + 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, } - Poll::Ready(None) => break false, - Poll::Pending => break true, } }); }; @@ -20769,6 +20778,36 @@ mod test { 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 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 a374711122b..1f6b73f933f 100644 --- a/rs/moq-net/src/model/track/fill.rs +++ b/rs/moq-net/src/model/track/fill.rs @@ -54,9 +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 let go of yet, which holds back the next + // What its handler accepted that no join has picked up yet, which holds back the next // group: see [`Fill::poll_wanted_changed`]. - untaken: RangeSet, + 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,11 +76,10 @@ impl Pending { } } - /// Count one join out of reading `groups`, which it counted in: it took them or stopped - /// waiting, either of which lets its handler move on. Returns the runs nobody reads any - /// more. + /// 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.untaken.remove(groups.clone()); + self.unseen.remove(groups.clone()); let counts: Vec<(ops::Range, u32)> = self .wants .overlapping(&groups) @@ -354,6 +353,19 @@ 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(); + if let Some(pending) = fills.pending.get_mut(&self.id) + && self.is_ours(pending) + && groups.iter().any(|run| pending.unseen.overlaps(run)) + { + for run in groups.iter() { + pending.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 @@ -529,7 +541,7 @@ impl FillState { queued: true, flight: false, wants: RangeMap::from_iter([(groups.clone(), 1)]), - untaken: RangeSet::new(), + unseen: RangeSet::new(), result, }, ); @@ -580,7 +592,7 @@ impl FillState { queued: true, flight: true, wants: RangeMap::from_iter([(groups.clone(), 1)]), - untaken: RangeSet::new(), + unseen: RangeSet::new(), result: Default::default(), }, ); @@ -630,7 +642,7 @@ impl FillState { queued: true, flight: false, wants: RangeMap::new(), - untaken: RangeSet::new(), + unseen: RangeSet::new(), result, }, ); @@ -955,7 +967,7 @@ 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. It does not wait for the readers to - /// take what was served before it; [`wanted_changed`](Self::wanted_changed) does. + /// pick up what was served before it; [`wanted_changed`](Self::wanted_changed) does. pub fn wanted(&self) -> Option { self.wanted_except(None) } @@ -999,9 +1011,9 @@ impl Fill { } } - /// The next group to serve: [`wanted`](Self::wanted) once a reader took every group this - /// fill accepted, or stopped waiting for it, and once it differs from what this handle - /// last returned. The first call returns at once unless a group is still untaken. + /// The next group to serve: [`wanted`](Self::wanted) once every group this fill accepted + /// reached a reader, which picks it up as it asks for its next group, and once it differs + /// from what this handle last returned. The first call returns at once. /// /// So `while let Some(sequence) = fill.wanted_changed().await` writes one group at a time, /// at the pace of the fastest reader, and ends once the fill is withdrawn or answered. A @@ -1018,17 +1030,17 @@ 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) || (wanted.is_some() && self.inner.is_untaken()) { + if self.prev == Some(wanted) || (wanted.is_some() && self.inner.is_unseen()) { return Poll::Pending; } self.prev = Some(wanted); Poll::Ready(wanted) } - /// Poll until a reader took every group this fill accepted or stopped waiting for it, or - /// nobody waits on the fill, as [`Self::poll_wanted_changed`] paces. + /// 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_taken(&self, waiter: &kio::Waiter) -> Poll<()> { + pub(crate) fn poll_paced(&self, waiter: &kio::Waiter) -> Poll<()> { if self.poll_withdrawn(waiter).is_ready() { return Poll::Ready(()); } @@ -1036,7 +1048,7 @@ impl Fill { self.inner .fills .poll(waiter, |fills| match fills.pending.get(&id) { - Some(pending) if !pending.untaken.is_empty() => Poll::Pending, + Some(pending) if !pending.unseen.is_empty() => Poll::Pending, _ => Poll::Ready(()), }) .map(drop) @@ -1182,13 +1194,13 @@ impl Inner { self.result.wanted() && self.fills.read().wants(self.id, sequence) } - /// Whether a group this fill accepted still waits for a reader to take it. - fn is_untaken(&self) -> bool { + /// 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.untaken.is_empty()) + .is_some_and(|pending| !pending.unseen.is_empty()) } fn max_delay(&self) -> Option { @@ -1272,12 +1284,12 @@ impl Inner { for range in fills.routed(self.id, &(sequence..end)) { fills.covers.remove(range); } - // Written for a reader that has yet to take it, which the next group waits on. + // Written for a reader that has yet to pick it up, which the next group waits on. if res.is_ok() && let Some(pending) = fills.pending.get_mut(&self.id) && pending.wants.contains_key(&sequence) { - pending.untaken.insert(sequence..end); + pending.unseen.insert(sequence..end); } drop(fills); res diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index 7efe9a9c444..161f749bb59 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -538,6 +538,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 +546,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 +565,10 @@ impl Misses { } } drop(state); + // Picked up, so its handler may serve the next group. + if !seen.is_empty() { + self.joins[&key].saw(&seen); + } self.release(&taken); // Judged again: the same join may have more to answer, or park the waiter now. } diff --git a/swift/Sources/Moq/Requests.swift b/swift/Sources/Moq/Requests.swift index d32f40d5209..3b83d320be9 100644 --- a/swift/Sources/Moq/Requests.swift +++ b/swift/Sources/Moq/Requests.swift @@ -87,8 +87,8 @@ public final class Fill: Sendable { ffi.wanted() } - /// Suspend until a reader took every group this fill accepted and `wanted` changed, and - /// return it. The first call returns at once unless an accepted group still waits. + /// Suspend until every group this fill accepted reached a reader and `wanted` changed, and + /// return it. The first call returns at once. public func wantedChanged() async throws -> UInt64? { try await ffi.wantedChanged() } From 0a4aaf49d291a7a84ed9f0242b57d53da8f0a9f0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 01:59:08 -0800 Subject: [PATCH 4/8] chore(dart): regenerate bindings for the paced fill docs Co-Authored-By: Claude Opus 5.5 --- dart/moq_ffi/lib/src/moq.dart | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index e157887146f..74c4037918d 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() != 12272) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqgroupproducer_abort() != 59787) { From 4379684a2e7ef423c21fe78a1ccb9bb53b7b74e1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 02:07:56 -0800 Subject: [PATCH 5/8] fix(net): record a fill's unseen group before answering its readers Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track/fill.rs | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/rs/moq-net/src/model/track/fill.rs b/rs/moq-net/src/model/track/fill.rs index 1f6b73f933f..de88daab611 100644 --- a/rs/moq-net/src/model/track/fill.rs +++ b/rs/moq-net/src/model/track/fill.rs @@ -1272,6 +1272,14 @@ impl Inner { deferred.wake(); res }); + // Written for a reader that has yet to pick it up, which the next group waits on. + // Recorded before the answer goes out, so a reader picking it up at once clears it. + if res.is_ok() + && let Some(pending) = self.fills.lock().pending.get_mut(&self.id) + && pending.wants.contains_key(&sequence) + { + pending.unseen.insert(sequence..end); + } if let Ok(mut outcome) = self.result.write() { outcome.accept(sequence..end); if let Ok(group) = &res @@ -1280,18 +1288,7 @@ impl Inner { outcome.partial.insert(sequence, group.handle()); } } - let mut fills = self.fills.lock(); - for range in fills.routed(self.id, &(sequence..end)) { - fills.covers.remove(range); - } - // Written for a reader that has yet to pick it up, which the next group waits on. - if res.is_ok() - && let Some(pending) = fills.pending.get_mut(&self.id) - && pending.wants.contains_key(&sequence) - { - pending.unseen.insert(sequence..end); - } - drop(fills); + uncover(&self.fills, self.id, sequence..end); res } From 516b90e8f401090687fe67506daed28022b66280 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 08:00:34 -0800 Subject: [PATCH 6/8] fix(net): pace a fill only on what a read picks up, recorded before it is cached A fill records a group unseen before caching it, so a reader taking it from the cache and releasing it before the answer goes out clears it. A drop poll's probe takes in answers without the publisher reading them, so pickups wait for the next read. A fetch joined to a batch fill picks up its group as it resolves. moq-transcode serves fills through wanted_changed, and the docs and every binding's MIGRATE.md describe the pacing. Co-Authored-By: Claude Opus 5.5 --- dart/moq/MIGRATE.md | 4 ++ dart/moq_ffi/lib/src/moq.dart | 2 +- go/wrapper/MIGRATE.md | 4 ++ go/wrapper/publish.go | 9 ++- kt/moq/MIGRATE.md | 4 ++ py/moq-rs/MIGRATE.md | 4 ++ py/moq-rs/moq/publish.py | 8 ++- rs/moq-ffi/MIGRATE.md | 4 ++ rs/moq-ffi/src/producer.rs | 10 ++-- rs/moq-net/MIGRATE.md | 3 +- rs/moq-net/src/model/track.rs | 96 +++++++++++++++++++++++++++++- rs/moq-net/src/model/track/fill.rs | 57 +++++++++++------- rs/moq-net/src/model/track/miss.rs | 16 ++++- rs/moq-transcode/src/rung.rs | 10 ++-- swift/MIGRATE.md | 4 ++ swift/Sources/Moq/Requests.swift | 13 ++-- 16 files changed, 202 insertions(+), 46 deletions(-) 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 74c4037918d..42518706383 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -24714,7 +24714,7 @@ void _checkApiChecksums() { 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() != 12272) { + 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 ccd838749ac..07282c691d9 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -322,7 +322,8 @@ func (d *TrackRequests) Cancel() { // 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 } @@ -365,8 +366,10 @@ func (f *Fill) Wanted() *uint64 { return f.inner.Wanted() } -// WantedChanged blocks until every group this fill accepted reached a reader -// and Wanted changed, 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 775240b2745..b1498a592da 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -213,7 +213,8 @@ class Fill: 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: @@ -260,8 +261,9 @@ def wanted(self) -> int | None: async def wanted_changed(self) -> int | None: """The next group to serve, once every group this fill accepted reached a reader. - Returns ``wanted`` once it changed and the readers picked up what was served; the first - call returns at once. + 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() 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 d6ea82dda3b..acf270ced14 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -482,7 +482,8 @@ impl MoqTrackRequests { /// [`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, @@ -536,9 +537,10 @@ impl MoqFill { self.inner.wanted() } - /// The next group to serve: `wanted` once every group this fill accepted reached a reader - /// and it changed. The first call returns at once. Serve one group per call, so the cache - /// never holds a batch the readers have yet to take. + /// 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 7cfd1675b4b..d5910bb5252 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -8,7 +8,8 @@ How to update your code for each breaking change, newest first. `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: a fill waits on readers that may first need another fill's groups. +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. ### `track::Consumer::unmetered` reads outside any stats session diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 61a1315a84d..17f20185381 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) } @@ -20816,6 +20836,80 @@ mod test { 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 de88daab611..39b8b64a720 100644 --- a/rs/moq-net/src/model/track/fill.rs +++ b/rs/moq-net/src/model/track/fill.rs @@ -356,13 +356,17 @@ impl Join { /// 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(); - if let Some(pending) = fills.pending.get_mut(&self.id) - && self.is_ours(pending) - && groups.iter().any(|run| pending.unseen.overlaps(run)) + // 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))) { - for run in groups.iter() { - pending.unseen.remove(run.clone()); - } + return; + } + let unseen = &mut fills.pending.get_mut(&self.id).expect("checked above").unseen; + for run in groups.iter() { + unseen.remove(run.clone()); } } @@ -811,8 +815,9 @@ pub(super) fn poll_requested( /// [`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, one group at a time as -/// [`wanted_changed`](Self::wanted_changed) yields them, so the cache never holds a batch its -/// readers have yet to take. Dropping the fill names +/// [`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 @@ -1011,14 +1016,15 @@ impl Fill { } } - /// The next group to serve: [`wanted`](Self::wanted) once every group this fill accepted - /// reached a reader, which picks it up as it asks for its next group, and once it differs - /// from what this handle last returned. The first call returns at once. + /// 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` writes one group at a time, - /// at the pace of the fastest reader, and ends once the fill is withdrawn or answered. A - /// batch written ahead of its readers would let the cache evict its earlier groups before - /// they take them. + /// 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: @@ -1261,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 { @@ -1272,13 +1291,11 @@ impl Inner { deferred.wake(); res }); - // Written for a reader that has yet to pick it up, which the next group waits on. - // Recorded before the answer goes out, so a reader picking it up at once clears it. - if res.is_ok() + if unseen + && res.is_err() && let Some(pending) = self.fills.lock().pending.get_mut(&self.id) - && pending.wants.contains_key(&sequence) { - pending.unseen.insert(sequence..end); + pending.unseen.remove(sequence..end); } if let Ok(mut outcome) = self.result.write() { outcome.accept(sequence..end); diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index 161f749bb59..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); } } } @@ -565,9 +569,8 @@ impl Misses { } } drop(state); - // Picked up, so its handler may serve the next group. if !seen.is_empty() { - self.joins[&key].saw(&seen); + 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. @@ -582,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 103ccf1857d..079aabce7a4 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -561,16 +561,14 @@ async fn fetches( fn spawn_fill( tasks: &mut tokio::task::JoinSet<()>, rung: Rung, - fill: moq_net::track::Fill, + mut fill: moq_net::track::Fill, permit: tokio::sync::OwnedSemaphorePermit, ) { tasks.spawn(async move { let _permit = permit; - let mut last = None; - while let Some(sequence) = fill.wanted() - && last != Some(sequence) - { - last = Some(sequence); + // 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. + while let Some(sequence) = fill.wanted_changed().await { if let Err(err) = fetch(&rung, &fill, sequence).await { tracing::warn!(%err, sequence, "transcode fetch failed"); } 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 3b83d320be9..6c4bc066d67 100644 --- a/swift/Sources/Moq/Requests.swift +++ b/swift/Sources/Moq/Requests.swift @@ -40,9 +40,11 @@ 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, 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`. +/// 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 @@ -87,8 +89,9 @@ public final class Fill: Sendable { ffi.wanted() } - /// Suspend until every group this fill accepted reached a reader and `wanted` changed, 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() } From 614670d66ea2c4fc5b6a73a2c93642216de00f8f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 08:30:37 -0800 Subject: [PATCH 7/8] fix(transcode): hold a fetch slot per group, not across a paced fill's wait Co-Authored-By: Claude Opus 5.5 --- rs/moq-transcode/src/rung.rs | 52 +++++++++++------------------------- 1 file changed, 15 insertions(+), 37 deletions(-) diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 079aabce7a4..5cada6282ac 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,18 +540,14 @@ 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, - mut 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) { tasks.spawn(async move { - let _permit = permit; // 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. + // 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. while let Some(sequence) = fill.wanted_changed().await { + 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"); } @@ -581,7 +560,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> { @@ -995,8 +974,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 From 708d032cafff44399f9709de432fd2f1392a084b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 09:06:27 -0800 Subject: [PATCH 8/8] fix(transcode): serve a retired rung's fills without waiting on their readers Retirement drains every fill before the track ends, and a paced fill parks until a reader picks up its last group. A reader that stopped reading kept the drain, and so the retired track, open forever. Once retired, a fill now serves what it still owes unpaced, once per group. Co-Authored-By: Claude Opus 5.5 --- rs/moq-transcode/src/rung.rs | 78 ++++++++++++++++++++++++++++++++++-- 1 file changed, 74 insertions(+), 4 deletions(-) diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 057b14c00d7..098f608ce7a 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -542,11 +542,24 @@ async fn fetches( /// 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 { - // 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. - while let Some(sequence) = fill.wanted_changed().await { + let mut last = None; + 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"); @@ -991,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)]