diff --git a/quest/m1/fuzz-findings/README.md b/quest/m1/fuzz-findings/README.md index 9dc7400afcf..d20a1a4f31c 100644 --- a/quest/m1/fuzz-findings/README.md +++ b/quest/m1/fuzz-findings/README.md @@ -1,4 +1,4 @@ -# Media fuzz findings +# [S] Media fuzz findings ## Goal @@ -16,9 +16,9 @@ maintainer). Its short local runs found the bugs the children fix; the minimized inputs are in [its PR comment](https://github.com/moq-dev/moq/pull/7313#issuecomment-6520535160). Each child commits its input under `regressions/` with its fix, since an -input committed before the fix would fail the merge gate. After the last -child, run every target nightly for a few sessions; a new finding becomes a -child here. +input committed before the fix would fail the merge gate. Every known finding +is now fixed: run every target nightly for a few sessions; a new finding +becomes a child here. For the crates outside the `kixelated` and `moq-dev` orgs, moq-mux fixes the bug on its own side and the bug is reported upstream (decided @@ -33,7 +33,3 @@ maintainer's approval. webm-iterable's block parsing with moq-mux's own block-header reader, which fixed the MKV block-header panic; its input is `rs/moq-mux/fuzz/regressions/mkv/short-block-header`. - -## Required - -- [A catalog update reuses its encoder](/quest/m1/fuzz-findings/catalog-encoder.md) - republishing a compressed catalog or HLS group stops allocating a fresh DEFLATE encoder (about 200 KB) each time diff --git a/quest/m1/fuzz-findings/catalog-encoder.md b/quest/m1/fuzz-findings/catalog-encoder.md deleted file mode 100644 index 3c7cb1a32de..00000000000 --- a/quest/m1/fuzz-findings/catalog-encoder.md +++ /dev/null @@ -1,30 +0,0 @@ -# [M] A catalog update reuses its encoder - -## Goal - -Republishing a compressed track's latest value, as every hang catalog and -HLS update does, no longer allocates a fresh DEFLATE encoder (about 200 KB) per -group. In the `annexb` target, the 202-byte input that caused 24 encoders -and about 4.9 MB of allocation stays inside the fuzz bound, and is committed -under `rs/moq-mux/fuzz/regressions/`. - -## Plan - -Decided 2026-10-30 (maintainer): the encoder per update is a bug, not an -accepted cost, so the fuzz bound gets no per-update allowance. - -Each compressed group is its own DEFLATE stream (draft-lcurley-moq-flate), -so `Flate::deflate` in `rs/moq-net/src/model/group.rs` creates a -`flate::Encoder` on a group's first write and drops it at `finish`. A -publisher that writes one frame per group, as a catalog does, pays the -allocation on every update. - -Recommended: keep a finished group's encoder, reset with `Compress::reset`, -on the track's cache account (`cache::Track`) and hand it to the next group -the same track opens, billed to the pool while parked so the cache budget -still counts it. At most one spare per track, so concurrent groups still -allocate their own. The spare lives as long as the track; say in the PR -whether idle compressed tracks holding it is acceptable. The wire format -does not change: a reset encoder starts a cold window, as a new one does. - -Public API: none. Wire: none. diff --git a/rs/moq-mux/fuzz/regressions/annexb/catalog-encoder b/rs/moq-mux/fuzz/regressions/annexb/catalog-encoder new file mode 100644 index 00000000000..755bb14bf1d Binary files /dev/null and b/rs/moq-mux/fuzz/regressions/annexb/catalog-encoder differ diff --git a/rs/moq-net/benches/group.rs b/rs/moq-net/benches/group.rs index 368a682e39a..847accd20c0 100644 --- a/rs/moq-net/benches/group.rs +++ b/rs/moq-net/benches/group.rs @@ -30,6 +30,9 @@ //! reader, as a paced ingest runs, swept over readers: each frame read reports progress, //! and the fastest reader's read wakes the parked writer. //! +//! `group_compressed_updates` republishes single-frame compressed groups, swept over +//! tracks and overlapping groups, to expose encoder allocation on the update path. +//! //! Run with `cargo bench -p moq-net`. #![expect( @@ -503,6 +506,56 @@ fn bench_inflate(c: &mut Criterion) { g.finish(); } +/// Repeated single-frame compressed updates on independent tracks. Groups overlap +/// within each update so only the one spare per track can be reused. +fn bench_compressed_updates(c: &mut Criterion) { + const UPDATES: usize = 32; + let payload = Bytes::from(format!("{{\"status\":\"{}\"}}", "ready".repeat(100))); + let mut g = c.benchmark_group("group_compressed_updates"); + for tracks in [1usize, 16] { + for concurrent in [1usize, 4] { + g.throughput(Throughput::Elements((UPDATES * tracks * concurrent) as u64)); + g.bench_function(BenchmarkId::new(format!("tracks{tracks}"), concurrent), |b| { + b.iter_batched( + || { + let broadcast = broadcast::Info::default().produce(); + let producers: Vec<_> = (0..tracks) + .map(|i| { + let track = broadcast + .create_track(format!("bench{i}"), track::Info::default().with_compress(true)) + .unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(None, payload.clone()).unwrap(); + group.finish().unwrap(); + track + }) + .collect(); + (broadcast, producers, Vec::with_capacity(concurrent)) + }, + |(broadcast, producers, mut groups)| { + for _ in 0..UPDATES { + for track in &producers { + for _ in 0..concurrent { + let mut group = track.append_group().unwrap(); + group.write_frame(None, payload.clone()).unwrap(); + groups.push(group); + } + for group in &mut groups { + group.finish().unwrap(); + } + groups.clear(); + } + } + (broadcast, producers, groups) + }, + BatchSize::SmallInput, + ); + }); + } + } + g.finish(); +} + /// Worker threads the contended readers are spread over. const THREADS: [usize; 3] = [1, 4, 16]; @@ -952,6 +1005,7 @@ criterion_group!( bench_stamp, bench_relay_frames, bench_inflate, + bench_compressed_updates, bench_contended, bench_streaming, bench_shed, diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index 1666325ee98..2269c660bfa 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -47,7 +47,7 @@ use std::task::{Poll, ready}; use std::time::Duration; use super::track::{self, Position, TrackState}; -use super::{aborts, expiry, group}; +use super::{aborts, expiry, flate, group}; /// Fixed bookkeeping charged per cached group on top of its frame payload bytes. /// @@ -719,6 +719,10 @@ pub(super) struct Track { // What the frames this track's groups are still receiving may allocate up front, // shared with every track minted from the same origin handle. frames: super::frame::Budget, + + // One finished local group's encoder, kept until another group needs it or the + // account drops. Raw codec only: an Extra here would keep this account alive. + encoder: Mutex>, } /// The greatest location a track has written or its source reported, followed by the track @@ -825,6 +829,7 @@ impl Track { aborts: OnceLock::new(), largest: Default::default(), declared: AtomicBool::new(false), + encoder: Mutex::default(), }); if let Some(key) = track.pool.register(&track) { let _ = track.sweep.set(key); @@ -1018,6 +1023,15 @@ impl Track { impl Drop for Track { fn drop(&mut self) { + if self + .encoder + .get_mut() + .unwrap_or_else(PoisonError::into_inner) + .take() + .is_some() + { + self.pool.sub(flate::ENCODER_HEAP); + } if let Some(key) = self.sweep.get() { self.pool.unregister(*key); } @@ -1288,6 +1302,34 @@ pub(super) struct Extra { } impl Extra { + /// Take this track's spare encoder with its charge, or bill a new one. + pub(super) fn take_encoder(&self) -> flate::Encoder { + if let Some(track) = &self.track { + let encoder = track.encoder.lock().unwrap_or_else(PoisonError::into_inner).take(); + if let Some(mut encoder) = encoder { + // Moving existing heap is not a write and must not accrue eviction debt. + self.bytes.fetch_add(flate::ENCODER_HEAP, Ordering::Relaxed); + encoder.reset(); + return encoder; + } + } + self.add(flate::ENCODER_HEAP); + flate::Encoder::new() + } + + /// Park a finished encoder and its charge, dropping it if the track has a spare. + pub(super) fn park_encoder(&self, encoder: flate::Encoder) { + if let Some(track) = &self.track { + let mut spare = track.encoder.lock().unwrap_or_else(PoisonError::into_inner); + if spare.is_none() { + self.bytes.fetch_sub(flate::ENCODER_HEAP, Ordering::Relaxed); + *spare = Some(encoder); + return; + } + } + self.sub(flate::ENCODER_HEAP); + } + /// Bill `n` more bytes. pub(super) fn add(&self, n: u64) { if let Some(track) = &self.track { @@ -1350,6 +1392,42 @@ mod test { Pool::new(config) } + /// Reuse transfers existing heap, concurrent users allocate separately, and only + /// one finished encoder stays parked until the account's final owner drops. + #[test] + fn encoder_spare_transfers_its_charge_without_write_debt() { + let pool = Pool::unbounded(); + let charge = charge(&pool); + let first = charge.extra(); + let second = charge.extra(); + let account = charge.track.as_ref().unwrap(); + account.take_written(); + let encoder = first.take_encoder(); + let concurrent = second.take_encoder(); + assert_eq!(pool.used(), ENTRY_OVERHEAD + 2 * flate::ENCODER_HEAP); + assert_eq!(account.take_written(), 2 * flate::ENCODER_HEAP); + + first.park_encoder(encoder); + assert_eq!(first.size(), 0); + assert_eq!(pool.used(), ENTRY_OVERHEAD + 2 * flate::ENCODER_HEAP); + let encoder = first.take_encoder(); + assert_eq!(first.size(), flate::ENCODER_HEAP); + assert_eq!(account.take_written(), 0, "reuse must not inflate write debt"); + first.park_encoder(encoder); + second.park_encoder(concurrent); + assert_eq!(second.size(), 0); + assert_eq!(pool.used(), ENTRY_OVERHEAD + flate::ENCODER_HEAP, "only one spare"); + + // A group and its inflated copy may outlive the track. They still own its + // account, so the parked encoder stays billed until that last owner drops. + drop(charge); + assert_eq!(pool.used(), flate::ENCODER_HEAP); + drop(first); + assert_eq!(pool.used(), flate::ENCODER_HEAP); + drop(second); + assert_eq!(pool.used(), 0); + } + struct Flag(std::sync::atomic::AtomicBool); impl std::task::Wake for Flag { diff --git a/rs/moq-net/src/model/flate.rs b/rs/moq-net/src/model/flate.rs index 14a139e027f..b178c142c77 100644 --- a/rs/moq-net/src/model/flate.rs +++ b/rs/moq-net/src/model/flate.rs @@ -26,9 +26,10 @@ const CHUNK: usize = 8 * 1024; /// The DEFLATE window, 2^15 bytes for raw DEFLATE at the default `windowBits`. const WINDOW: u64 = 1 << 15; -/// Heap an [`Encoder`] holds, as zlib sizes it at the default `memLevel` of 8: the doubled -/// window, its `prev` chain and hash `head` tables, and the pending buffer, 64 KiB each. -pub(super) const ENCODER_HEAP: u64 = 8 * WINDOW; +/// Heap an [`Encoder`] holds with zlib-rs at the default `memLevel` of 8: 64 KiB each +/// for the doubled window, `prev` chain and pending buffer, a 128 KiB hash table and +/// a separate 48 KiB symbol buffer, plus the codec state and alignment rounded to 16 KiB. +pub(super) const ENCODER_HEAP: u64 = 12 * WINDOW; /// Heap a [`Decoder`] holds: the window, plus its inflate state rounded up to 8 KiB. pub(super) const DECODER_HEAP: u64 = WINDOW + 8 * 1024; @@ -44,6 +45,11 @@ impl Encoder { Self(Box::new(Compress::new(flate2::Compression::new(LEVEL), false))) } + /// Start another group with a cold window, retaining the codec's allocations. + pub(super) fn reset(&mut self) { + self.0.reset(); + } + /// Compress the next frame, returning its slice minus the sync-flush marker. Empty in /// yields empty out, leaving the window untouched. pub(super) fn frame(&mut self, payload: &[u8]) -> Bytes { diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index e7d9164d635..1039f909ba6 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -391,18 +391,15 @@ impl Flate { /// Compress a locally written frame, returning its stored slice. fn deflate(&mut self, payload: &[u8]) -> Bytes { let plain = &self.plain; - let encoder = self.encoder.get_or_insert_with(|| { - plain.bill.add(flate::ENCODER_HEAP); - flate::Encoder::new() - }); + let encoder = self.encoder.get_or_insert_with(|| plain.bill.take_encoder()); encoder.frame(payload) } - /// Drop the codec state once no frame follows at `fin`: the encoder, and the decoder if - /// its readers already inflated every frame. + /// Park the encoder once no frame follows at `fin`, and drop the decoder if its readers + /// already inflated every frame. fn finish(&mut self, fin: Option) { - if self.encoder.take().is_some() { - self.plain.bill.sub(flate::ENCODER_HEAP); + if let Some(encoder) = self.encoder.take() { + self.plain.bill.park_encoder(encoder); } let Some(fin) = fin else { return; @@ -6276,6 +6273,47 @@ mod test { ); } + /// Every group starts cold even when a finished encoder is reused. Open groups + /// keep independent windows, and aborting one does not consume another's spare. + #[test] + fn reused_encoders_start_independent_group_streams() { + let pool = cache::Pool::unbounded(); + let account = cache::Track::new(pool.clone(), Default::default(), kio::Weak::new()); + let info = track::Info::default().with_compress(true); + let mut concurrent = Producer::new(0u64.into(), info.clone(), account.clone()); + concurrent.write_frame(None, verbose(0)).unwrap(); + for sequence in 1..5u64 { + let mut group = Producer::new(sequence.into(), info.clone(), account.clone()); + let mut reference = flate::Encoder::new(); + let frames = [verbose(0), Bytes::new(), verbose(sequence as usize)]; + let slices: Vec<_> = frames.iter().map(|frame| reference.frame(frame)).collect(); + for frame in &frames { + group.write_frame(None, frame.clone()).unwrap(); + } + group.finish().unwrap(); + group.finish().unwrap(); + let mut raw = group.consume().compressed(); + let mut decoder = flate::Decoder::new(); + for (frame, expected) in frames.iter().zip(&slices) { + let Poll::Ready(Ok(Some(object))) = raw.poll_read_object(&kio::Waiter::noop()) else { + panic!("a stored slice is ready"); + }; + let slice = object.frame.payload.contiguous(0); + assert_eq!(slice, *expected, "reuse must emit a cold stream"); + assert_eq!(decoder.frame(&slice, frame.len()).unwrap(), *frame); + } + } + concurrent.write_frame(None, verbose(1)).unwrap(); + let mut reader = concurrent.consume(); + assert_eq!(read_plain(&mut reader).unwrap().unwrap(), verbose(0)); + assert_eq!(read_plain(&mut reader).unwrap().unwrap(), verbose(1)); + concurrent.abort(Error::Cancel).unwrap(); + drop(reader); + assert_eq!(pool.used(), flate::ENCODER_HEAP, "the spare survives a sibling's abort"); + drop(account); + assert_eq!(pool.used(), 0); + } + /// A compressed group stores DEFLATE slices, hands plain readers the frames as written /// from one shared inflated copy, and hands egress the slices with their sizes. #[test] @@ -6306,7 +6344,7 @@ mod test { } reference.finish().unwrap(); assert!( - stored + (plain as u64) / 2 < uncompressed.used(), + stored - flate::ENCODER_HEAP + (plain as u64) / 2 < uncompressed.used(), "slices are stored: {stored} against {}", uncompressed.used() ); @@ -6533,8 +6571,8 @@ mod test { assert_eq!(metered(), stored); } - /// The codec state counts toward the pool while it lives: the encoder until the group - /// finishes, and a reader's decoder until the last frame inflates. + /// The codec state counts toward the pool while it lives: the encoder moves to its + /// track when the group finishes, and a decoder drops when the last frame inflates. #[test] #[cfg(feature = "async")] fn codec_state_is_billed_while_held() { @@ -6555,7 +6593,11 @@ mod test { assert_eq!(group.handle.cache_size(), held, "an open group keeps both"); group.finish().unwrap(); - assert_eq!(group.handle.cache_size(), stored + inflated, "finishing drops both"); + assert_eq!( + group.handle.cache_size(), + stored + inflated, + "finishing releases both from the group" + ); assert_eq!(read_plain(&mut reader).unwrap().unwrap(), verbose(1)); // A finished group's decoder goes with the last frame it inflates. @@ -6862,13 +6904,17 @@ mod test { group.abort(Error::Evicted).unwrap(); assert_eq!( pool.used(), - Flate::RELEASE_HEAP, - "the job still owns the new guard allocation" + Flate::RELEASE_HEAP + flate::ENCODER_HEAP, + "the job owns its guard, and the account owns its spare" ); assert!(matches!(job.get(0), Err(Error::Evicted))); drop(job); - assert_eq!(pool.used(), Flate::RELEASE_HEAP, "clones share one intrinsic charge"); + assert_eq!( + pool.used(), + Flate::RELEASE_HEAP + flate::ENCODER_HEAP, + "clones share one intrinsic charge" + ); drop(retained); - assert_eq!(pool.used(), 0, "the final guard releases its own charge"); + assert_eq!(pool.used(), 0, "the final guard also releases its account's spare"); } }