diff --git a/quest/m1/ts-stats-module.md b/quest/m1/ts-stats-module.md index a68a4203fb..7968893eda 100644 --- a/quest/m1/ts-stats-module.md +++ b/quest/m1/ts-stats-module.md @@ -31,10 +31,9 @@ Decided while planning the follow-ups of - Update moq-cli's `publish.rs`, the only consumer that names these types. moq-srt only uses `stats::Log`. - Main quests still add fields under the old names: - [TS PSI reassembly](/quest/m1/ts-psi-reassembly.md), - [TS import health](/quest/m2/ts-import-health.md) and - [TS export liveness](/quest/m2/ts-export-liveness.md). The rename carries - those fields at merge time. Update the type names in the quests still open + [TS PSI reassembly](/quest/m1/ts-psi-reassembly.md) and + [TS import health](/quest/m2/ts-import-health.md). `Export::stats` (#4577) + also returns the old `ts::Stats`. The rename carries those at merge time. Update the type names in the quests still open when this lands, including [media stats schema](/quest/m1/stats/schema.md) and [Rust reporters](/quest/m1/stats/rust.md). diff --git a/quest/m2/README.md b/quest/m2/README.md index c825bbc800..9aaa4dc6fd 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -98,7 +98,6 @@ waiting on an upstream release waits in [m4](/quest/m4/README.md). - [GOP overhead](/quest/m2/gop-overhead.md) - price the I-frames a short GOP pays for, deciding whether a long GOP plus a keyframe request is worth designing - [Per-program SI](/quest/m2/ts-program-si.md) - a selected TS program's broadcast carries only its own service's SDT and EIT - [TS import health](/quest/m2/ts-import-health.md) - `moq import ts` counts the TR 101 290 errors of the feed it receives, PCR and PTS graded on its own values -- [TS export liveness](/quest/m2/ts-export-liveness.md) - `moq export ts` reports each elementary stream's access units and quiet time, catching a per-track stall - [TS health stats](/quest/m2/ts-health-stats.md) - the TS counters ride the stats plumbing beside the media counters - [Teleoperation](/quest/m2/teleop/README.md) - MoQ carries robot video down and control up on one session as a library capability - [Media QA on other engines](/quest/m2/browser-media-qa-engines.md) - the media harness measures a Firefox or WebKit player over the fallback and names what each engine lacks diff --git a/quest/m2/ts-export-liveness.md b/quest/m2/ts-export-liveness.md deleted file mode 100644 index f8c2a92b03..0000000000 --- a/quest/m2/ts-export-liveness.md +++ /dev/null @@ -1,50 +0,0 @@ -# [S] moq export ts reports each elementary stream's access units and how long it has been quiet - -## Goal - -`Export::stats` returns `ts::Stats` with the import's per-PID row (#4502) for -every elementary stream the exporter writes: `units`, the access-unit count, -and `quiet`, the time since the last, on the output's own PCR. An operator can tell that one track stalled -at the exporter while PSI and the other PIDs kept flowing. The other fields of -`ts::Stats` stay zero at egress. `moq export ts` logs a line when a row stops -advancing. - -## Plan - -Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838): - -- **Liveness only, no ETSI counters at egress.** The media-aware lane - regenerates PAT, PMT, PCR and CC, so an egress check grades our muxer, never - the contribution feed. Sync, TEI, CRC, CC, PAT and PMT on bytes the muxer - constructs fail only on a code defect, which the `test/ts` hard checks catch - in CI, and `PAT_error` timed on the output's own PCR never fires on a whole - stall because that clock freezes too. `PCR_accuracy_error` would be - permanently in alarm: the exporter's PCR fails ±500 ns on every file sample - and the downstream groomer's output passes on every one - ([T33](https://github.com/tdrapier-wbd/moq-mpegts-paper/blob/d4c7573f9519a1c9c4882fab2021fbf82616fdcc/lab/test-33-gate2-preparation.md)). - PCR accuracy and wire timing belong to the groomer's monitoring of its own - output, outside this tree. -- **The runtime failure worth catching is a per-track stall.** #3533 stalled - video and primary audio at the exporter while PSI and the other PIDs - continued and the relay kept transmitting; only a per-PID liveness detector - on a subscriber found it - ([T27](https://github.com/tdrapier-wbd/moq-mpegts-paper/blob/d4c7573f9519a1c9c4882fab2021fbf82616fdcc/lab/test-27-liveness-detector.md)). -- **Same type as ingest.** `Export::stats` returns `ts::Stats`, so a - dashboard reads one schema at both edges. Additive on main. -- **No window in-tree.** The raw exporter delivers by group, with 102 - simultaneous media-time gaps over 0.3 s in 39 s on every PID, and thresholds - learned there differ from the groomed wire's for the same content (T27). A - window means something only with its monitoring point named, so the - consumer picks it. - -Implementation: - -- `rs/moq-mux/src/container/ts/export.rs`: count an access unit per PID when - its PES is written, and the quiet time on the output PCR. Fix the stale "TR 101 - 290 flags a gap over 40 ms" comment near `PCR_INTERVAL` (V1.4.1's limit is - 100 ms), and the matching 40 ms defaults and the "inserts no null packets, - and paces PCR once per media frame" sentence in `test/ts/README.md`. -- `rs/moq-cli/src/subscribe.rs` logs a stopped row the way `publish.rs` does. -- Tests in `export_test.rs`: a healthy export advances every row; a track that - stops mid-run grows its quiet time while PSI and the other PIDs continue, the - #3533 shape. diff --git a/quest/m2/ts-health-stats.md b/quest/m2/ts-health-stats.md index 28e8fc6b9d..17a521065c 100644 --- a/quest/m2/ts-health-stats.md +++ b/quest/m2/ts-health-stats.md @@ -43,4 +43,3 @@ Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838): - [Media stats schema](/quest/m1/stats/schema.md) - the snapshot the counters flatten into - [Rust reporters](/quest/m1/stats/rust.md) - `moq import --stats` and the stats interval - [TS import health](/quest/m2/ts-import-health.md) - the ingest counters -- [TS export liveness](/quest/m2/ts-export-liveness.md) - the egress rows diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 2d1d8f67c5..96d3ba18bc 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -322,6 +322,10 @@ impl Subscribe { // hand-rolled poll instead of `ts.next()`. let mut delivery = Delivery::new(self.args.max_age); let linger = self.args.linger; + // Reports a track that stops reaching the output while the rest keeps flowing, + // the way `publish` reports one that stops arriving. + let mut log = moq_mux::container::ts::stats::Log::default(); + let mut sampled = tokio::time::Instant::now(); loop { let end = loop { let mut waited = false; @@ -341,6 +345,11 @@ impl Subscribe { }; delivery.update(&frame, ts.discontinuity()); delivery.deliver(&frame, waited, &mut stdout).await?; + + if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL { + sampled = tokio::time::Instant::now(); + log.sample(ts.stats()); + } }; // Any end waits out the linger, and on expiry the last one is the result: a diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index e493c9a3af..99da67de8f 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -58,7 +58,7 @@ const PSI_INTERVAL: Duration = Duration::from_millis(500); /// table asking for faster repetition than this still gets it. const SI_REVISION_INTERVAL: Duration = Duration::from_secs(1); /// Emit a PCR on every crossing of this media-time grid ([`Export::emit`]). -/// TR 101 290 flags a gap over 40 ms; broadcast muxes emit every 25-40 ms. +/// TR 101 290 V1.4.1 flags a gap over 100 ms; broadcast muxes emit every 25-40 ms. pub(super) const PCR_INTERVAL: Duration = Duration::from_millis(25); /// How many missed PCR slots to backfill at most: one second's worth. Frames /// coarser than the grid cross several slots at a time and every one is filled so @@ -158,8 +158,9 @@ pub struct Export { /// Offsets into [`pending`](Self::pending) where a keyframe's packets begin, so /// the output frame carrying one keeps the flag. keyframes: Vec, - /// Output frames ready to hand out, one per grid slot the last span covered. - queue: VecDeque, + /// Output frames ready to hand out, one per grid slot the last span covered, each with + /// what it tells [`Self::stats`] once returned. + queue: VecDeque<(Frame, Tally)>, /// Continuity counter of the last packet emitted on the PCR PID. A clock packet /// carries no payload, so it repeats whatever preceded it on the wire rather /// than advancing the counter ([`Export::pcr_at`]). @@ -197,6 +198,24 @@ pub struct Export { /// up before it ever configures video. `None` until the tables are built, and for /// programs with no video track (nothing to align to). video_start: Option, + /// Offsets into [`pending`](Self::pending) where each access unit's packets begin, with + /// its PID, so the output frame carrying it counts it. + units: Vec<(usize, u16)>, + /// Clock packets written into the output frame being laid out ([`Self::push`]). + tally: Tally, + /// Each elementary stream's access units and silence on the PCR returned ([`Self::stats`]). + liveness: super::import::Liveness, +} + +/// What a queued output frame tells [`Export::stats`], applied only once the frame is +/// returned: a rewind drops the queue unwritten. +#[derive(Default)] +struct Tally { + /// The frame's clock packets in order: the PCR in 27 MHz ticks, and whether it flags a + /// new time base. + pcrs: Vec<(u64, bool)>, + /// The PID of each access unit whose packets begin in the frame, after its clock packets. + units: Vec, } struct Pending { @@ -408,6 +427,23 @@ enum Kind { }, } +impl Kind { + /// The track suffix [`Import`](super::Import) gives a PID carrying this kind, so a stream's + /// row is named alike at both edges. + fn suffix(&self) -> &'static str { + match self { + Kind::Video(StreamType::H265) => ".hev1", + Kind::Video(_) => ".avc3", + Kind::Aac { .. } => ".aac", + Kind::Opus { .. } => ".opus", + Kind::Mp2 { .. } => ".mp2", + Kind::Ac3 => ".ac3", + Kind::Eac3 => ".eac3", + Kind::Verbatim { .. } => ".ts", + } + } +} + /// The null stuffing owed to the multiplex rate ([`Export::stuff`]). #[derive(Default)] struct Stuffing { @@ -698,6 +734,9 @@ impl Export { mux_rate: None, mux_rate_override: None, stuffing: Stuffing::default(), + units: Vec::new(), + tally: Tally::default(), + liveness: Default::default(), }) } @@ -819,8 +858,7 @@ impl Export { // own values imply and lets the caller's pacer release each at the instant // it asserts. See [`Self::advance`]. loop { - if let Some(out) = self.queue.pop_front() { - self.emitted_epoch = self.epoch; + if let Some(out) = self.pop() { return Poll::Ready(Ok(Some(out))); } let Some(name) = self.pick_next_track(waiter) else { @@ -877,8 +915,7 @@ impl Export { let drained = !self.tracks.is_empty() && self.tracks.values().all(|t| t.finished); if drained { self.emit(None)?; - if let Some(out) = self.queue.pop_front() { - self.emitted_epoch = self.epoch; + if let Some(out) = self.pop() { return Poll::Ready(Ok(Some(out))); } // SI emission rides media frames, so a snapshot that arrived behind the @@ -899,6 +936,22 @@ impl Export { Poll::Pending } + /// Return the next queued frame, committing what it carries to the output boundary. + fn pop(&mut self) -> Option { + let (out, tally) = self.queue.pop_front()?; + self.emitted_epoch = self.epoch; + for (pcr, discontinuity) in tally.pcrs { + if discontinuity { + self.liveness.discontinuity(); + } + self.liveness.written_pcr(pcr); + } + for pid in tally.units { + self.liveness.delivered(pid, 1); + } + Some(out) + } + /// The trailing SI frame for end of stream: every entry's current sections, /// re-emitted once. Duplicates of already-emitted sections are harmless /// repetitions; `None` when there are no tables or no program was ever built. @@ -1257,6 +1310,30 @@ impl Export { self.emitted_epoch } + /// Snapshot the access units each elementary stream has written, and how long each has + /// been quiet on the PCR the output carries. + /// + /// A track stalled upstream stops advancing its row while the PSI and the other PIDs + /// keep flowing, which nothing graded on the output bytes alone can see. Empty until the + /// program tables are built. Cheap enough to poll per frame. + pub fn stats(&self) -> super::Stats { + let mut stats = super::Stats::default(); + if self.psi.is_none() { + return stats; + } + for track in self.tracks.values() { + let (units, quiet) = self.liveness.stream(track.pid); + let row = super::StreamStats { + track: track.kind.suffix(), + units, + quiet, + ..Default::default() + }; + stats.streams.insert(track.pid, row); + } + stats + } + /// Carry on with the broadcast that replaced the one this export was reading. /// /// Call it once [`Source::returned`](crate::Source::returned) has resolved, after @@ -1301,6 +1378,8 @@ impl Export { } self.pending.clear(); self.keyframes.clear(); + self.units.clear(); + self.tally = Tally::default(); self.queue.clear(); self.watermark = None; // The stall carries across with its budget, and so does a held frame's @@ -1499,6 +1578,9 @@ impl Export { es_info, }; + for track in &tracks { + self.liveness.register(track.pid); + } self.psi = Some(Psi { pat, pmt, @@ -1664,6 +1746,8 @@ impl Export { } } + // A malformed section writes nothing, so the unit counts only if bytes went in. + let written = out.len(); match es_payload { // Section-framed verbatim (SCTE-35, ...) rides in private sections, not PES; // carry the bytes verbatim. @@ -1689,6 +1773,9 @@ impl Export { self.write_pes(&mut out, &unit, &es_payload)?; } } + if out.len() > written { + self.units.push((self.pending.len() + written, pid)); + } if keyframe { self.keyframes.push(self.pending.len()); } @@ -1752,6 +1839,7 @@ impl Export { self.span_counters = None; let bytes = std::mem::take(&mut self.pending); let keyframes = std::mem::take(&mut self.keyframes); + let units = std::mem::take(&mut self.units); let Some(to) = self.low.take() else { return Ok(()) }; let packets = bytes.len() / TsPacket::SIZE; @@ -1814,7 +1902,7 @@ impl Export { }; self.send(&mut payload, &bytes[cut * TsPacket::SIZE..next * TsPacket::SIZE]); self.stuff(index, &mut payload); - self.push(at, payload, &keyframes, cut, next); + self.push(at, payload, &keyframes, &units, cut, next); cut = next; at = boundary; let before = counter_before(&bytes, cut * TsPacket::SIZE, pcr_pid, self.pcr_cc); @@ -1824,7 +1912,7 @@ impl Export { } self.send(&mut payload, &bytes[cut * TsPacket::SIZE..]); - self.push(at, payload, &keyframes, cut, packets); + self.push(at, payload, &keyframes, &units, cut, packets); if let Some(cc) = counter_before(&bytes, bytes.len(), pcr_pid, None) { self.pcr_cc = Some(cc); } @@ -1891,19 +1979,34 @@ impl Export { } /// Queue one output frame, unless it would be empty. `from`..`to` are the packet - /// indices it carries, which decide whether a keyframe begins in it. - fn push(&mut self, timestamp: Timestamp, payload: Vec, keyframes: &[usize], from: usize, to: usize) { + /// indices it carries, which decide whether a keyframe, or which access units, begin in it. + fn push( + &mut self, + timestamp: Timestamp, + payload: Vec, + keyframes: &[usize], + units: &[(usize, u16)], + from: usize, + to: usize, + ) { if payload.is_empty() { return; } let (from, to) = (from * TsPacket::SIZE, to * TsPacket::SIZE); let keyframe = keyframes.iter().any(|&at| at >= from && at < to); - self.queue.push_back(Frame { + let mut tally = std::mem::take(&mut self.tally); + tally.units = units + .iter() + .filter(|&&(at, _)| at >= from && at < to) + .map(|&(_, pid)| pid) + .collect(); + let frame = Frame { timestamp, duration: None, payload: Bytes::from(payload), keyframe, - }); + }; + self.queue.push_back((frame, tally)); } /// The clock packet for grid slot `index`, and record that the slot is served. @@ -1938,9 +2041,11 @@ impl Export { }; self.last_pcr = Some(index); let mut packet = pcr_packet(pcr_pid, ticks, cc)?; - if std::mem::take(&mut self.pcr_discontinuity) { + let discontinuity = std::mem::take(&mut self.pcr_discontinuity); + if discontinuity { packet[5] |= 0x80; } + self.tally.pcrs.push(((ticks & TS_TIMESTAMP_MASK) * 300, discontinuity)); Ok(packet) } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index f51b49f195..36fa3b2606 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -17,7 +17,7 @@ use mpeg2ts::ts::{ReadTsPacket, TsPacketReader, TsPayload}; use crate::catalog::hang::Container as HangContainer; use crate::container::ts::export::PCR_INTERVAL; -use crate::container::ts::{Export, catalog as tscat}; +use crate::container::ts::{Export, Stats, catalog as tscat}; use crate::container::{Frame, Kind, Producer}; use moq_net::Timestamp; @@ -5718,3 +5718,240 @@ async fn resume_after_a_finish() { async fn resume_after_a_drop() { resume_after(false).await; } + +/// The stats count only output that was returned. A frame the muxer refuses fails the export +/// with the span before it queued but never returned, and the resume discards it. +#[tokio::test(start_paused = true)] +async fn export_stats_skip_output_a_failure_discards() { + let origin = crate::source::produce_origin(); + let source = crate::Source::new(origin.consume(), "live"); + let publish = || { + let mut broadcast = origin.publish("live", Default::default()).unwrap(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + (broadcast, catalog) + }; + let units = |stats: Stats| stats.streams.values().map(|row| row.units).sum::() as usize; + + let (mut broadcast, mut catalog) = publish(); + let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let ended = source.broadcast().await.unwrap(); + let mut export = Export::new(source.clone()) + .await + .unwrap() + .with_max_age(RECORDING_MAX_AGE); + for ms in (0..200).step_by(20) { + write_aac(&mut track, ms); + } + let mut frames = drain_frames(&mut export).await; + // One byte past what an ADTS header can frame. + track + .write(Frame { + timestamp: Timestamp::from_millis(200).unwrap(), + duration: None, + payload: Bytes::from(vec![0; 8185]), + keyframe: true, + }) + .unwrap(); + let (rest, end) = drain_to_end(&mut export).await; + frames.extend(rest); + assert!(end.is_err(), "an unframeable AAC frame fails the export"); + assert!(pes_count(&frames) < 10, "the span before the failure stayed queued"); + assert_eq!(units(export.stats()), pes_count(&frames)); + + drop((broadcast, catalog, track)); + let (mut broadcast, mut catalog) = publish(); + let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + source.returned(&ended).await.unwrap(); + export.resume().await.unwrap(); + for ms in (0..200).step_by(20) { + write_aac(&mut track, ms); + } + frames.extend(drain_frames(&mut export).await); + track.finish().unwrap(); + catalog.finish().unwrap(); + let (rest, end) = drain_to_end(&mut export).await; + frames.extend(rest); + end.unwrap(); + assert_eq!( + units(export.stats()), + pes_count(&frames), + "the discarded span never counts" + ); +} + +/// Export 25 fps video and two AAC tracks for [`TICKS`] video frames, the video and the first +/// audio track stopping after `stop` while the second carries on. Returns the stats sampled +/// after `sample` and at the end, and the frames rendered in between. +/// +/// Drained after every write, like [`export_twice`], so the export reads every frame and a row +/// stops only because its track did. +async fn export_liveness(sample: u64, stop: u64) -> (Stats, Stats, Vec) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + + let track = broadcast + .create_track("video.avc1", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let mut cfg = VideoConfig::new(H264 { + profile: 0x42, + constraints: 0xc0, + level: 0x1f, + inline: false, + }); + cfg.container = Container::Legacy; + cfg.description = + Some(crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap()); + catalog + .modify() + .unwrap() + .video + .renditions + .insert(track.name().to_string(), cfg); + let mut video = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data)); + let mut audio = [ + aac_rendition(&mut broadcast, &mut catalog, "primary.aac"), + aac_rendition(&mut broadcast, &mut catalog, "secondary.aac"), + ]; + + let mut export = Export::new(crate::source::announced(&consumer)).await.unwrap(); + assert!(export.stats().is_empty(), "no row before the program tables"); + let (mut sampled, mut frames) = (None, Vec::new()); + let mut audio_index = 0; + for tick in 0..TICKS { + if tick < stop { + let keyframe = tick % GOP == 0; + let slice = if keyframe { + vec![0x65u8; 3_000] + } else { + vec![0x41u8; 400] + }; + video + .write(Frame { + timestamp: Timestamp::from_micros(tick * VIDEO_US).unwrap(), + duration: None, + payload: length_prefixed(&[&slice]), + keyframe, + }) + .unwrap(); + } + while audio_index * AUDIO_US < (tick + 1) * VIDEO_US { + let running = if tick < stop { &mut audio[..] } else { &mut audio[1..] }; + for track in running { + track + .write(Frame { + timestamp: Timestamp::from_micros(audio_index * AUDIO_US).unwrap(), + duration: None, + payload: Bytes::from_iter((0..180u16).map(|i| (i ^ audio_index as u16) as u8)), + keyframe: audio_index % AUDIO_GROUP == 0, + }) + .unwrap(); + } + audio_index += 1; + } + + let out = drain_frames(&mut export).await; + if sampled.is_some() { + frames.extend(out); + } + if tick + 1 == sample { + sampled = Some(export.stats()); + } + } + (sampled.expect("sampled mid-run"), export.stats(), frames) +} + +/// A healthy export advances every elementary stream's row, each quiet for no longer than +/// its own frame spacing plus the mux buffer, and leaves the import-only counters at zero. +#[tokio::test(start_paused = true)] +async fn export_stats_advance_every_stream() { + let (mid, end, _) = export_liveness(TICKS / 2, TICKS).await; + + let tracks: Vec<&str> = end.streams.values().map(|row| row.track).collect(); + assert_eq!(tracks, [".aac", ".aac", ".avc3"], "one row per elementary stream"); + for (pid, row) in &end.streams { + let before = &mid.streams[pid]; + assert!(before.units > 0, "PID {pid:#x} wrote nothing by mid-run"); + assert!( + row.units > before.units, + "PID {pid:#x} stopped advancing: {before:?} -> {row:?}" + ); + let quiet = row.quiet.expect("the output carries a PCR"); + assert!(quiet < Duration::from_millis(200), "PID {pid:#x} quiet for {quiet:?}"); + assert_eq!((row.resyncs, row.discarded, row.unconfirmed), (0, 0, 0)); + } +} + +/// The #3533 shape: video and the primary audio stop reaching the exporter while the +/// secondary audio carries on. Their rows freeze and their quiet time grows on the output's +/// own PCR, which keeps running on the surviving track, as do the PSI and that track's row. +#[tokio::test(start_paused = true)] +async fn export_stats_catch_a_track_that_stops() { + let stop = TICKS / 2; + // Sampled once the mux buffer has written out what the stopped tracks sent last. + let (sampled, end, after) = export_liveness(stop + 5, stop).await; + let stalled = Duration::from_micros((TICKS - stop) * VIDEO_US); + + let advanced = |pid: &u16| end.streams[pid].units > sampled.streams[pid].units; + let (running, stopped): (Vec, Vec) = end.streams.keys().copied().partition(advanced); + let [running] = running[..] else { + panic!("expected one stream to keep advancing, got {running:?}: {end:?}"); + }; + assert_eq!(end.streams[&running].track, ".aac"); + assert_eq!(stopped.len(), 2, "video and the primary audio stopped"); + + let quiet = |pid: u16| end.streams[&pid].quiet.expect("the output carries a PCR"); + assert!(quiet(running) < Duration::from_millis(200), "{:?}", quiet(running)); + for pid in stopped { + // The mux buffer holds the last span back, so the output's clock trails the media. + assert!( + quiet(pid) > stalled - Duration::from_millis(200), + "PID {pid:#x} quiet for only {:?} of a {stalled:?} stall", + quiet(pid) + ); + } + assert!( + count_pid(&after, 0x0000) >= 3, + "the PAT kept repeating through the stall" + ); + assert!(count_pid(&after, running) > 0, "the surviving track kept flowing"); +} + +/// A stopped track's silence counts the whole of a media gap on the surviving track, not just +/// the one second of clock the exporter backfills: the output's PCR jumps across the rest +/// unflagged, and that jump is time the stopped track was silent. +#[tokio::test(start_paused = true)] +async fn export_stats_count_a_gap_longer_than_the_backfill() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut stopped = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let mut running = aac_rendition(&mut broadcast, &mut catalog, "b.aac"); + + let mut export = Export::new(crate::source::announced(&consumer)).await.unwrap(); + for ms in (0..2_000).step_by(20) { + write_aac(&mut stopped, ms); + write_aac(&mut running, ms); + drain_frames(&mut export).await; + } + // Five seconds with nothing on either track, then only the second resumes. + for ms in (7_000..8_000).step_by(20) { + write_aac(&mut running, ms); + drain_frames(&mut export).await; + } + + let stats = export.stats(); + let mut rows: Vec<_> = stats.streams.values().collect(); + rows.sort_by_key(|row| row.units); + let [stopped, running] = rows[..] else { + panic!("expected two rows: {stats:?}"); + }; + assert!(stopped.units < running.units, "{stats:?}"); + let quiet = stopped.quiet.expect("the output carries a PCR"); + assert!( + quiet > Duration::from_millis(5_800), + "stopped at 2 s, quiet for only {quiet:?} at 8 s" + ); + let quiet = running.quiet.expect("the output carries a PCR"); + assert!(quiet < Duration::from_millis(200), "{quiet:?}"); +} diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index fe39363d79..839f042597 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -992,11 +992,16 @@ fn list_programs(programs: &[u16]) -> String { /// importer, so what an operator alarms on is the rate: a feed that resyncs once an hour is /// healthy, one that resyncs every second is losing audio, and one whose access units stop /// advancing while the mux keeps flowing has lost that stream. +/// +/// [`Export::stats`](super::Export::stats) returns the same rows for the streams it writes, +/// so one schema reads both edges. Only `units` and `quiet` move there: the exporter builds +/// every frame header itself, so it has no frame sync to lose. #[derive(Clone, Debug, Default, PartialEq, Eq)] #[non_exhaustive] pub struct Stats { - /// One row per elementary stream PID the importer carries, ordered by PID. A PID the - /// importer drops (an undecoded stream without the `mpegts` catalog section) has none. + /// One row per elementary stream PID the importer carries or the exporter writes, ordered + /// by PID. A PID the importer drops (an undecoded stream without the `mpegts` catalog + /// section) has none. pub streams: BTreeMap, /// PAT and PMT sections dropped because their CRC-32 did not match. Each one left the /// table already in force standing, or, before any, delayed the program's start to the @@ -1017,13 +1022,16 @@ impl Stats { #[non_exhaustive] pub struct StreamStats { /// The current or most recent MoQ track suffix for this PID (`.avc3`, `.mp2`, `.ts`, ...), - /// or empty for MPEG-1/2 video, which is read for its clock and not published. + /// or empty for MPEG-1/2 video, which is read for its clock and not published. At export, + /// the suffix an import of the output would give the PID. pub track: &'static str, /// Access units the stream delivered: frames published for decoded media, PES payloads or - /// sections carried verbatim, and PES read on MPEG-1/2 video. + /// sections carried verbatim, and PES read on MPEG-1/2 video. At export, the PES or + /// sections written for each frame. pub units: u64, /// Transport time since the stream last delivered an access unit, or since the PMT - /// declared it, measured on the program clock (PCR). `None` until the first PCR arrives. + /// declared it, measured on the program clock (PCR): the one read at import, the one + /// written at export. `None` until the first PCR. /// /// No threshold applies: a sparse stream such as SCTE-35 is legitimately quiet for /// seconds, so how long is too long is for whoever alarms to decide. @@ -1487,9 +1495,9 @@ impl Continuity { /// Measured on the program clock, the PCR summed interval by interval, so it runs straight /// through the 33-bit wrap, a signalled time-base reset and a corrupt PCR instead of jumping /// with them. Not the media clock: that follows the video PTS, and stops with the very stream -/// this catches. +/// this catches. [`Export`](super::Export) runs the same meter on the PCR it writes. #[derive(Default)] -struct Liveness { +pub(super) struct Liveness { /// The last PCR in 27 MHz ticks, forgotten at a reset so no interval spans one. pcr: Option, /// PCR ticks elapsed since the first PCR, if one has arrived. @@ -1501,32 +1509,44 @@ struct Liveness { impl Liveness { /// A PCR arrived on the program's clock PID. - fn pcr(&mut self, pcr: u64) { + /// + /// Stepped one interval at a time, so a corrupt PCR, or the clock stepping back without a + /// flag, costs one interval rather than inventing hours of silence on every PID. The bound + /// is the mux-rate meter's. + pub(super) fn pcr(&mut self, pcr: u64) { + self.step(pcr, super::mux_rate::MAX_INTERVAL); + } + + /// A PCR the caller wrote itself. Its clock breaks only where the caller flags a + /// discontinuity, so every forward step counts, including the jump across a media gap + /// longer than the backfill covers. + pub(super) fn written_pcr(&mut self, pcr: u64) { + self.step(pcr, super::mux_rate::PCR_WRAP); + } + + fn step(&mut self, pcr: u64, max: u64) { let elapsed = self.elapsed.get_or_insert(0); if let Some(last) = self.pcr.replace(pcr) { let step = (pcr + super::mux_rate::PCR_WRAP - last) % super::mux_rate::PCR_WRAP; - // Stepped one interval at a time, so a corrupt PCR, or the clock stepping back - // without a flag, costs one interval rather than inventing hours of silence on - // every PID. The bound is the mux-rate meter's. - if step <= super::mux_rate::MAX_INTERVAL { + if step <= max { *elapsed += step; } } } /// The clock restarted or changed PID: the next interval measures nothing. - fn discontinuity(&mut self) { + pub(super) fn discontinuity(&mut self) { self.pcr = None; } /// The PMT declared `pid`. Its silence counts from here until it delivers. - fn register(&mut self, pid: u16) { + pub(super) fn register(&mut self, pid: u16) { let now = self.elapsed.unwrap_or(0); self.streams.entry(pid).or_insert((0, now)); } /// `pid` delivered `units` access units just now. - fn delivered(&mut self, pid: u16, units: u64) { + pub(super) fn delivered(&mut self, pid: u16, units: u64) { if units == 0 { return; } @@ -1538,7 +1558,7 @@ impl Liveness { } /// `pid`'s access units and how long it has been quiet. See [`StreamStats`]. - fn stream(&self, pid: u16) -> (u64, Option) { + pub(super) fn stream(&self, pid: u16) -> (u64, Option) { let Some(&(units, last)) = self.streams.get(&pid) else { return (0, None); }; diff --git a/rs/moq-mux/src/container/ts/stats.rs b/rs/moq-mux/src/container/ts/stats.rs index 0172a22f03..519c62f180 100644 --- a/rs/moq-mux/src/container/ts/stats.rs +++ b/rs/moq-mux/src/container/ts/stats.rs @@ -1,11 +1,12 @@ -//! Reporting on the counters [`Import::stats`](super::Import::stats) returns. +//! Reporting on the counters [`Import::stats`](super::Import::stats) and +//! [`Export::stats`](super::Export::stats) return. use std::collections::BTreeSet; use super::{Stats, StreamStats}; -/// Logs what moved in an importer's [`Stats`] from one sample to the next, so every front -/// door that runs a TS importer reports the same lines. +/// Logs what moved in an importer's or exporter's [`Stats`] from one sample to the next, so +/// every front door that runs one reports the same lines. /// /// Feed it a snapshot every [`INTERVAL`](Self::INTERVAL) of wall time. The caller owns the /// timer, since what counts as "now" differs between a CLI, a server, and a test. diff --git a/test/ts/README.md b/test/ts/README.md index 4fb5950b9d..336ceab858 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -7,7 +7,9 @@ runs [TSDuck](https://tsduck.io) plus a custom analyzer over it. This is a diagnostic gate, not just a pass/fail: the exporter ([`rs/moq-mux/src/container/ts/export.rs`](../../rs/moq-mux/src/container/ts/export.rs)) -is VBR, inserts no null packets, and paces PCR once per media frame, so several +pads with null packets to the multiplex rate the source recorded (or `--mux-rate`), +leaves a source without one unpadded, never delays media to fit the rate, and +puts a PCR on its own packet every 25 ms of media time, so several broadcast-shape checks are expected to flag. The report quantifies exactly where and by how much. @@ -80,7 +82,7 @@ Severities: **hard** checks fail the run by default; **shape** checks report as | `pcr-presence` | hard | a PCR PID is declared and carries PCR | | `pcr-monotonic` | hard | PCR strictly increases (one 33-bit wrap tolerated), except into a PCR that signals `discontinuity_indicator` | | `duration-fidelity` | hard | exported PCR span tracks the source's duration (round-trip only) | -| `pcr-repetition` | shape | consecutive PCRs within the limit (default 40 ms) | +| `pcr-repetition` | shape | consecutive PCRs within the limit (default 100 ms, TR 101 290 V1.4.1) | | `pcr-jitter` | shape | per-interval PCR jitter vs the nominal bitrate (pcrverify model) | | `null-ratio` | shape | null/stuffing fraction (flags only a pathological excess) | | `service-descriptors` | shape | an SDT naming the service is present | @@ -146,7 +148,7 @@ grades only how evenly the bytes are laid over the PCRs. | `sync` | hard | no invalid sync bytes / transport-error packets | | `continuity` | hard | no discontinuities, and a payload-less packet must not advance the counter (ISO 13818-1 2.4.3.3) | | `pcr-single-pid` | hard | every PCR rides one PID | -| `pcr-value-interval` | hard | no interval above `--repetition-ms` (default 40, TR 101 290), within one time base | +| `pcr-value-interval` | hard | no interval above `--repetition-ms` (default 100, TR 101 290 V1.4.1), within one time base | | `pcr-release-timing` | hard | no more than `--release-pct-max` of intervals arrive further than `--release-ms` from the interval their own values assert, and accumulated drift stays within `--drift-ms`, being the standing lag the sender is allowed to hold; a sample below `--live-min-pcr` PCRs or `--live-cover-pct` of the window is a failure, not a pass (`--live` only) | | `pcr-position` | shape | share of PCR packets within `--adjacent-packets` of the previous one | | `pcr-schedule` | shape | share of PCR intervals, on the busiest PCR PID, whose bytes are within `--schedule-tolerance-pct` (default 1) or one packet of what `--mux-rate` implies (estimated from the capture if not given); hard, at that share, when `--schedule-pct-min` is given | diff --git a/test/ts/compliance.py b/test/ts/compliance.py index 66aae98fe7..2774948c6f 100755 --- a/test/ts/compliance.py +++ b/test/ts/compliance.py @@ -72,7 +72,7 @@ class Check: class Thresholds: """IRD limits and T-STD model parameters, all overridable from the CLI.""" - pcr_repetition_ms: float = 40.0 + pcr_repetition_ms: float = 100.0 pcr_jitter_us: float = 500.0 null_ratio_max: float = 0.90 bitrate_cov_max: float = 0.10 diff --git a/test/ts/pcr-timing.py b/test/ts/pcr-timing.py index b318fe3904..9c08abe889 100755 --- a/test/ts/pcr-timing.py +++ b/test/ts/pcr-timing.py @@ -605,7 +605,7 @@ def main(): ap.add_argument("--live", action="store_true", help="read stdin and stamp arrivals, grading release timing") ap.add_argument("--seconds", type=float, default=45.0, help="live capture window (default 45)") ap.add_argument( - "--repetition-ms", type=float, default=40.0, help="max PCR value interval, TR 101 290 (default 40)" + "--repetition-ms", type=float, default=100.0, help="max PCR value interval, TR 101 290 V1.4.1 (default 100)" ) ap.add_argument( "--release-ms", diff --git a/test/ts/run.sh b/test/ts/run.sh index 0858b89dad..d30320d1e4 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -7,7 +7,8 @@ # TSDuck + custom analyzer in compliance.py against the capture. The point is to # tell whether what the subscriber emits is something an Integrated # Receiver/Decoder would accept, and to quantify where it diverges (the exporter -# is VBR, emits no null packets, and paces PCR per frame). +# pads to the recorded multiplex rate but never delays media to fit it, and puts a +# PCR every 25 ms of media time). # # Modes: # ./run.sh # generate a clip, round-trip it, analyze