diff --git a/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md b/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md index 21eff93ccd..7ab04e037c 100644 --- a/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md +++ b/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md @@ -90,4 +90,3 @@ the closed producer's cache or terminal state. ## Related - [#4491](https://github.com/moq-dev/moq/pull/4491) - edits `resume.rs` and its takeover tests -- [Parked reads wake](/quest/m1/parked-read-wakes.md) - edits the same `resume.rs` wakeups diff --git a/quest/m1/README.md b/quest/m1/README.md index fc4f552346..e2f0be1687 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -51,7 +51,6 @@ QUIC studies there on that rule. - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges in either order and replaces FETCH; relays fill misses by range, including over moq-transport - [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - bursty small-group tracks cross two relays without lost groups, unanswered FETCHes, or stalls - [Late lower groups](/quest/m1/lite-late-lower-group.md) - a moq-lite subscriber with a floor receives a group created below the first served one, as moq-transport does -- [Parked reads wake](/quest/m1/parked-read-wakes.md) - a read parked on an evicted or aborted group wakes and re-judges, for resume successors and plain tracks alike - [Reader end parity](/quest/m1/reader-end-parity.md) - JS readers see a track's end once the newest group reaches the declared end, as Rust readers do - [JS cache window](/quest/m1/js-prune-cache-window.md) - js/net ages idle groups on a cache window and keeps max_age as media-time staleness, as Rust does - [IETF FETCH abandonment](/quest/m1/ietf-fetch-abandonment.md) - the Rust moq-transport subscriber cancels a group FETCH once its last reader leaves diff --git a/quest/m1/parked-read-wakes.md b/quest/m1/parked-read-wakes.md deleted file mode 100644 index 77b75479a2..0000000000 --- a/quest/m1/parked-read-wakes.md +++ /dev/null @@ -1,30 +0,0 @@ -# [S] Parked reads wake - -## Goal - -A read parked on a group that is evicted or aborted always wakes and -re-judges, for resume successors and plain tracks alike, pinned by mock-time -regression tests. - -## Plan - -Follow-ups from [#4484](https://github.com/moq-dev/moq/pull/4484), which made -`resume::Successor` recompute its answer on every judgment: - -- Its regression tests cover an aborted unstamped successor only. Eviction - wakes through a different path, so add a resume-level test where the - successor is evicted before its first frame. If it already passes, this - item lands as a test. -- A stamped group that is aborted without being evicted wakes no parked - reader, on a plain track as well as a successor. It only matters under a - timestamp rewind. Reproduce it first, then wake the waiters at the source. - -Settled: one quest, since both are the same wakeup invariant in -`rs/moq-net/src/model/track.rs` and `resume.rs`. - -Public API: none. Wire: none. - -## Related - -- [#4491](https://github.com/moq-dev/moq/pull/4491) - a resumed group holds demand on its replacement copy in the same `resume.rs` wakeups -- [#2991](/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md) - extends the `resume.rs` takeover tests diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index 14f7661d3b..637157a266 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -320,6 +320,46 @@ fn bench_subscriber_join(c: &mut Criterion) { group.finish(); } +/// Parked reads re-judge their reach, including the successor's abort waiter. +/// Sweep readers and cached groups so added per-reader scans show up as a slope. +fn bench_parked_read(c: &mut Criterion) { + let mut group = c.benchmark_group("track_parked_read"); + for cached in [8, 64, 512] { + for readers in [1, 8, 64] { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", None).unwrap(); + let mut head = track.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, Bytes::from_static(b"head")).unwrap(); + for _ in 1..cached { + let mut next = track.append_group().unwrap(); + next.write_frame(Timestamp::ZERO, Bytes::from_static(b"next")).unwrap(); + next.finish().unwrap(); + } + let waiters: Vec<_> = (0..readers).map(|_| kio::Waiter::noop()).collect(); + let mut held: Vec<_> = waiters + .iter() + .map(|waiter| { + let mut sub = track.subscribe(track::Subscription::default().with_max_age(Duration::from_secs(1))); + let Poll::Ready(Ok(Some(mut head))) = sub.poll_recv_group(waiter) else { + panic!("head is cached"); + }; + assert!(matches!(head.poll_read_frame(waiter), Poll::Ready(Ok(Some(_))))); + head + }) + .collect(); + group.throughput(Throughput::Elements(readers as u64)); + group.bench_function(BenchmarkId::new(format!("cached_{cached}"), readers), |b| { + b.iter(|| { + for (head, waiter) in held.iter_mut().zip(&waiters) { + assert!(black_box(head.poll_read_frame(waiter)).is_pending()); + } + }); + }); + } + } + group.finish(); +} + criterion_group!( benches, bench_fanout, @@ -327,6 +367,7 @@ criterion_group!( bench_aborted_scan, bench_subscriber_churn, bench_subscriber_churn_after_peak, - bench_subscriber_join + bench_subscriber_join, + bench_parked_read ); criterion_main!(benches); diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index acbd0d00f9..e7a5a2ce23 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -3721,6 +3721,87 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + #[tokio::test] + async fn aborted_stamped_successor_wakes_a_parked_read() { + tokio::time::pause(); + let (a, a_read) = track_pair("a"); + let (mut b, b_read) = track_pair("b"); + let mut head = a.create_group(0u64.into()).unwrap(); + head.write_frame(Timestamp::ZERO, b"a0".to_vec()).unwrap(); + let mut successor = b.create_group(1u64.into()).unwrap(); + successor + .write_frame(Duration::from_secs(30).try_into().unwrap(), b"b1".to_vec()) + .unwrap(); + write_group_at(&mut b, 2, "b2", Duration::from_secs(1)); + write_group_at(&mut b, 3, "edge", Duration::from_secs(20)); + let mut producer = Producer::new(); + producer.switch(a_read, None).unwrap(); + producer.switch(b_read, Position::group(1)).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the rewound successor extends the reach"); + assert_eq!(read(&mut reading), b"a0"); + + let (counter, waker) = CountWaker::new(); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + let before = counter.count(); + successor.abort(Error::Cancel).unwrap(); + assert!( + counter.count() > before, + "the stamped successor's abort lost its wakeup" + ); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + + /// Cache eviction removes an unstamped successor from its track, so the next + /// segment's first group becomes the bound for the read already parked on A. + #[tokio::test] + async fn evicted_unstamped_successor_re_resolves_across_segments() { + tokio::time::pause(); + let (a, a_read) = track_pair("a"); + let pool = crate::cache::Pool::new(crate::cache::Config::default().with_expiry(Duration::from_secs(1))); + let info = broadcast::Info { + pool: pool.clone(), + ..Default::default() + }; + let b = track::Producer::new(Arc::new(info), "b", None); + let (mut c, c_read) = track_pair("c"); + let mut head = a.create_group(0u64.into()).unwrap(); + head.write_frame(Timestamp::ZERO, b"a0".to_vec()).unwrap(); + let successor = b.create_group(1u64.into()).unwrap(); + // The live edge is protected from eviction and lies beyond B's segment. + let _protected = b.create_group(4u64.into()).unwrap(); + write_group_at(&mut c, 2, "c2", Duration::from_secs(1)); + write_group_at(&mut c, 3, "edge", Duration::from_secs(20)); + let mut producer = Producer::new(); + producer.switch(a_read, None).unwrap(); + producer.switch(b.consume(), Position::group(1)).unwrap(); + producer.switch(c_read, Position::group(2)).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the unstamped successor leaves reach unbounded"); + assert_eq!(read(&mut reading), b"a0"); + + let (counter, waker) = CountWaker::new(); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + let before = counter.count(); + crate::model::clock::advance(Duration::from_secs(2)); + pool.gc(crate::model::clock::now()); + assert!(successor.is_aborted(), "cache GC evicted the successor"); + assert!( + b.consume().peek_group(1).is_none(), + "the slot was removed from the cache" + ); + assert!(counter.count() > before, "the successor's eviction lost its wakeup"); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + #[tokio::test] async fn pruned_segment_boundary_is_judged_against_later_segments() { let (mut a, a_read) = track_pair("a"); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 84edd324f1..64496c148a 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2589,10 +2589,16 @@ impl Consumer { ConsumerKind::Plain(state) => { let mut start = None; let _ = state.poll(waiter, |state| { - start = state.first_servable(from, cap).map(|group| { + while let Some(group) = state.first_servable(from, cap) { + // A stamped group can abort without changing its track. Register + // before checking it again, so an abort either wakes or re-resolves. + let _ = group.poll_closed(waiter); let _ = group.poll_timestamp(waiter); - group.timestamp() - }); + if !group.is_aborted() { + start = Some(group.timestamp()); + break; + } + } Poll::<()>::Pending }); start @@ -3363,6 +3369,11 @@ impl group::Expiry for GroupExpiry { let _ = self.state.poll(waiter, |state| { let budget = clamp_max_age(max_age, state.max_age_bound()); loop { + // An abort can change reach even after the successor is stamped. + // Register before judging, so a racing abort is observed or wakes us. + if let Some(group) = state.first_servable(self.sequence.saturating_add(1), cap) { + let _ = group.poll_closed(waiter); + } let edge = state.drift_edge(cap, outer, successor); expired = state.is_stale(self.sequence, &edge, budget); if expired { @@ -6009,6 +6020,39 @@ mod test { assert!(matches!(result, Ok(None)), "the held group ends: {result:?}"); } + /// A rewound successor can keep a drained group within budget until its abort + /// moves the reach to the next cached group, without changing the track itself. + #[tokio::test] + async fn aborted_stamped_successor_wakes_a_parked_read() { + tokio::time::pause(); + let mut producer = track_producer("test", None); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap(); + let mut successor = producer.append_group().unwrap(); + successor + .write_frame(Timestamp::from_millis(30_000).unwrap(), b"next".as_slice()) + .unwrap(); + append_at(&mut producer, 1000); + append_at(&mut producer, 20_000); + let mut sub = producer.subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the rewound successor extends the reach"); + assert!(reading.read_frame().await.unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + successor.abort(Error::Cancel).unwrap(); + assert!( + woken.load(Ordering::SeqCst), + "the stamped successor's abort lost its wakeup" + ); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + /// A first timestamp on a *newer* group can convict a held one, so the held reader has /// to be woken by it. The conviction needs a group beyond the held one's successor: /// a group is bounded by where its successor begins, so the successor itself never