diff --git a/doc/bin/cli.md b/doc/bin/cli.md index ae5e352632..02d36e054a 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -463,7 +463,8 @@ The `stats` and `echo` sinks print reports as JSON lines. `export stats` follows one broadcast's stats track until the broadcast ends; `export echo` prints `{"broadcast", "feedback"}` for every `.echo` broadcast under `--broadcast` as they come and go, reading one that ends for 5 s more for its -final report. Either sink stops reading a report that grows past 1 MiB. +final report. Either sink stops reading a report that grows past 8 MiB, twice +the largest a TS import publishes. ```bash moq ... --broadcast event import --stats ts < feed.ts @@ -478,6 +479,13 @@ A report holds 1,024 rendition IDs at most: past that, each new ID replaces one that has ended. [Stats](/concept/stats#media-reports) has the schema. +A TS import adds an `mpegts` section: each elementary stream's access units +and quiet time, and the TR 101 290 counters it logs. A TS export adds its +streams' rows and a `release` section: frames dropped late, the margin frames +arrive with, and its clock's offset and drift from the source. A passthrough +reports `release` alone at export and neither at import. See +[MPEG-TS sections](/concept/stats#mpeg-ts-sections). + ## Debugging `RUST_LOG=debug` prints the negotiated version and every subscription. diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 55006d4990..8f1227c393 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -39,6 +39,64 @@ it reads and records its newest arrival. See the [hang draft](/draft/moq-hang#media-stats) for track framing, field units, and the feedback convention. +### MPEG-TS sections + +A TS import or export adds sections of its own, which +`moq_mux::container::ts::stats::Report` reads. As above, counters are +cumulative and omitted at zero, and a gauge is omitted until it is measured. + +`mpegts` holds `streams`, one row per elementary stream keyed by its PID in +decimal, and the stream-wide counters. A row has `track` (the track suffix), +`class` (`audio`, `video`, or `data`; an unknown class reads as `data`), +`units` (access units delivered), and `quiet`, a gauge: milliseconds of +program clock since its last access unit. Its losses are `resyncs`, bytes +`discarded` scanning for audio sync, `unconfirmed` frames, `damaged` units, +`unplaced` units the program clock could not place, the times the stream was +`refused`, and its shares of `continuityCountError`, `transportError`, and +`ptsError`. + +The stream-wide counters count ETSI TR 101 290 checks at that standard's +fixed limits: + +| Field | Check | +| --- | --- | +| `tsSyncLoss` | 1.1 `TS_sync_loss` | +| `syncByteError` | 1.2 `Sync_byte_error` | +| `patError` | 1.3 `PAT_error_2` | +| `continuityCountError` | 1.4 `Continuity_count_error` | +| `pmtError` | 1.5 `PMT_error_2` | +| `transportError` | 2.1 `Transport_error` | +| `crcError` | 2.2 `CRC_error`, on the PAT and PMT only | +| `pcrRepetitionError` | 2.3a `PCR_repetition_error` | +| `pcrDiscontinuityIndicatorError` | 2.3b `PCR_discontinuity_indicator_error` | +| `ptsError` | 2.5 `PTS_error` | + +An import grades the stream as it arrived. An export builds every header +itself and runs no checks, so its counters stay zero and only its rows' +`units` and `quiet` move. Neither grades the wire a receiver sees downstream. +The PCR checks are graded on PCR values, not arrival: they speak for the +encoder's insertion, since the network re-times arrival, and a forward step +over 100 ms counts as both. 1.6 `PID_error` is covered, more strictly, by each +row's `units` and `quiet`: a dead video path behind a live mux still sends +PCR-only packets on its PID, which a packet count reads as live. +`PCR_accuracy_error` is not measured. No verdict is computed: a sparse stream +such as SCTE-35 is quiet for seconds by design, so an alarm reads counter +rates and `quiet` against its own thresholds. + +An export also reports `release`, how it keeps up with the source: what it +drops, trims and never receives, and how its output clock tracks the +source's. `export ts --passthrough` reports it without `mpegts`. + +| Field | Meaning | +| --- | --- | +| `dropped` | Frames lost from the output after their stream began: missed their release deadline, or a video track dropped them waiting for its next keyframe. | +| `trimmed` | Frames dropped before their stream's first went out, so the output starts on a whole group led by a keyframe. Nothing is missing for these. | +| `gaps` | Times the source skipped media it will never deliver: a group lost or expired upstream, or a resubscription. A declared break is not one. | +| `margin` | Gauge, milliseconds rounded away from zero: the least time any frame arrived before its deadline over the last second. Negative when one missed, however narrowly. | +| `offset` | Gauge, milliseconds rounded away from zero: how far the output clock sits from the source's plus the delay. Positive holds frames longer than the delay. The clock pulls it shut over ten minutes. | +| `drift` | Gauge, parts per million: the source clock's rate against ours, positive running fast. | +| `outOfTolerance` | Measurements that found the source clock past the 30 ppm ISO/IEC 13818-1 allows, which the output cannot follow. | + ## Broadcasts Each node publishes under a prefix, `.stats` by default, which diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index f543017f97..28acd4f2ea 100644 --- a/quest/m1/ts-reliability/README.md +++ b/quest/m1/ts-reliability/README.md @@ -105,6 +105,5 @@ remain with their current quests. ## Related - [TS over RTP legs](/quest/m2/ts-rtp-egress.md) - TS identity alone is not ST 2022-7, which also needs matching RTP headers -- [TS health counters](/quest/m2/ts-health-stats.md) - ingress and egress evidence need distinct monitoring points - [MSFTS mapping](/quest/m3/msfts-convergence.md) - document the exact equivalence and differences between the carriage modes diff --git a/quest/m2/README.md b/quest/m2/README.md index 502885431f..c9edb8210c 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -144,7 +144,6 @@ ranked just below deterministic TS legs, which it requires. - [Probes resume on the peer](/quest/m2/quic-outage-probe.md) - after an outage, the next probe follows the peer's first packet instead of the backed-off timer - [Keep-alive from the idle timeout](/quest/m2/quic-keep-alive.md) - the keep-alive interval defaults to a fraction of the negotiated idle timeout; an explicit `quic.keep_alive` overrides - [Socket close](/quest/m2/quic-socket-close.md) - moq-tokio's QUIC endpoint releases its socket on close, so it drops its wrapper -- [TS health stats](/quest/m2/ts-health-stats.md) - the TS counters ride the stats plumbing beside the media counters - [Windows.Graphics.Capture](/quest/m2/capture-wgc.md) - the WGC display and window backend verified on real Windows hardware - [Windows capture parity](/quest/m2/capture-windows.md) - system audio and a settled app-capture policy - [Linux capture parity](/quest/m2/capture-linux.md) - Wayland window/system-audio capture with explicit display-selection and app-capture limits diff --git a/quest/m2/stats/health.md b/quest/m2/stats/health.md index 99511272a4..155ed2ab86 100644 --- a/quest/m2/stats/health.md +++ b/quest/m2/stats/health.md @@ -42,7 +42,7 @@ Guidance, to be settled while building: starts over rather than going negative. - Container sections such as `mpegts` are excluded from the verdict; their counters are read by their own consumers - ([TS health counters](/quest/m2/ts-health-stats.md)). + ([MPEG-TS sections](/doc/concept/stats.md#mpeg-ts-sections)). - Thresholds are a documented default the caller can override, not policy baked into the types. Start from a few measured cases rather than guesses. - A verdict carries its evidence: which inputs drove it and who observed them. @@ -65,4 +65,3 @@ Public API: new health types in `hang` and `@moq/hang`. Wire: none. - [QoS](/quest/m2/qos/README.md) - the relay's delivery counters, which a combined downstream verdict also reads - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - a new epoch is one way a counter reset shows -- [TS health counters](/quest/m2/ts-health-stats.md) - container counters kept out of the verdict diff --git a/quest/m2/stats/rust.md b/quest/m2/stats/rust.md index 985ade5a44..03b32a5008 100644 --- a/quest/m2/stats/rust.md +++ b/quest/m2/stats/rust.md @@ -41,5 +41,4 @@ ## Related -- [TS health stats](/quest/m2/ts-health-stats.md) - adds TS-specific counters to the reports - [Encrypted native CLI](/quest/m2/e2ee/cli.md) - refuses stats and feedback on an encrypted broadcast diff --git a/quest/m2/ts-health-stats.md b/quest/m2/ts-health-stats.md deleted file mode 100644 index d72d74cb68..0000000000 --- a/quest/m2/ts-health-stats.md +++ /dev/null @@ -1,59 +0,0 @@ -# [S] TS health counters ride the stats plumbing - -## Goal - -`moq import ts --stats` and `moq export ts --echo` publish the `ts::stats` -counters so a dashboard reads the TR 101 290 counters and per-PID liveness -beside the media and delivery counters, with no new transport. No health -roll-up in-tree: green, amber or red per priority is the consumer's reading of -counter rates and PID quiet times. - -## Plan - -Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838): - -The quest audit confirmed this stays in m2 and owns TS counter serialization -and integration. Rust reporters supplies only the generic per-rendition -plumbing; it does not implement this quest's `mpegts` payloads. - -- **Surface through the stats plumbing**, as the issue settled. The - [media stats schema](/doc/concept/stats.md#media-reports) settles the shape: an - import flattens `ts::stats::Snapshot` into the publisher's stats snapshot under the - catalog's `mpegts` key, the counters stay owned by moq-mux beside - `ts::Ext`, and `hang` stays TS-free. -- **The export reports through `.echo`** (decided 2026-10-12, from - [#5360](https://github.com/moq-dev/moq/issues/5360)). An export is a viewer, - so `moq export ts --echo ` folds `ts::stats::Snapshot` into its - solicited `.echo` snapshot under `mpegts`, through the `E` flatten the - [media stats schema](/doc/concept/stats.md#media-reports) gives `echo::Snapshot`. - The egress is where an operator otherwise cannot see per-PID outages: a - 57 s video hole passed every TR 101 290 P1 check in the reporter's lab, - and only per-PID access-unit counts caught it. There is no export - `--stats` flag. -- **Re-timing gets its own section beside `mpegts`** (decided 2026-10-16, - on [#5643](https://github.com/moq-dev/moq/pull/5643)). An export that - re-times onto an output clock reports what `ts::stats::Snapshot` cannot - hold: `ts::stats::Export`'s stream-wide `dropped`, `trimmed`, `gaps`, - `drift`, and `out_of_tolerance`, which its conversion to `Snapshot` - discards, plus the release margin and the output clock's offset from the - source. They describe the export's own output - clock rather than viewer playout, so they sit flat beside `transport` and - `renditions` now instead of waiting for the - [latency ledger](/quest/m2/latency-ledger.md)'s `.echo` shape. -- **Counters sum, gauges do not.** Every check is a cumulative counter, which - keeps `.z` deltas small and lets an aggregate sum; the PID's `quiet` is a gauge and - merges newest-wins. -- `ts::stats::Snapshot` gains serde, every field defaulted and unknown fields - ignored. Its rows' `track` is already an owned `String`, so they deserialize. -- Docs: `doc/concept/stats.md` documents each counter's TR 101 290 check, its - monitoring point (ingest grades the feed, egress grades our muxer, and - neither is the groomed wire), the value domain of the PCR checks, and that - the per-PID rows cover `PID_error`; `doc/bin/cli.md` the flags. -- Test: publish a TS clip with `--stats` and export it with `--echo` to a - soliciting catalog, read both back, and assert every ETSI - counter reads zero on a clean clip while every elementary stream's - access-unit count advances. - -## Required - -- [Rust reporters](/quest/m2/stats/rust.md) - `moq import --stats` and the stats interval diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 202341d706..ccde284316 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -923,6 +923,7 @@ async fn run_stdout( return subscribe.run().await; }; let mut echo = report::Echo::new(echo, subscribe.source().clone(), catalog); + let subscribe = subscribe.with_sections(echo.sections()); let result = tokio::select! { result = subscribe.run() => result, never = echo.run() => match never {}, diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 8c37632f78..66651054d1 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -606,6 +606,12 @@ async fn decode( // they have to surface as they accumulate rather than at exit. let mut log = ts::stats::Log::default(); let mut sampled = tokio::time::Instant::now(); + let sections = report.as_ref().map(crate::report::Stats::sections); + let report_mpegts = |snapshot: &ts::stats::Snapshot| { + if let Some(sections) = §ions { + sections.set(|report| report.mpegts = Some(snapshot.clone())); + } + }; // Run the read/decode loop so an error surfaces here rather than // dropping the decoder (and its tracks) with a bare Error::Dropped. @@ -626,6 +632,7 @@ async fn decode( if sampled.elapsed() >= ts::stats::Log::INTERVAL { sampled = tokio::time::Instant::now(); if let Some(stats) = decoder.stats() { + report_mpegts(&stats); log.sample(stats); } } @@ -656,6 +663,7 @@ async fn decode( // The drain at end of input can publish a frame nothing vouched for, so the // final snapshot is only complete after `finish`. if let Some(stats) = decoder.stats() { + report_mpegts(&stats); log.finish(&stats); } match outcome { diff --git a/rs/moq-cli/src/report.rs b/rs/moq-cli/src/report.rs index 9c6a350b4f..eeaac78fbb 100644 --- a/rs/moq-cli/src/report.rs +++ b/rs/moq-cli/src/report.rs @@ -7,11 +7,13 @@ use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; +use std::sync::{Arc, Mutex, PoisonError}; use std::time::Duration; use anyhow::Context; use hang::moq_net; use moq_mux::catalog::{CatalogFormat, Stream}; +use moq_mux::container::ts; use moq_tokio::connection::Monitor; use tokio::io::AsyncWriteExt; @@ -27,9 +29,11 @@ const COMPRESSED: &str = ".z"; /// Lines `export echo` queues for stdout before its readers wait for it to catch up. const LINES: usize = 64; -/// The most reconstructed state a sink holds for one report. A report is a few KiB, so this -/// only stops a publisher whose patches keep adding keys. -const MAX_REPORT: usize = 1024 * 1024; +/// The most reconstructed state a sink holds for one report. A report is a few KiB, but a TS +/// import keeps a row for each PID it has carried, and the largest it can publish (a row for +/// each of the 8,192 PIDs, every counter at its limit) is under half this. So this only stops +/// a publisher whose patches keep adding keys. +const MAX_REPORT: usize = 8 * 1024 * 1024; /// How long `export echo` lets a reader drain after its broadcast ends. The final snapshot can /// still be in flight, but the track can also outlive its route. @@ -131,12 +135,28 @@ fn transport(connection: Option<&Monitor>, direction: Direction) -> hang::stats: transport } +/// The MPEG-TS sections a report carries beside its media counters, as the import or export +/// loop that samples them last set them. Empty for every other container. +#[derive(Clone, Default)] +pub struct Sections(Arc>); + +impl Sections { + pub fn set(&self, update: impl FnOnce(&mut ts::stats::Report)) { + update(&mut self.0.lock().unwrap_or_else(PoisonError::into_inner)); + } + + fn get(&self) -> ts::stats::Report { + self.0.lock().unwrap_or_else(PoisonError::into_inner).clone() + } +} + /// An import's stats track: what each of its tracks has written, from its catalog's counters, -/// and the connection's transport. +/// the connection's transport, and any MPEG-TS sections. pub struct Stats { - tracks: Pair, + tracks: Pair>, renditions: Box BTreeMap + Send>, connection: Option, + sections: Sections, } impl Stats { @@ -158,13 +178,20 @@ impl Stats { tracks, renditions: Box::new(move || catalog.stats()), connection, + sections: Sections::default(), }) } - fn snapshot(&self) -> hang::stats::Snapshot { + /// Where the import sets its MPEG-TS sections. + pub fn sections(&self) -> Sections { + self.sections.clone() + } + + fn snapshot(&self) -> hang::stats::Snapshot { let mut snapshot = hang::stats::Snapshot::default(); snapshot.transport = transport(self.connection.as_ref(), Direction::Send); snapshot.renditions = (self.renditions)(); + snapshot.ext = self.sections.get(); snapshot } @@ -232,18 +259,19 @@ struct Feedback { path: moq_net::PathOwned, // Held for the route: dropping it retracts the announcement. _broadcast: moq_net::broadcast::Producer, - tracks: Pair, + tracks: Pair>, } /// An export's solicited feedback: once its catalog carries an `echo` section, a /// `/.echo` broadcast whose `feedback.json` reports what each rendition the export -/// reads has received, and the connection's transport. Each catalog update reconciles it: a -/// removed section or a changed path retracts the broadcast it replaces. It follows the catalog -/// of the instance the export first reads, and ends when the export does. +/// reads has received, the connection's transport, and any MPEG-TS sections. Each catalog update +/// reconciles it: a removed section or a changed path retracts the broadcast it replaces. It +/// follows the catalog of the instance the export first reads, and ends when the export does. pub struct Echo { target: EchoTarget, source: moq_mux::Source, format: CatalogFormat, + sections: Sections, feedback: Option, } @@ -254,10 +282,16 @@ impl Echo { target, source, format, + sections: Sections::default(), feedback: None, } } + /// Where the export sets its MPEG-TS sections. + pub fn sections(&self) -> Sections { + self.sections.clone() + } + /// Follow the catalog's `echo` section and write a snapshot every interval, for as long as the /// export runs beside it. pub async fn run(&mut self) -> Infallible { @@ -331,6 +365,7 @@ impl Echo { let mut snapshot = hang::echo::Snapshot::default(); snapshot.transport = transport(self.target.connection.as_ref(), Direction::Receive); snapshot.renditions = self.source.stats(); + snapshot.ext = self.sections.get(); feedback.tracks.update(&snapshot); } @@ -580,6 +615,85 @@ mod tests { assert!(matches!(read.next().await, Err(moq_json::Error::TooLarge(MAX_REPORT)))); } + /// The largest report a TS import can publish reads whole: a row for each of the 8,192 + /// PIDs, every counter at its limit, beside the 1,024 renditions a registry holds. + #[tokio::test] + async fn the_largest_ts_report_reads_whole() { + const SAFE: u64 = hang::catalog::MAX_SAFE_INTEGER; + let mut mpegts = ts::stats::Snapshot::default(); + for counter in [ + &mut mpegts.crc_error, + &mut mpegts.ts_sync_loss, + &mut mpegts.sync_byte_error, + &mut mpegts.pat_error, + &mut mpegts.continuity_count_error, + &mut mpegts.pmt_error, + &mut mpegts.transport_error, + &mut mpegts.pcr_repetition_error, + &mut mpegts.pcr_discontinuity_indicator_error, + &mut mpegts.pts_error, + ] { + *counter = u64::MAX; + } + for pid in 0..8192 { + let mut row = ts::stats::Stream::default(); + row.track = ".avc3".to_string(); + row.class = ts::stats::Class::Video; + row.quiet = Some(Duration::from_millis(u64::MAX)); + for counter in [ + &mut row.units, + &mut row.resyncs, + &mut row.discarded, + &mut row.unconfirmed, + &mut row.damaged, + &mut row.unplaced, + &mut row.refused, + &mut row.continuity_count_error, + &mut row.transport_error, + &mut row.pts_error, + ] { + *counter = u64::MAX; + } + mpegts.streams.insert(pid, row); + } + let mut report = hang::stats::Snapshot::::default(); + report.ext.mpegts = Some(mpegts); + report.transport.rtt = Some(Duration::from_millis(SAFE)); + report.transport.age = Some(Duration::from_millis(SAFE)); + ( + report.transport.rate, + report.transport.lost_bytes, + report.transport.lost_packets, + ) = (Some(SAFE), Some(SAFE), Some(SAFE)); + for n in 0..1024 { + let mut rendition = hang::stats::Rendition::default(); + ( + rendition.frames, + rendition.bytes, + rendition.keyframes, + rendition.skipped, + ) = (SAFE, SAFE, SAFE, SAFE); + rendition.bitrate = Some(SAFE); + report.renditions.insert(format!("0x{n:04x}-0x1b.avc3"), rendition); + } + let value = serde_json::to_value(&report).unwrap(); + + let track = moq_net::broadcast::Info::new() + .produce() + .create_track(STATS_TRACK, None) + .unwrap(); + let mut read = reader(track.subscribe(None)); + let mut producer = moq_json::snapshot::Producer::::new(track, Default::default()); + producer.update(&value).unwrap(); + assert_eq!(read.latest().await.unwrap().unwrap().value, value); + + let size = serde_json::to_vec(&value).unwrap().len(); + assert!( + size <= MAX_REPORT / 2, + "a {size}-byte report leaves the cap too little room" + ); + } + /// A feedback track can outlive its route, so a reader left to drain stops after `DRAIN`. #[tokio::test(start_paused = true)] async fn an_ended_feedback_reader_stops_after_draining() { diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 79aa364e93..ff9c031e51 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -253,6 +253,23 @@ pub struct Subscribe { source: moq_mux::Source, catalog: CatalogFormat, args: SubscribeArgs, + /// Where a TS export sets its MPEG-TS sections, when it reports them. + sections: Option, +} + +/// Log how a TS export's release clock keeps up with the source. +fn log_release(release: &moq_mux::container::ts::stats::Release, message: &'static str) { + use moq_mux::container::ts::stats::SignedDuration; + tracing::info!( + dropped = release.dropped, + trimmed = release.trimmed, + gaps = release.gaps, + margin_ms = ?release.margin.map(SignedDuration::as_millis), + offset_ms = ?release.offset.map(SignedDuration::as_millis), + drift_ppm = ?release.drift, + out_of_tolerance = release.out_of_tolerance, + "{message}" + ); } impl Subscribe { @@ -276,6 +293,7 @@ impl Subscribe { source, catalog, args, + sections: None, }) } @@ -284,6 +302,12 @@ impl Subscribe { &self.source } + /// Set a TS export's per-PID rows and release clock in `sections` as it samples them. + pub fn with_sections(mut self, sections: crate::report::Sections) -> Self { + self.sections = Some(sections); + self + } + /// Build the catalog stream, narrowed by the rendition selection flags. The /// catalog source honors the requested format (e.g. compressed `HangZ` or `Msf`). async fn stream(&self) -> anyhow::Result> { @@ -406,15 +430,14 @@ impl Subscribe { let mut sampled = tokio::time::Instant::now(); // The export's drop, gap and out-of-tolerance counts, as last logged. let mut reported = (0, 0, 0, 0); - let release = |stats: &moq_mux::container::ts::stats::Export| { - tracing::info!( - dropped = stats.dropped, - trimmed = stats.trimmed, - gaps = stats.gaps, - drift_ppm = ?stats.drift, - out_of_tolerance = stats.out_of_tolerance, - "TS export release clock" - ); + let sections = self.sections; + let report = |stats: &moq_mux::container::ts::stats::Export| { + if let Some(sections) = §ions { + sections.set(|report| { + report.mpegts = Some(stats.clone().into()); + report.release = Some(stats.release.clone()); + }); + } }; let end = loop { let frame = match ts.next().await { @@ -428,15 +451,19 @@ impl Subscribe { if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL { sampled = tokio::time::Instant::now(); let stats = ts.export().stats(); - let counts = (stats.dropped, stats.trimmed, stats.gaps, stats.out_of_tolerance); + report(&stats); + let release = &stats.release; + let counts = (release.dropped, release.trimmed, release.gaps, release.out_of_tolerance); if counts != reported { reported = counts; - release(&stats); + log_release(release, "TS export release clock"); } log.sample(stats.into()); } }; - release(&ts.export().stats()); + let stats = ts.export().stats(); + report(&stats); + log_release(&stats.release, "TS export release clock"); match end { Err(err @ moq_mux::Error::Replaced(_)) => { Err(anyhow::Error::from(err).context("pass --stitch to follow a replacement as a program switch")) @@ -452,30 +479,31 @@ impl Subscribe { let mut release = moq_mux::container::ts::Release::new(self.source, self.catalog) .await? .with_delay(self.args.max_delay); - let log = |release: &moq_mux::container::ts::Release| { - tracing::info!( - dropped = release.dropped(), - drift_ppm = ?release.drift().map(|drift| drift * 1e6), - out_of_tolerance = release.out_of_tolerance(), - "TS passthrough release clock" - ); + let report = |stats: &moq_mux::container::ts::stats::Release| { + if let Some(sections) = &self.sections { + sections.set(|report| report.release = Some(stats.clone())); + } }; let mut sampled = tokio::time::Instant::now(); - let mut reported = (0, 0); + let mut reported = (0, 0, 0, 0); while let Some(object) = release.next().await? { stdout.write_all(&object.payload).await?; stdout.flush().await?; if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL { sampled = tokio::time::Instant::now(); - let counts = (release.dropped(), release.out_of_tolerance()); + let stats = release.stats(); + report(&stats); + let counts = (stats.dropped, stats.trimmed, stats.gaps, stats.out_of_tolerance); if counts != reported { reported = counts; - log(&release); + log_release(&stats, "TS passthrough release clock"); } } } - log(&release); + let stats = release.stats(); + report(&stats); + log_release(&stats, "TS passthrough release clock"); Ok(()) } diff --git a/rs/moq-cli/tests/integration/stats.rs b/rs/moq-cli/tests/integration/stats.rs index a9a5eec651..acbd1ae7b5 100644 --- a/rs/moq-cli/tests/integration/stats.rs +++ b/rs/moq-cli/tests/integration/stats.rs @@ -20,9 +20,10 @@ const PACE: Duration = Duration::from_millis(150); /// Past the two seconds of PCR a live TS catalog waits for, so the broadcast is up by then. const HALF: usize = CLIP.len() / 2; -/// Feed `clip` to `write` at about real time. +/// Feed `clip` to `write` at about real time, never catching up in a burst. async fn feed(clip: &[u8], mut write: impl AsyncFnMut(&[u8])) { let mut pace = tokio::time::interval(PACE); + pace.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); for chunk in clip.chunks(CHUNK) { pace.tick().await; write(chunk).await; @@ -91,8 +92,9 @@ async fn wait(child: &mut Child) -> std::process::ExitStatus { .expect("wait for moq") } -/// What a TS import of `clip` writes to each rendition, from an in-process import. -fn written(clip: &[u8]) -> BTreeMap { +/// What a TS import of `clip` writes to each rendition, and its MPEG-TS counters, from an +/// in-process import. +fn written(clip: &[u8]) -> (BTreeMap, ts::stats::Snapshot) { let mut broadcast = moq_net::broadcast::Info::new().produce(); let config = moq_mux::catalog::Config::default().with_catalog(moq_mux::catalog::hang::Catalog::::default()); @@ -100,16 +102,120 @@ fn written(clip: &[u8]) -> BTreeMap { let mut import = ts::Import::new(broadcast, catalog.reserve()); import.decode(clip).unwrap(); import.finish().unwrap(); - catalog.stats() + (catalog.stats(), import.stats()) } -/// The stats an import names in its catalog count each rendition while it runs, and its last -/// report holds what the import wrote in full. +/// Each elementary stream's access units, by PID. +fn units(mpegts: &ts::stats::Snapshot) -> BTreeMap { + mpegts.streams.iter().map(|(pid, row)| (*pid, row.units)).collect() +} + +/// Every TR 101 290 counter, PSI CRC error, and per-PID loss reads zero. +fn assert_clean(mpegts: &ts::stats::Snapshot) { + let mut stream = mpegts.clone(); + stream.streams.clear(); + assert_eq!( + stream, + ts::stats::Snapshot::default(), + "a stream-wide error was counted" + ); + for (pid, row) in &mpegts.streams { + let lost = [ + row.resyncs, + row.discarded, + row.unconfirmed, + row.damaged, + row.continuity_count_error, + row.transport_error, + row.pts_error, + ]; + assert_eq!(lost, [0; 7], "PID {pid} counted a loss: {row:?}"); + } +} + +/// The PCR a packet's adaptation field carries, in 27 MHz ticks. +fn pcr(pkt: &[u8; 188]) -> Option { + if pkt[3] & 0x20 == 0 || pkt[4] < 7 || pkt[5] & 0x10 == 0 { + return None; + } + let base = (u64::from(pkt[6]) << 25) + | (u64::from(pkt[7]) << 17) + | (u64::from(pkt[8]) << 9) + | (u64::from(pkt[9]) << 1) + | (u64::from(pkt[10]) >> 7); + Some(base * 300 + ((u64::from(pkt[10] & 0x01) << 8) | u64::from(pkt[11]))) +} + +/// `clip` with the PES on `pid` suppressed from its first PES start at or after `from` seconds +/// of program clock to its first at or after `to`, as an encoder whose one input died behind a +/// running mux emits it: a packet that carried the PCR keeps it, every other becomes null +/// stuffing, and the counters after the gap are renumbered so continuity stays legal. The +/// moq-mux import tests' stimulus. +fn suppress(clip: &[u8], pid: u16, from: f64, to: f64) -> Vec { + let mut null = [0xff; 188]; + null[..4].copy_from_slice(&[0x47, 0x1f, 0xff, 0x10]); + + let mut out = Vec::with_capacity(clip.len()); + let (mut first, mut now) = (None, 0.0); + let (mut active, mut done) = (false, false); + let (mut last_cc, mut dropped) = (0, 0u8); + for pkt in clip.as_chunks::<188>().0 { + let mut pkt = *pkt; + if let Some(pcr) = pcr(&pkt) { + now = (pcr - *first.get_or_insert(pcr)) as f64 / 27e6; + } + if (u16::from(pkt[1] & 0x1f) << 8 | u16::from(pkt[2])) == pid { + if pkt[1] & 0x40 != 0 { + if !active && !done && now >= from { + active = true; + } else if active && now >= to { + (active, done) = (false, true); + } + } + let payload = pkt[3] & 0x10 != 0; + if active { + dropped = (dropped + u8::from(payload)) & 0x0f; + pkt = match pcr(&pkt) { + Some(_) => { + let mut clock = [0xff; 188]; + clock[..6].copy_from_slice(&[0x47, pkt[1] & 0x1f, pkt[2], 0x20 | last_cc, 183, 0x10]); + clock[6..12].copy_from_slice(&pkt[6..12]); + clock + } + None => null, + }; + } else { + pkt[3] = (pkt[3] & 0xf0) | (pkt[3].wrapping_sub(dropped) & 0x0f); + if payload { + last_cc = pkt[3] & 0x0f; + } + } + } + out.extend_from_slice(&pkt); + } + assert!(done, "the clip resumes the PID before it ends"); + out +} + +/// Each PID's access units in `first` fell short of `last`'s: the rows advance while it runs. +fn assert_advancing(first: &ts::stats::Snapshot, last: &ts::stats::Snapshot) { + assert!(!last.streams.is_empty(), "no elementary stream reported"); + let first = units(first); + for (pid, units) in units(last) { + let before = first.get(&pid).copied().unwrap_or(0); + assert!(before < units, "PID {pid}: {before} access units, then {units}"); + } +} + +/// The stats an import names in its catalog count each rendition and elementary stream while +/// it runs, and its last report holds what the import wrote in full, with every TR 101 290 +/// counter at zero on a clean clip. #[tokio::test] async fn import_stats_count_what_each_rendition_wrote() { let relay = relay().await; - let expected = written(CLIP); + let (expected, health) = written(CLIP); assert_eq!(expected.len(), 2, "the clip carries video and audio: {expected:?}"); + assert_clean(&health); let mut import = moq(&relay, "demo", &["import", "--stats", "--max-age", "2s", "ts"]) .stdin(Stdio::piped()) @@ -124,7 +230,8 @@ async fn import_stats_count_what_each_rendition_wrote() { stdin.write_all(chunk).await.expect("write stdin") }) .await; - let mut reports: Vec = vec![next(&mut lines).await.expect("a report mid-import")]; + let mut reports: Vec> = + vec![next(&mut lines).await.expect("a report mid-import")]; feed(&CLIP[HALF..], async |chunk| { stdin.write_all(chunk).await.expect("write stdin") }) @@ -150,6 +257,83 @@ async fn import_stats_count_what_each_rendition_wrote() { reports.iter().any(|report| report.transport.rtt.is_some()), "the import reports its connection" ); + + let mpegts = last.ext.mpegts.as_ref().expect("a TS import reports its counters"); + assert_clean(mpegts); + assert_eq!(units(mpegts), units(&health), "the last report holds every access unit"); + let first = reports.iter().find_map(|report| report.ext.mpegts.as_ref()).unwrap(); + assert_advancing(first, mpegts); + assert!( + reports.iter().all(|report| report.ext.release.is_none()), + "an import has no release clock" + ); +} + +/// A video PID whose PES stops while the mux keeps its PCR and continuity legal: the import's +/// reports show that row's `quiet` growing while every other row keeps delivering, the row +/// counting again once it returns, and no TR 101 290 priority 1 check firing. +#[tokio::test] +async fn import_stats_report_a_silent_video_pid() { + let relay = relay().await; + let (_, clean) = written(CLIP); + let pid = |class| { + let mut rows = clean.streams.iter().filter(|(_, row)| row.class == class); + *rows.next().expect("the clip carries one").0 + }; + let (video, audio) = (pid(ts::stats::Class::Video), pid(ts::stats::Class::Audio)); + let stimulus = suppress(CLIP, video, 2.5, 4.5); + let (_, health) = written(&stimulus); + + let mut import = moq(&relay, "demo", &["import", "--stats", "--max-age", "2s", "ts"]) + .stdin(Stdio::piped()) + .spawn() + .expect("spawn the import"); + let mut stdin = import.stdin.take().expect("stdin"); + let (mut sink, mut lines) = sink(&relay, "demo", "stats"); + + // Up to the silence, then a report, so the sink is reading before the video stops. + feed(&stimulus[..HALF], async |chunk| { + stdin.write_all(chunk).await.expect("write stdin") + }) + .await; + let mut reports: Vec> = + vec![next(&mut lines).await.expect("a report mid-import")]; + feed(&stimulus[HALF..], async |chunk| { + stdin.write_all(chunk).await.expect("write stdin") + }) + .await; + drop(stdin); + while let Some(report) = next(&mut lines).await { + reports.push(report); + } + assert!(wait(&mut import).await.success()); + assert!(wait(&mut sink).await.success()); + + let rows: Vec<&ts::stats::Snapshot> = reports.iter().filter_map(|report| report.ext.mpegts.as_ref()).collect(); + let quiet = |mpegts: &ts::stats::Snapshot, pid: u16| mpegts.streams[&pid].quiet.unwrap_or_default(); + let silent = rows.iter().map(|mpegts| quiet(mpegts, video)).max().unwrap(); + // Two seconds of silence, sampled about once a second. + assert!( + silent >= Duration::from_millis(500), + "the video row was quiet for {silent:?} at most" + ); + let busy = rows.iter().map(|mpegts| quiet(mpegts, audio)).max().unwrap(); + assert!( + busy < Duration::from_millis(500), + "the audio row was quiet for {busy:?}" + ); + + let last = rows.last().unwrap(); + assert_eq!(units(last), units(&health), "the last report holds every access unit"); + assert!(quiet(last, video) < silent, "the video row delivered again"); + let priority1 = [ + last.ts_sync_loss, + last.sync_byte_error, + last.pat_error, + last.continuity_count_error, + last.pmt_error, + ]; + assert_eq!(priority1, [0; 5], "the silence is legal on the wire"); } /// A sink that joins once the import has finished, while it lingers, still reads its final @@ -157,7 +341,7 @@ async fn import_stats_count_what_each_rendition_wrote() { #[tokio::test] async fn a_late_sink_reads_the_final_stats() { let relay = relay().await; - let expected = written(CLIP); + let (expected, health) = written(CLIP); let mut import = moq(&relay, "demo", &["import", "--stats", "--max-age", "10s", "ts"]) .stdin(Stdio::piped()) @@ -179,8 +363,9 @@ async fn a_late_sink_reads_the_final_stats() { ended.await.expect("the import never reached the end of its input"); let (mut sink, mut lines) = sink(&relay, "demo", "stats"); - let report: hang::stats::Snapshot = next(&mut lines).await.expect("the final report"); + let report: hang::stats::Snapshot = next(&mut lines).await.expect("the final report"); assert_eq!(report.renditions, expected); + assert_eq!(report.ext.mpegts.as_ref().map(units), Some(units(&health))); sink.kill().await.unwrap(); import.kill().await.unwrap(); } @@ -189,15 +374,16 @@ async fn a_late_sink_reads_the_final_stats() { #[derive(serde::Deserialize)] struct Feedback { broadcast: String, - feedback: hang::echo::Snapshot, + feedback: hang::echo::Snapshot, } /// An export asked for feedback publishes it where the catalog solicits it, counting what it -/// reads as it reads it, and `export echo` finds it under that prefix. +/// reads as it reads it, with a TS export's per-PID rows and release clock, and `export echo` +/// finds it under that prefix. #[tokio::test] async fn export_echo_reports_what_the_export_reads() { let relay = relay().await; - let expected = written(CLIP); + let (expected, _) = written(CLIP); // A publisher soliciting feedback under `viewers`, a sibling of its broadcast. let mut broadcast = relay.origin.create_broadcast("demo").expect("create the broadcast"); @@ -212,7 +398,8 @@ async fn export_echo_reports_what_the_export_reads() { broadcast.announce(moq_net::origin::Route::default()).unwrap(); let (mut sink, mut lines) = sink(&relay, "viewers", "echo"); - let mut export = moq(&relay, "demo", &["export", "--echo", "alice", "ts"]) + // Held longer than the feed pauses for the first feedback line, so the pause is no stall. + let mut export = moq(&relay, "demo", &["export", "--echo", "alice", "ts", "--delay", "4s"]) .stdout(Stdio::piped()) .spawn() .expect("spawn the export"); @@ -268,4 +455,21 @@ async fn export_echo_reports_what_the_export_reads() { feedback.iter().any(|line| line.feedback.transport.rtt.is_some()), "the export reports its connection" ); + + let rows = last.ext.mpegts.as_ref().expect("a TS export reports its rows"); + assert_clean(rows); + let first = feedback + .iter() + .find_map(|line| line.feedback.ext.mpegts.as_ref()) + .unwrap(); + assert_advancing(first, rows); + let release = last + .ext + .release + .as_ref() + .expect("a TS export reports its release clock"); + assert!( + release.margin.is_some(), + "a second of frames gives a margin: {release:?}" + ); } diff --git a/rs/moq-mux/MIGRATE.md b/rs/moq-mux/MIGRATE.md index 12aab1196f..4f91eaa5d1 100644 --- a/rs/moq-mux/MIGRATE.md +++ b/rs/moq-mux/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `ts::stats::Export`'s release counters move into `release` (#5849) + +`dropped`, `trimmed`, `gaps`, `drift`, and `out_of_tolerance` are now fields of `Export::release`, a `ts::stats::Release`, which `ts::Release::stats` also returns on its own. +Read `export.release.dropped` where you read `export.dropped`, and likewise for the rest. + ### Catalog read errors use `hang`'s variants (#6114) `catalog::hang::Consumer` reports a lost catalog track as `Error::Moq` instead of `Error::Json(moq_json::Error::Net(_))`, and any other catalog decode failure (such as a bad DEFLATE stream) as `Error::Hang(hang::Error::Json(_))` instead of `Error::Json(_)`. diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 7ad13c6f47..4318907b16 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -1683,20 +1683,18 @@ where /// empty until the program tables are built. Cheap enough to poll per frame. pub fn stats(&self) -> super::stats::Export { let mut stats = super::stats::Export { - dropped: self.jitter.dropped(), - trimmed: self.jitter.trimmed() + self.trimmed, - gaps: self - .tracks - .values() - .map(|track| { - let skips = track.base.1 + track.source.skips(); - skips - (track.base.0 + track.source.markers()) - }) - .sum(), - drift: self.jitter.drift().map(|drift| drift * 1e6), - out_of_tolerance: self.jitter.out_of_tolerance(), + release: super::stats::Release::of(&self.jitter), ..Default::default() }; + stats.release.trimmed += self.trimmed; + stats.release.gaps = self + .tracks + .values() + .map(|track| { + let skips = track.base.1 + track.source.skips(); + skips - (track.base.0 + track.source.markers()) + }) + .sum(); if self.psi.is_none() { return stats; } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index a8e9470edd..97530a6b4b 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -1049,7 +1049,7 @@ async fn export_trims_leading_pictures_after_each_delivery_break() { video, expected, "continuous leading pictures survive; only join and later-hole orphans are trimmed" ); - let stats = export.stats(); + let stats = export.stats().release; assert_eq!( (stats.dropped, stats.trimmed), (2, 2), @@ -5563,7 +5563,7 @@ async fn a_start_group_cut_short_starts_at_the_next() { audio.iter().all(|&p| p >= pts(GOP)), "no audio leads the keyframe the export starts at: {audio:?}" ); - let stats = export.stats(); + let stats = export.stats().release; assert_eq!(stats.dropped, 0, "nothing that went out lost a frame"); assert_eq!(stats.gaps, 1, "the cut is counted"); assert!(stats.trimmed >= 5, "the cut-off start is counted: {}", stats.trimmed); @@ -5640,7 +5640,7 @@ async fn audio_dropped_ahead_of_a_moved_start_is_not_started() { audio.iter().all(|&p| p >= pts(GOP)), "no audio leads the keyframe: {audio:?}" ); - let stats = export.stats(); + let stats = export.stats().release; assert_eq!(stats.dropped, 0, "nothing was lost from the output"); assert!(stats.trimmed > 0); } @@ -5684,7 +5684,7 @@ async fn a_stale_start_group_holds_the_audio_behind_the_next_keyframe() { audio.iter().all(|&p| p >= pts(GOP)), "no audio leads the keyframe the export starts at: {audio:?}" ); - let stats = export.stats(); + let stats = export.stats().release; assert_eq!(stats.dropped, 0, "nothing was lost from the output"); assert!(stats.trimmed > 0); } diff --git a/rs/moq-mux/src/container/ts/export_timing_test.rs b/rs/moq-mux/src/container/ts/export_timing_test.rs index 26d0326f5c..37d0ff918c 100644 --- a/rs/moq-mux/src/container/ts/export_timing_test.rs +++ b/rs/moq-mux/src/container/ts/export_timing_test.rs @@ -20,6 +20,7 @@ use crate::catalog::hang::Container as HangContainer; use crate::container::ts::Export; use crate::container::ts::export::PCR_INTERVAL; use crate::container::ts::stats::Export as Stats; +use crate::container::ts::stats::SignedDuration; use crate::container::{Container as _, Frame, Producer}; const SPS: &[u8] = &[0x67, 0x42, 0xc0, 0x1f, 0xde]; @@ -714,8 +715,15 @@ async fn a_slow_source_clock_keeps_the_delay() { let (spread, dropped, stats) = on_a_scaled_clock(1.0 - 30e-6, DRIFT_TICKS).await; assert_eq!(dropped, 0, "frames went late as the source fell behind the anchor"); assert!(spread <= DRIFT_SLACK, "the delay walked by {spread:?}"); - let drift = stats.drift.expect("a measurement"); + let drift = stats.release.drift.expect("a measurement"); assert!((drift + 30.0).abs() < 0.5, "measured {drift} ppm"); + let margin = stats.release.margin.expect("a margin"); + assert!( + margin > SignedDuration::default(), + "a clean path keeps time to spare: {} ms", + margin.as_millis() + ); + assert!(stats.release.offset.is_some(), "the offset is measured with the rate"); } #[tokio::test(start_paused = true)] @@ -723,7 +731,7 @@ async fn a_fast_source_clock_keeps_the_delay() { let (spread, dropped, stats) = on_a_scaled_clock(1.0 + 30e-6, DRIFT_TICKS).await; assert_eq!(dropped, 0); assert!(spread <= DRIFT_SLACK, "the delay walked by {spread:?}"); - let drift = stats.drift.expect("a measurement"); + let drift = stats.release.drift.expect("a measurement"); assert!((drift - 30.0).abs() < 0.5, "measured {drift} ppm"); } @@ -731,9 +739,9 @@ async fn a_fast_source_clock_keeps_the_delay() { #[tokio::test(start_paused = true)] async fn a_source_past_the_limit_is_counted() { let (_, _, stats) = on_a_scaled_clock(1.0004, 300_000_000 / VIDEO_US).await; - let drift = stats.drift.expect("a measurement"); + let drift = stats.release.drift.expect("a measurement"); assert!((drift - 400.0).abs() < 1.0, "measured {drift} ppm"); - assert!(stats.out_of_tolerance > 0, "not counted"); + assert!(stats.release.out_of_tolerance > 0, "not counted"); } /// The same source fails the export once it has used half the delay, about ten minutes in, diff --git a/rs/moq-mux/src/container/ts/release.rs b/rs/moq-mux/src/container/ts/release.rs index 27457ce4e5..af89466185 100644 --- a/rs/moq-mux/src/container/ts/release.rs +++ b/rs/moq-mux/src/container/ts/release.rs @@ -182,6 +182,15 @@ impl Release { self.jitter.out_of_tolerance() } + /// Snapshot how the release clock is keeping up with the source. + pub fn stats(&self) -> super::stats::Release { + let mut stats = super::stats::Release::of(&self.jitter); + if let Some((_, track)) = &self.track { + stats.gaps = track.skips() - track.markers(); + } + stats + } + /// How many objects are waiting in the jitter buffer. #[cfg(test)] pub(super) fn buffered(&self) -> usize { diff --git a/rs/moq-mux/src/container/ts/stats.rs b/rs/moq-mux/src/container/ts/stats.rs index 059fd5b2b1..b2f4a97fe9 100644 --- a/rs/moq-mux/src/container/ts/stats.rs +++ b/rs/moq-mux/src/container/ts/stats.rs @@ -1,8 +1,33 @@ //! The counters [`Import::stats`](super::Import::stats) and //! [`Export::stats`](super::Export::stats) return, and reporting on them. +//! +//! [`Report`] carries them in a media report, as the `mpegts` and `release` sections beside +//! `transport` and `renditions`. Counters are cumulative and omitted at zero. use std::collections::{BTreeMap, BTreeSet}; +use serde::{Deserialize, Serialize}; + +fn is_zero(value: &u64) -> bool { + *value == 0 +} + +/// The sections an MPEG-TS import or export adds to its media report: the `E` of +/// [`hang::stats::Snapshot`] or [`hang::echo::Snapshot`], flattened beside `transport` and +/// `renditions`. +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +#[serde(default)] +#[non_exhaustive] +pub struct Report { + /// What each elementary stream delivered and the transport stream's errors: an import's + /// [`Snapshot`], or the rows of an [`Export`]. A passthrough has none. + #[serde(skip_serializing_if = "Option::is_none")] + pub mpegts: Option, + /// How an export's release clock keeps up with the source. An import has none. + #[serde(skip_serializing_if = "Option::is_none")] + pub release: Option, +} + /// What each demuxed elementary stream delivered, and the audio frame sync it lost or could /// not verify, keyed by elementary stream PID; the PSI sections the stream as a whole lost to /// corruption; and its TR 101 290 errors. @@ -20,52 +45,67 @@ use std::collections::{BTreeMap, BTreeSet}; /// /// [`Export`] carries the same rows for the streams an exporter writes, so one schema reads /// both edges. -#[derive(Clone, Debug, Default, PartialEq, Eq)] +#[serde_with::serde_as] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default, rename_all = "camelCase")] #[non_exhaustive] pub struct Snapshot { /// 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. + // Keys read from strings: a flattened section loses JSON's integer-key parsing. + #[serde_as(as = "BTreeMap")] + #[serde(skip_serializing_if = "BTreeMap::is_empty")] 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 /// next good repetition. Counted for the stream rather than per elementary stream, since /// the PSI PIDs have no row of their own. TR 101 290 2.2 `CRC_error`, on the PAT and PMT /// only: the reduced-SI set of its table 5.1b. Captured SI is not CRC-checked. + #[serde(skip_serializing_if = "is_zero")] pub crc_error: u64, /// 1.1 `TS_sync_loss`: sync lost after two consecutive corrupt sync bytes on the packet /// grid, and acquired again after five correct ones (ISO/IEC 13818-1 G.1). + #[serde(skip_serializing_if = "is_zero")] pub ts_sync_loss: u64, /// 1.2 `Sync_byte_error`: a byte other than 0x47 where the grid expects a sync byte, while /// in sync. + #[serde(skip_serializing_if = "is_zero")] pub sync_byte_error: u64, /// 1.3 `PAT_error_2`: each 0.5 s of program clock without a PAT section on PID 0, and each /// packet on PID 0 that is scrambled or starts a section other than a PAT. + #[serde(skip_serializing_if = "is_zero")] pub pat_error: u64, /// 1.4 `Continuity_count_error`, on every PID but the null PID: a counter gap, or a third /// copy of a packet. A gap declared by `discontinuity_indicator`, the one duplicate ISO/IEC /// 13818-1 2.4.3.3 permits, and a payload-less packet repeating its counter are not errors. /// A loss of exactly 16 packets leaves the counter in step, and only a byte comparison /// tells a loss of 15 from a duplicate, so the count is a floor. + #[serde(skip_serializing_if = "is_zero")] pub continuity_count_error: u64, /// 1.5 `PMT_error_2`: each 0.5 s of program clock without a PMT section on a PMT PID the /// PAT names for the imported program, and each scrambled packet on one. + #[serde(skip_serializing_if = "is_zero")] pub pmt_error: u64, /// 2.1 `Transport_error`: packets with `transport_error_indicator` set, on any PID. Such a /// packet counts nothing else. + #[serde(skip_serializing_if = "is_zero")] pub transport_error: u64, /// 2.3a `PCR_repetition_error`: consecutive PCR values on the PCR PID more than 100 ms /// apart. Graded on the values, not on arrival: that speaks for the encoder's insertion, /// which is what ingest can, while the network in front of the importer re-times arrival. /// The interval across a `discontinuity_indicator` is not graded. + #[serde(skip_serializing_if = "is_zero")] pub pcr_repetition_error: u64, /// 2.3b `PCR_discontinuity_indicator_error`: consecutive PCR values outside 0 to 100 ms /// apart, backwards included, without `discontinuity_indicator`. On values a run of missing /// PCRs and an unsignalled jump are the same difference, so a forward step over 100 ms /// counts here and as a repetition error both. + #[serde(skip_serializing_if = "is_zero")] pub pcr_discontinuity_indicator_error: u64, /// 2.5 `PTS_error`: each 700 ms of program clock without a PTS on an elementary stream /// that has carried one. Verbatim streams are not graded: SCTE-35 and subtitles have no /// cadence to hold. + #[serde(skip_serializing_if = "is_zero")] pub pts_error: u64, } @@ -77,55 +117,70 @@ impl Snapshot { } /// What one elementary stream delivered, lost, or could not verify. See [`Snapshot`]. -#[derive(Clone, Debug, Default, PartialEq, Eq)] +#[serde_with::serde_as] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default, rename_all = "camelCase")] #[non_exhaustive] pub struct Stream { /// 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. At export, /// the suffix an import of the output would give the PID. + #[serde(skip_serializing_if = "String::is_empty")] pub track: String, /// How this PID was classified. See [`Class`]. pub class: Class, /// Access units the stream delivered: frames published for decoded media, whole PES or /// sections carried verbatim, and PES read on MPEG-1/2 video. At export, the PES or /// sections written for each frame. + #[serde(skip_serializing_if = "is_zero")] 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): the one read at import, the one - /// written at export. `None` until the first PCR. + /// written at export. `None` until the first PCR. Whole milliseconds in a report. /// /// 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. + #[serde_as(as = "Option>")] + #[serde(skip_serializing_if = "Option::is_none")] pub quiet: Option, /// Completed resyncs: the stream lost frame sync and locked onto a confirmed frame /// again. Each one is a gap in the audio. + #[serde(skip_serializing_if = "is_zero")] pub resyncs: u64, /// Bytes thrown away scanning for sync, across every resync and any scan still in /// progress. + #[serde(skip_serializing_if = "is_zero")] pub discarded: u64, /// Frames published that nothing vouched for: the offset came from a scan, or the bytes /// were joined onto a tail carried across a PES boundary, and no successor frame could /// confirm it. That substitutes audio rather than leaving a gap, which is why it is /// counted separately from a resync. + #[serde(skip_serializing_if = "is_zero")] pub unconfirmed: u64, /// Damaged packets, PES, or access units refused whole on this PID. Export stays zero. + #[serde(skip_serializing_if = "is_zero")] pub damaged: u64, /// Whole PES or sections dropped because the program clock could not place /// them: before the first PCR, after the last with no interval to extrapolate from, or /// before the broadcast clock's zero. Any PES lands before that zero while its stream is /// still behind a wrap the import's first PTS already crossed. Export stays zero. + #[serde(skip_serializing_if = "is_zero")] pub unplaced: u64, /// Times this PID's stream was refused rather than carried altered: an undecoded stream /// in a program with no PCR to place it by, or a private PES stream with a clock field only /// the PTS and DTS of are rebased (ESCR, a `pack_header_field`, a TREF), PES-level scrambling, /// an unbounded non-video PES, or a header export refuses. Its track ends and leaves the /// catalog. Export stays zero. + #[serde(skip_serializing_if = "is_zero")] pub refused: u64, /// This PID's share of [`Snapshot::continuity_count_error`]. + #[serde(skip_serializing_if = "is_zero")] pub continuity_count_error: u64, /// This PID's share of [`Snapshot::transport_error`]. + #[serde(skip_serializing_if = "is_zero")] pub transport_error: u64, /// This PID's share of [`Snapshot::pts_error`]. + #[serde(skip_serializing_if = "is_zero")] pub pts_error: u64, } @@ -154,16 +209,20 @@ impl Stream { /// Not the PMT `stream_type`. `0x86` is DTS audio or, with a CUEI registration, an SCTE-35 /// section, and those take different routes. The stopped-stream log grades audio and video /// only; a sparse data PID stays counted. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] pub enum Class { - /// Undecoded or sparse data (SCTE-35, ID3, private sections, private PES), or a PCR PID - /// that carries no elementary stream. A quiet second is not a stall. - #[default] - Data, /// Continuous audio. - Audio, + Audio = 1, /// Continuous video, including MPEG-1/2 video read only for its clock. - Video, + Video = 2, + /// Undecoded or sparse data (SCTE-35, ID3, private sections, private PES), or a PCR PID + /// that carries no elementary stream. A quiet second is not a stall. A class a later + /// version adds reads as data. It must stay last for `#[serde(other)]`; the explicit + /// discriminants keep every variant's `as` value. + #[default] + #[serde(other)] + Data = 0, } impl Class { @@ -185,29 +244,126 @@ impl Class { pub struct Export { /// One row per elementary stream PID the exporter writes, ordered by PID. pub streams: BTreeMap, + /// How the release clock is keeping up with the source. + pub release: Release, +} + +/// How an export keeps up with the source: the frames it drops, trims, and never receives, +/// and how its release clock tracks the source's. What [`Export::stats`](super::Export::stats) +/// reports beside its rows, and [`Release::stats`](super::Release::stats) on its own. +#[serde_with::serde_as] +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +#[serde(default, rename_all = "camelCase")] +#[non_exhaustive] +pub struct Release { /// Frames dropped after their stream's first went out: for arriving after their deadline, /// for falling due before a new timeline's clock was acquired, and the frames a video track /// then dropped waiting for its next keyframe or leading the keyframe it resumed at, late /// or after a hole upstream once the output held frames ahead of the hole. + #[serde(skip_serializing_if = "is_zero")] pub dropped: u64, /// Frames dropped before their stream's first went out, so the output starts on a whole /// group led by the video keyframe: what a joiner is handed older than the delay, a start /// group cut short upstream, the first keyframe's leading pictures, and what precedes the /// first keyframe or the codec config. /// Nothing is missing from the output for these. + #[serde(skip_serializing_if = "is_zero")] pub trimmed: u64, /// Times a source skipped media it will never deliver: a group lost or expired upstream, /// or a resubscription. A declared break is not one. Each is logged with the span it /// skipped. + #[serde(skip_serializing_if = "is_zero")] pub gaps: u64, + /// The least time any frame arrived before its deadline over the last whole second: how + /// much more latency the path could take before frames drop. Negative when a frame missed + /// its deadline. `None` until a second has passed, and without a delay. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub margin: Option, + /// How far the output clock sits from the source's plus the delay, as last measured: + /// positive holds frames longer than the delay, negative shorter. The clock pulls it shut + /// over ten minutes. `None` until there is a measurement. + #[serde_as(as = "Option")] + #[serde(skip_serializing_if = "Option::is_none")] + pub offset: Option, /// How fast the source's clock runs against ours, in parts per million (positive runs /// fast), as last measured. `None` until there is a measurement. + #[serde(skip_serializing_if = "Option::is_none")] pub drift: Option, /// How many measurements found the source's clock further off ours than the 30 ppm an /// MPEG-TS system clock may be (ISO/IEC 13818-1 2.4.2.1), which the output cannot follow. + #[serde(skip_serializing_if = "is_zero")] pub out_of_tolerance: u64, } +impl Release { + pub(crate) fn of(jitter: &crate::jitter::Buffer) -> Self { + Self { + dropped: jitter.dropped(), + trimmed: jitter.trimmed(), + gaps: 0, + margin: jitter.margin().map(SignedDuration::from_secs_f64), + offset: jitter.offset().map(SignedDuration::from_secs_f64), + drift: jitter.drift().map(|drift| drift * 1e6), + out_of_tolerance: jitter.out_of_tolerance(), + } + } +} + +/// A span of time that can fall below zero, which [`Duration`](std::time::Duration) cannot: a +/// [`Release`]'s margin and offset. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct SignedDuration { + nanos: i64, +} + +impl SignedDuration { + /// `millis` milliseconds, below zero when negative. + pub const fn from_millis(millis: i64) -> Self { + Self { + nanos: millis.saturating_mul(1_000_000), + } + } + + /// `seconds`, below zero when negative, saturating at about 292 years either way. + pub(crate) fn from_secs_f64(seconds: f64) -> Self { + Self { + nanos: (seconds * 1e9).round() as i64, + } + } + + /// Whether the span falls below zero. + pub const fn is_negative(self) -> bool { + self.nanos < 0 + } + + /// How long the span is, whichever side of zero it falls. + pub const fn unsigned_abs(self) -> std::time::Duration { + std::time::Duration::from_nanos(self.nanos.unsigned_abs()) + } + + /// Whole milliseconds, rounded away from zero, so a span short of one keeps its sign: what + /// a report carries. + pub const fn as_millis(self) -> i64 { + self.nanos / 1_000_000 + (self.nanos % 1_000_000).signum() + } +} + +/// A [`SignedDuration`] in a report: [`SignedDuration::as_millis`]. +struct SignedMillis; + +impl serde_with::SerializeAs for SignedMillis { + fn serialize_as(value: &SignedDuration, serializer: S) -> Result { + value.as_millis().serialize(serializer) + } +} + +impl<'de> serde_with::DeserializeAs<'de, SignedDuration> for SignedMillis { + fn deserialize_as>(deserializer: D) -> Result { + i64::deserialize(deserializer).map(SignedDuration::from_millis) + } +} + /// An export's rows with every stream-wide counter zero, so one [`Log`] reads either edge. impl From for Snapshot { fn from(export: Export) -> Self { @@ -339,6 +495,14 @@ impl Log { mod test { use super::*; + /// `Class` keeps its `as` values with `Data` last, and an unknown name reads as data. + #[test] + fn class_keeps_its_discriminants() { + assert_eq!([Class::Data as u8, Class::Audio as u8, Class::Video as u8], [0, 1, 2]); + let class: Class = serde_json::from_str(r#""subtitle""#).unwrap(); + assert_eq!(class, Class::Data); + } + /// A stream whose count stops across a sample is logged once, with its silence, and again /// only after it has delivered in between; a stream that keeps counting never is, and /// neither reports frame sync it never lost. @@ -528,6 +692,81 @@ mod test { }); } + /// A report carries the counters beside `transport` and `renditions`, camelCase, keyed by + /// PID, `quiet` in milliseconds, and zero counters omitted; it reads back whole, and a class + /// a later version adds reads as data. + #[test] + fn a_report_flattens_beside_the_media_counters() { + let mut mpegts = Snapshot { + continuity_count_error: 2, + ..Default::default() + }; + mpegts.streams.insert( + 256, + Stream { + track: ".avc3".to_string(), + class: Class::Video, + units: 120, + quiet: Some(std::time::Duration::from_micros(40_400)), + continuity_count_error: 2, + ..Default::default() + }, + ); + let release = Release { + margin: Some(SignedDuration::from_millis(-12)), + offset: Some(SignedDuration::from_millis(3)), + drift: Some(1.5), + ..Default::default() + }; + let mut snapshot = hang::stats::Snapshot::::default(); + snapshot.ext.mpegts = Some(mpegts); + snapshot.ext.release = Some(release); + + let json = serde_json::to_value(&snapshot).unwrap(); + assert_eq!( + json, + serde_json::json!({ + "mpegts": { + "streams": { + "256": { "track": ".avc3", "class": "video", "units": 120, "quiet": 40, "continuityCountError": 2 } + }, + "continuityCountError": 2 + }, + "release": { "margin": -12, "offset": 3, "drift": 1.5 } + }) + ); + let mut read: hang::stats::Snapshot = serde_json::from_value(json).unwrap(); + let quiet = &mut read.ext.mpegts.as_mut().unwrap().streams.get_mut(&256).unwrap().quiet; + assert_eq!(*quiet, Some(std::time::Duration::from_millis(40))); + *quiet = snapshot.ext.mpegts.as_ref().unwrap().streams[&256].quiet; + assert_eq!(read, snapshot); + + let row: Stream = serde_json::from_value(serde_json::json!({ "class": "subtitle", "later": 1 })).unwrap(); + assert_eq!(row.class, Class::Data); + } + + /// A span short of a millisecond keeps its sign in a report, so a frame that missed its + /// deadline by less still reads as a miss. + #[test] + fn a_release_span_under_a_millisecond_keeps_its_sign() { + let release = Release { + margin: Some(SignedDuration::from_secs_f64(-0.000_3)), + offset: Some(SignedDuration::from_secs_f64(0.000_2)), + ..Default::default() + }; + assert!(release.margin.unwrap().is_negative()); + assert_eq!( + release.margin.unwrap().unsigned_abs(), + std::time::Duration::from_micros(300) + ); + assert_eq!( + serde_json::to_value(&release).unwrap(), + serde_json::json!({ "margin": -1, "offset": 1 }) + ); + assert_eq!(SignedDuration::from_secs_f64(-0.012).as_millis(), -12); + assert_eq!(SignedDuration::default().as_millis(), 0); + } + /// Damage is reported per PID when its cumulative count moves, including the final sample. #[test] #[tracing_test::traced_test] diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 7d6d245b5a..c13400e9e1 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -162,11 +162,18 @@ pub(crate) struct Buffer { trimmed: u64, /// Steps at which the source's clock ran further off ours than the clock may follow. out_of_tolerance: u64, + /// The least time any frame arrived before its deadline in the last whole [`MARGIN`], in + /// seconds: negative for a frame that missed it. + margin: Option, + /// The [`MARGIN`] under way: when its first frame was read, and the least time to spare so far. + spare: Option<(Instant, f64)>, } /// How often the clock is steered, in our time: long enough that the floor covers a /// group's worth of frames sent ahead by different amounts. const STEER: Duration = Duration::from_secs(2); +/// How long [`Buffer::margin`] takes the least of, in our time: a report's interval. +const MARGIN: Duration = Duration::from_secs(1); /// How far back the floors are kept, in decode time: a spell of queueing shorter than half /// of it moves neither the rate nor the slack. const RATE_WINDOW: f64 = 600.0; @@ -258,6 +265,9 @@ struct Steer { steered: f64, /// The source's clock rate off ours, as last measured: positive runs fast. source: Option, + /// How far the slack of the track sent latest sat off the delay, in seconds, as last + /// measured: the gap the clock pulls shut. + offset: Option, /// What is left of the gap between the slack and the delay that the rate's first /// measurement found, in seconds: what anchoring on frames that reached us late or early /// left, which the rate did not cost. @@ -272,6 +282,7 @@ impl Default for Steer { tracks: BTreeMap::new(), steered: 0.0, source: None, + offset: None, startup: None, } } @@ -350,6 +361,8 @@ impl Buffer { dropped: 0, trimmed: 0, out_of_tolerance: 0, + margin: None, + spare: None, } } @@ -544,6 +557,9 @@ impl Buffer { /// group its source cut short) takes everything the track holds with it, so its output starts on a /// whole group at its next sync frame rather than on a group's cut-off start. fn queue(&mut self, key: K, arrival: Arrival, generation: u64, deadline: Option, stale: bool) -> Push { + if let Some(deadline) = deadline.filter(|_| !stale && !self.delay.is_zero()) { + self.spare(arrival.read, since(arrival.arrived, deadline) as f64 / 1e9); + } let track = self.tracks.entry(key).or_insert_with(|| Track { queue: VecDeque::new(), restart: arrival.restart, @@ -600,6 +616,16 @@ impl Buffer { push } + /// Account for a frame read at `read` that arrived `spare` seconds before its deadline. + fn spare(&mut self, read: Instant, spare: f64) { + let window = self.spare.get_or_insert((read, spare)); + window.1 = window.1.min(spare); + if read.saturating_duration_since(window.0) >= MARGIN { + self.margin = Some(window.1); + self.spare = None; + } + } + /// Whether a frame arriving at `arrived` and due at `deadline` is on the clock: not late, /// and held no more than a delay past both its arrival and everything already queued. fn lands(&self, deadline: Option, arrived: Instant) -> bool { @@ -721,6 +747,7 @@ impl Buffer { }); let gap = at.clamp(low, high) + self.steer.steered; self.steer.source = Some(rate); + self.steer.offset = Some(gap); // The clock pulls the startup gap shut, so it shrinks with the gap and never reopens. let startup = self.steer.startup.get_or_insert(gap); *startup = startup.clamp(gap.min(0.0), gap.max(0.0)); @@ -962,6 +989,21 @@ impl Buffer { pub fn out_of_tolerance(&self) -> u64 { self.out_of_tolerance } + + /// The least time any frame arrived before its deadline in the last whole second of ours, + /// in seconds: negative for a frame that missed it. `None` until a second has passed with + /// a delay. + pub fn margin(&self) -> Option { + self.margin + } + + /// How far the clock sits off the source's plus the delay, in seconds, as last measured: + /// positive holds the track sent latest longer than the delay, negative shorter. The + /// clock pulls it shut over [`RESPONSE`]. `None` until the clock has measured the source's + /// rate, as [`Self::drift`]. + pub fn offset(&self) -> Option { + self.steer.offset + } } /// `nanos` after `at`, or before it if negative, if an instant holds it. @@ -1130,6 +1172,39 @@ mod tests { assert!(buffer.is_empty()); } + /// The margin is the least any frame had to spare over a whole second, a frame that missed + /// its deadline included, and holds until the next second has passed. + #[tokio::test(start_paused = true)] + async fn the_margin_is_the_least_spare_over_a_second() { + let start = Instant::now(); + let at = |ms: u64| start + Duration::from_millis(ms); + let mut buffer = Buffer::new(DELAY).replay(); + assert_eq!(buffer.push(1, arrival(at(0), 0, "v0")).unwrap(), Push::Queued); + assert_eq!(buffer.push(1, arrival(at(30), 40, "v40")).unwrap(), Push::Queued); + // Due at 180 ms. + assert_eq!(buffer.push(1, arrival(at(200), 80, "v80")).unwrap(), Push::Late); + assert_eq!(buffer.margin(), None, "no second has passed"); + + assert_eq!( + buffer.push(1, arrival(at(1_000), 1_000, "v1000")).unwrap(), + Push::Queued + ); + let margin = buffer.margin().expect("a second has passed"); + assert!((margin + 0.020).abs() < 1e-6, "margin {margin} s"); + + assert_eq!( + buffer.push(1, arrival(at(1_500), 1_500, "v1500")).unwrap(), + Push::Queued + ); + assert!(buffer.margin().unwrap() < 0.0, "held until the next second has passed"); + assert_eq!( + buffer.push(1, arrival(at(2_500), 2_500, "v2500")).unwrap(), + Push::Queued + ); + let margin = buffer.margin().unwrap(); + assert!((margin - 0.100).abs() < 1e-6, "the next second was clean: {margin} s"); + } + /// Two buffers fed the same frames with different arrival skew, every frame inside /// its deadline, emit the same order. #[tokio::test(start_paused = true)] @@ -1513,6 +1588,9 @@ mod tests { out_of_tolerance: u64, /// How far off the delay the frame the run failed on would have been held, in seconds. off: f64, + /// The buffer's margin and offset at the end. + margin: Option, + offset: Option, } /// Feed `hours` of 2 fps frames from a source whose clock runs `ppm` slower than ours @@ -1532,6 +1610,8 @@ mod tests { clock_before: 0.0, out_of_tolerance: 0, off: 0.0, + margin: None, + offset: None, }; let mut last: Option<(f64, Timestamp)> = None; let mut stepped = false; @@ -1596,6 +1676,8 @@ mod tests { run.drift = buffer.drift(); run.clock = buffer.clock.map_or(0.0, |clock| clock.drift); run.out_of_tolerance = buffer.out_of_tolerance(); + run.margin = buffer.margin(); + run.offset = buffer.offset(); run } @@ -1610,6 +1692,14 @@ mod tests { let run = drift(ppm, 24.0, 16.0, |_| Duration::ZERO).await; assert!(run.err.is_none(), "{ppm} ppm: {:?}", run.err); assert!(run.worst <= bound, "{ppm} ppm: held {} s off the delay", run.worst); + // One track queued nowhere: every frame's spare is its slack, the delay and the offset. + let offset = run.offset.expect("an offset"); + assert!(offset.abs() <= bound, "{ppm} ppm: offset {offset} s"); + let margin = run.margin.expect("a margin"); + assert!( + (margin - 0.5 - offset).abs() < 0.001, + "{ppm} ppm: margin {margin} s, offset {offset} s" + ); let drift = run.drift.expect("a measurement") * 1e6; assert!((drift + ppm).abs() < 0.1, "{ppm} ppm: measured {drift} ppm"); assert!(