diff --git a/.github/workflows/ts.yml b/.github/workflows/ts.yml index 03e93c9a7f9..71d06cb737c 100644 --- a/.github/workflows/ts.yml +++ b/.github/workflows/ts.yml @@ -72,6 +72,15 @@ jobs: HARNESS_RUNS: ${{ runner.temp }}/moq-test-runs shell: bash -leo pipefail {0} + # MPEG-2 video the importer carries as whole PES: FFmpeg decodes the capture to the + # source's pictures, and the video passes the T-STD model. + - name: TS MPEG-2 video + if: ${{ !cancelled() }} + run: nix develop --command just test ts --mpeg2 + env: + MOQ_TEST_RUNS: ${{ runner.temp }}/moq-test-runs + shell: bash -leo pipefail {0} + # A KLV stream joins the program mid-stream and leaves it: the export announces each # layout at the next PMT version, and carries the stream only while a PMT lists it. - name: TS PMT revision diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 72466365b83..236cdd7c9af 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -94,8 +94,10 @@ For a source already in MPEG-TS, `ffmpeg -c copy -f mpegts` re-muxes and drops P or use [`import srt`](/bin/srt). MPEG-TS import carries H.264/H.265 and AAC/MP2/AC-3/E-AC-3/Opus, and passes -SCTE-35 and every other stream it does not decode (KLV, subtitles, DVB AC-3) -through as tracks. Video AUs can span PES, and a PES can contain several. Supply +SCTE-35, DSM-CC sections, and every other stream it does not decode (MPEG-2 +video, LATM AAC, KLV, subtitles, DVB AC-3) through as tracks. MPEG-2 video still +paces the program by its timestamps, and a subscriber joins it at a sequence +header. H.264/H.265 AUs can span PES, and a PES can contain several. Supply a PTS for each AU or declare a proportional POC clock: AVC needs `fixed_frame_rate_flag` and `temporal_poc_flag` in the TS timing descriptor plus fixed-rate SPS VUI; HEVC needs VUI's `vui_poc_proportional_to_timing_flag`. Missing timing @@ -103,7 +105,10 @@ ends the import instead of repeating a timestamp. Private and undecoded PES go t (PCR); one before the first PCR is dropped. A stream it cannot carry unaltered is refused with a warning: any in a program with no PCR, and one whose PES carry an ESCR, pack header or TREF, are scrambled, leave a non-video PES unbounded, -or have a malformed header. +or have a malformed header, as is `0x86` without a SCTE-35 registration, which +may be DTS audio. A PID no table claims, such as conditional access or ATSC +PSIP, is dropped with one warning and counted in the +[stats](/concept/stats#mpeg-ts-sections); `--passthrough` keeps them. SCTE-35 cues and private PES share the media clock, or anchor on the first PCR when the PMT declares only data. Audio-only and data-only programs round-trip without video; a data-only program still requires PCR. A program declaring diff --git a/doc/concept/stats.md b/doc/concept/stats.md index e94a8e798b5..9c03f08540c 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -55,7 +55,17 @@ 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`. +`ptsError`. A TS-scrambled packet on a carried stream counts as `refused`. + +Every PID but the null PID either takes a route or is counted. A listed stream +the import cannot carry, such as `0x86` without a SCTE-35 registration, gets a +row with an empty `track` and one `refused`. Once every program's PMT is in, or +has been missing for 0.5 s of program clock, a PID that no PAT, PMT, or +captured SI table claims, such as the CAT, conditional access, or ATSC PSIP, +gets a `data` row with an empty `track` and `refused` counting its packets, +with one warning when it starts. An import of one `--program` of a +multi-program stream cannot tell another program's PIDs from these, so it +counts none of them. The stream-wide counters count ETSI TR 101 290 checks at that standard's fixed limits: diff --git a/drafts/draft-lcurley-moq-mpegts.md b/drafts/draft-lcurley-moq-mpegts.md index 450f7b1c76a..f070598e9c2 100644 --- a/drafts/draft-lcurley-moq-mpegts.md +++ b/drafts/draft-lcurley-moq-mpegts.md @@ -236,7 +236,7 @@ type Verbatim = { } ~~~ -`streamType` is the PMT `stream_type` to re-announce: 0x86 for SCTE-35 {{scte35}}, 0x05 for private sections. +`streamType` is the PMT `stream_type` to re-announce: 0x86 for SCTE-35 {{scte35}}, 0x05 for private sections, 0x0A to 0x0D for ISO/IEC 13818-6 DSM-CC sections. `framing` is required, and `"section"` is its only value: each frame is one complete section, header and CRC included, timestamped with its placement: the program clock at its first section byte, on the same timeline as a whole PES below. A consumer MUST refuse a catalog whose `verbatim` record names any other `framing` or none, rather than guess; in particular it MUST NOT read the `"pes"` framing of draft -00, or a record without `framing`, as sections. @@ -251,11 +251,12 @@ type PesPacket = { } ~~~ -`streamType` is the PMT `stream_type` to re-announce, as for `Verbatim`: 0x06 for private data such as asynchronous KLV, teletext, DVB subtitles and DVB AC-3, 0x15 for synchronous metadata. +`streamType` is the PMT `stream_type` to re-announce, as for `Verbatim`: 0x01 and 0x02 for MPEG-1 and MPEG-2 video, 0x11 for LATM AAC, 0x06 for private data such as asynchronous KLV, teletext, DVB subtitles and DVB AC-3, 0x15 for synchronous metadata. Each frame MUST be exactly one PES packet, from its start code through its payload, with a `PES_packet_length` that matches the frame, or zero only where {{mpeg2}} lets a video stream leave it unbounded. The `stream_id`, the PTS and DTS when present, and every other header field ride in the frame, with the PTS and DTS on the catalog clock; a PES without a PTS carries none. The frame's timestamp is the packet's placement: the program clock at its first byte, on the same timeline. -Each frame MUST be a keyframe in its own group, as for `verbatim`. +Each frame MUST be a keyframe in its own group, as for `verbatim`, except on MPEG-1 and MPEG-2 video (0x01 and 0x02), whose pictures depend on earlier ones: there a PES in which a `sequence_header_code` (0x000001B3) begins, including one whose last bytes the next PES completes, MUST be a keyframe opening a group, and the PES up to the next one MUST join it. +A publisher drops the video PES before the first sequence header, and after a lost one until the next, rather than publish a picture without what it references. A consumer MUST refuse a frame that is not one well-formed PES packet. The record is a member beside `verbatim` rather than a `framing` value, so a consumer that predates it ignores the track instead of refusing the catalog. @@ -370,7 +371,7 @@ This document defines no packaging value; {{msfts}} defines `mpeg2ts` for the pa A consumer rebuilding a transport stream: - MUST place each track on its recorded `pid`, and MUST assign an unused PID to a track with none. -- MUST put the PMT on `program.pmtPid` and build the PAT and PMT from `program`. +- MUST put the PMT on `program.pmtPid` and build the PAT and PMT from `program`. When it carries a NIT (the `si` entry for `table_id` 0x40 on PID 0x0010), the PAT MUST also list program 0 with that `network_PID`, so a receiver can find it. - MUST choose a PCR PID and emit PCRs on it at the program clock's repetition interval. With no media stream it MUST use an unused PID with adaptation-only PCR packets and no PMT elementary-stream entry. It MUST NOT refuse a program for lacking video. - MUST follow the catalog as it changes: when a track joins or leaves `tracks` or `programDescriptors` changes, it MUST emit a PMT at the next `version_number` (modulo 32) before any packet of a stream it adds, and MUST NOT emit a stream's packets after a PMT that no longer lists it. Each version it emits MUST differ from the one before, and no older version may follow a newer one. The PCR PID MUST NOT move until the stream it rides ends. - MUST re-emit each track's `descriptors` as its ES-level descriptors, and `programDescriptors` as the PMT's `program_info`. @@ -455,6 +456,8 @@ A broadcast demultiplexed from a DVB transport stream: video and audio described - Section-framed `verbatim` timestamps use PCR-derived placement at the first section byte, as whole PES do, rather than a video presentation time. - Added the `pesPacket` record: each frame one whole PES packet, timestamped with its placement. A consumer that predates it drops the track rather than refusing the catalog. - `verbatim` carries sections only, breaking -00: `framing` is required and `"section"` its only value, and `streamId` is gone. A consumer refuses a -00 catalog with `"pes"` framing or a `verbatim` record without `framing`. A publisher carries an undecoded PES stream as `pesPacket`, and refuses one it cannot carry unaltered. +- MPEG-1/2 video and LATM AAC are named among `pesPacket` streams, and the DSM-CC types 0x0A to 0x0D among section-framed `verbatim` ones. A rebuilt PAT lists the network entry when the NIT is carried. +- MPEG-1/2 video `pesPacket` frames group from one sequence header to the next, rather than one PES per group. - A flagged PCR jump on an `m2ts` track follows hang's discontinuity marker rather than a skipped group sequence. - An `m2ts` track unwraps PCRs from its history, clamps extrapolated object times at a forward discontinuity, and cuts a fallback group after one second. `randomAccess` covers every published group; `muxRate` follows the measured source rate. - A `tracks` entry is never changed in place: a stream the PMT revises continues under a new name, and one it drops leaves the section. `programDescriptors` follows the PMT; `program`'s `transportStreamId` and `programNumber` do not change. diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index 9d51ab2861f..a4d89f9cd17 100644 --- a/quest/m1/ts-reliability/README.md +++ b/quest/m1/ts-reliability/README.md @@ -67,9 +67,9 @@ refused means an error or a counted drop. | Private and unknown PES types, LATM (0x11), metadata (0x15) | Preserved: `pesPacket`, PTS and DTS rebased | landed | | PES clock fields (ESCR, pack header, TREF), PES scrambling | Refused, counted in `refused` | landed | | Private sections (0x05), SCTE-35 (0x86 + `CUEI`) | Preserved: `verbatim` section, `pts_adjustment` rebased | landed | -| MPEG-1/2 video (0x01/0x02) | Preserved: `pesPacket`, keeping its clock role; today clock only, payload discarded | [routes](/quest/m1/ts-reliability/stream-routes.md) | -| DSM-CC types (0x0A to 0x0D), 0x86 without `CUEI` | Today PES-parsed into damage, or dropped uncounted | [routes](/quest/m1/ts-reliability/stream-routes.md) | -| CAT, EMM/ECM, ATSC PSIP, other SI PIDs, unlisted PIDs, TS scrambling | Outside the profile, raw lane; counted when the import reads every program. Today dropped uncounted | [routes](/quest/m1/ts-reliability/stream-routes.md) | +| MPEG-1/2 video (0x01/0x02) | Preserved: `pesPacket` grouped from one sequence header to the next, keeping its clock role | landed | +| DSM-CC types (0x0A to 0x0D), 0x86 without `CUEI` | DSM-CC preserved as a `verbatim` section; 0x86 without `CUEI` refused, counted in `refused` | landed | +| CAT, EMM/ECM, ATSC PSIP, other SI PIDs, unlisted PIDs, TS scrambling | Outside the profile, raw lane; counted in `refused`, unclaimed PIDs when the import reads every program | landed | | DVB SI on 0x0010, 0x0011, 0x0012, 0x0014 | Preserved: snapshot track per PID and `table_id` | landed | | Program identity, program and ES descriptors | Preserved: descriptors follow each PMT revision; an identity change refuses | landed | | PMT add, remove, retype; PID reuse | Normalized without a new epoch: a changed or reused PID ends its track and continues under a fresh name, and the export rebuilds the PMT at the next `version_number` | landed | @@ -79,7 +79,7 @@ refused means an error or a counted drop. | Mux rate, null stuffing | Normalized: re-padded to `muxRate` while the source is CBR | landed | | PSI and SI repetition | Normalized: PAT/PMT every 500 ms and at keyframes, SI at DVB maxima | landed | | Media PES headers, TS packet layer | Normalized: rebuilt by export; the raw lane keeps them | landed | -| PAT network PID entry | Preserved: listed on 0x0010 when a NIT is carried; today dropped | [routes](/quest/m1/ts-reliability/stream-routes.md) | +| PAT network PID entry | Preserved: listed on 0x0010 when a NIT is carried | landed | Decided in the 2026-10-22 root-cause audit: the rows above kept getting patched one at a time because two foundations were missing. Decode time is @@ -99,7 +99,6 @@ catalog linger (#5305). ## Required - [PAT changes](/quest/m1/ts-reliability/pat-lifecycle.md) - a revised PAT never leaves a broadcast meaning another program -- [Stream routes](/quest/m1/ts-reliability/stream-routes.md) - every PID gets a declared route, and what is not carried is counted - [TS workflow passes](/quest/m1/ts-reliability/validation-green.md) - the scheduled TS lanes pass on main again, with no lane loosened ## Related diff --git a/quest/m1/ts-reliability/pat-lifecycle.md b/quest/m1/ts-reliability/pat-lifecycle.md index 37d58b1e598..c07cafd64b2 100644 --- a/quest/m1/ts-reliability/pat-lifecycle.md +++ b/quest/m1/ts-reliability/pat-lifecycle.md @@ -13,7 +13,9 @@ What the code does today: - A PAT naming another PMT PID for the program adds a reassembler but never drops the old one, and `program.pmtPid` keeps the first. -- `Programs` ignores a program a later PAT adds, without a warning. +- `Programs` ignores a program a later PAT adds, without a warning. Its + PIDs hold back the unclaimed-PID count for 0.5 s of program clock, then + count as unclaimed in `ts::stats`, since no program reads them. - Under `--program all`, the shared parser hands a later PAT that removes one program to every program's `Import`; the removed program's `handle_pat` errors, `Programs::decode` propagates it, and the whole multiplex ingest diff --git a/quest/m1/ts-reliability/stream-routes.md b/quest/m1/ts-reliability/stream-routes.md deleted file mode 100644 index b9979ab89c8..00000000000 --- a/quest/m1/ts-reliability/stream-routes.md +++ /dev/null @@ -1,108 +0,0 @@ -# [M] Every TS PID gets a declared route - -## Goal - -TS import gives every PID of the programs it imports, other than the null -PID 0x1FFF, one declared route: decoded, whole-PES `pesPacket`, `verbatim` -section, captured PSI/SI, or not carried. Null packets are stuffing, rebuilt -by export, and stay uncounted. A PID the semantic lane does not carry shows -in `ts::stats` with a count and one warning, so an operator sees what to -move to the raw lane. Multiplex-wide PIDs no program claims are counted when -the import reads every program (an SPTS, or `--program all`); a -`--program N` import of an MPTS reports only its own program. A known -`stream_type` never implies carriage it does not get. - -## Plan - -Found 2026-10-20 in the fidelity audit (`Import::ensure_stream`, -`Import::decode`): - -- **MPEG-1/2 video (0x01/0x02)** is `Stream::Clock`: its PTS anchors the - timebase and its payload is discarded, so TS to Hang to TS loses the - program's video without an error, while an unknown `stream_type` survives - as `pesPacket`. Decided 2026-10-20 by the maintainer: carry it as an - opaque `pesPacket` and keep its clock role, so the round trip keeps the - video and a Hang player still sees no rendition it cannot decode. The - draft already lets a video PES leave `PES_packet_length` unbounded. - Rejected: refusing it loudly and leaving the video to the raw lane, which - drops a stream the semantic lane can carry unchanged. -- **Section-framed types outside 0x05 and CUEI-marked 0x86** are PES-parsed - today. The ISO 13818-6 DSM-CC types (0x0A to 0x0D) ride sections, so every - unit fails the start code and counts as `damaged`. Target: route each - DSM-CC type 13818-6 confirms as section-framed through `ensure_section`. -- **0x86 without CUEI** drops with a warning and no count. Count it in - `refused`, since it is ambiguous by design. -- **LATM/LOAS AAC (0x11)** already rides `pesPacket` with the `mpegts` - section, though the `Import` docs and the `ensure_stream` comment say it is - dropped. Fix the docs to match. -- **PIDs outside the PMT and the captured SI** (CAT on 0x0001 and the - EMM/ECM PIDs it or a CA descriptor names, TSDT, ATSC PSIP on 0x1FFB and the - PIDs its MGT signals, the DVB RST/DIT/SIT PIDs, and any PID no table lists) - drop silently. Count each in a `ts::stats` row. ATSC PSIP and conditional access are outside the semantic - profile; the guide points them at the raw lane. -- **The PAT's network PID entry** (`program_number` 0) is filtered out on - import, and `Export::build_psi` writes a PAT with only the program, while - the NIT itself passes through on 0x0010 as a snapshot track. Export lists - program 0 on 0x0010 whenever the catalog carries a NIT snapshot, so the - rebuilt PAT points a receiver at the NIT it carries. -- **TS-scrambled packets** on an elementary PID become PES parse damage. - Count them in `refused` instead, since the semantic lane cannot carry them - and the raw lane can. - -A PID counts as dropped only once every program the current PAT lists has -delivered its PMT: packets on a PID before then are acquisition, as -`Import::decode` already treats them, so media that precedes its PMT on a -mid-stream join never shows as refused, including one program's media -arriving before its PMT while another program's PMT is already known. A PAT -that adds a program suspends the count for PIDs no PMT claims until that -program's PMT arrives. A listed PMT that does not arrive within the 0.5 s of -program clock TR 101 290 allows (`PMT_error_2`, already counted) stops -holding the gate, so a stale or corrupt program never disables the count. - -Under `--program all`, one parser reads the multiplex and `Programs` routes -each PID only to the programs whose PAT entry or PMT claims it -(`Programs::readers`), while `Programs::stats` merges their rows by PID, so a -per-program drop row would shadow the program that carries the PID. `Programs` owns the -unclaimed-PID rows instead: a PID is dropped only when no program's PMT and -no captured SI claims it, counted once at the multiplex rather than per -program. A `--program N` import of an MPTS reassembles only its own PMT, so -it cannot tell another program's PIDs from unclaimed ones; it counts only its -own program's PIDs and leaves multiplex-wide accounting to `--program all`. -When the PAT lists only the selected program, it counts as above. Chosen -over passively acquiring every PMT, which adds a reassembler per program to -an import that serves one. -Classifying a PID reads `Programs::readers`, the PID-indexed table of every -claimed PID kept as PMTs arrive: a PID no program reads is unclaimed. Extend -`rs/moq-mux/benches/ts_programs.rs` to sweep PID count beside program count -and bitrate, so any slope the accounting adds shows. - -A dropped PID reuses `stats::Stream`, with no new field: a row keyed by its -PID in `Snapshot::streams`, an empty `track`, `Class::Data`, and `refused` -counting the PID's dropped packets, with one warning on the first. Widen the -`Snapshot::streams` and `refused` docs to say so, and report that as a -public API change in meaning. A PID keeps one row for its whole life, -whoever owns it: a PMT that claims a dropped PID, releases a carried one, or -claims it again moves that row between the carrying `Import` and the -dropped-PID owner (`Programs` under `--program all`) with every cumulative -counter intact, since counters never go backward. `track` and `class` follow -the current owner, empty and `Data` while dropped, and two rows never -coexist under one key. Coordinate the serialized shape with -the [MPEG-TS sections](/doc/concept/stats.md#mpeg-ts-sections) of a media report. - -Rust: `rs/moq-mux` TS import (and export for MPEG-2 video `pesPacket` -packetization). JS: none. Draft -([mpegts](/drafts/draft-lcurley-moq-mpegts.md)): list MPEG-2 video among the -`pesPacket` examples and the DSM-CC types among the section-framed ones. - -Tests: generated `fixture::Mux` inputs with MPEG-2 video, each DSM-CC type, -0x86 without CUEI, LATM, a PAT network entry with a NIT, a CAT plus -scrambled ES, a PSIP base PID, an unlisted PID, media before its first PMT, -a two-program MPTS under `--program all` whose second program's media -precedes its PMT, a PAT listing a program whose PMT never arrives, one -PID's full lifecycle (unlisted, added by a PMT revision, removed while its -packets continue, then added again) with its row's counters checked at each -step, and the same MPTS under `--program 1` with no rows for -program 2's PIDs, asserting the catalog, the -stats rows (the unlisted PID's row asserted field by field), and an export -round trip of the carried streams. The MPEG-2 video arm also runs the TSDuck -round trip and an FFmpeg decode of the export in `test/ts`. diff --git a/rs/justfile b/rs/justfile index 78bb68fdfd9..6826419d200 100644 --- a/rs/justfile +++ b/rs/justfile @@ -502,6 +502,11 @@ bench-latency: bench-ts-schedule *args: bash sh/local.sh cargo bench --locked -p moq-mux --bench ts_schedule -- "$@" +# Ingest an MPTS across program count, bitrate and stray PIDs, the strays against null packets that skip the accounting. +[positional-arguments] +bench-ts-programs *args: + cargo bench --locked -p moq-mux --bench ts_programs -- "$@" + # Sweep container export tracks and queued frames independently. [positional-arguments] bench-export *args: diff --git a/rs/moq-mux/benches/ts_programs.rs b/rs/moq-mux/benches/ts_programs.rs index 138d738ce9a..5c9e3a7763f 100644 --- a/rs/moq-mux/benches/ts_programs.rs +++ b/rs/moq-mux/benches/ts_programs.rs @@ -1,7 +1,8 @@ //! `--program all` TS ingest across program count and per-program bitrate, over one second of //! a generated multiplex. Throughput is the whole input, so a parse that repeats per program //! shows as a falling rate with program count. A PAT that also names programs no PMT -//! describes shows what a program costs before it starts. +//! describes shows what a program costs before it starts. Stray PIDs no table lists sweep the +//! unclaimed-PID accounting, against the same packets sent as null stuffing, which skips it. use std::time::Duration; @@ -58,6 +59,13 @@ fn undescribed_pid(program: u16) -> u16 { 0x1000 + program } +/// A PID no table lists, clear of every other. +fn stray_pid(stray: u16) -> u16 { + 0x1800 + stray +} + +const NULL_PID: u16 = 0x1fff; + #[derive(Default)] struct Mux { out: Vec, @@ -131,10 +139,25 @@ impl Mux { } } +/// What the multiplex carries besides its described programs. +#[derive(Clone, Copy)] +struct Extra { + /// Programs the PAT names whose PMT never arrives. + undescribed: u16, + /// PIDs no table lists, one packet each per frame. + strays: u16, + /// Send the strays' packets as null stuffing instead. + null: bool, +} + /// One second of `programs` H.264 programs at `bitrate` bits per second each, interleaved -/// frame by frame, each program carrying its PCR on its video PID. The PAT also names -/// `undescribed` programs whose PMT never arrives. -fn multiplex(programs: u16, undescribed: u16, bitrate: u64) -> Vec { +/// frame by frame, each program carrying its PCR on its video PID, beside `extra`. +fn multiplex(programs: u16, extra: Extra, bitrate: u64) -> Vec { + let Extra { + undescribed, + strays, + null, + } = extra; let packets = (bitrate / FPS / 8).div_ceil(188) as usize; let mut mux = Mux::default(); let entries: Vec = (1..=programs) @@ -176,6 +199,12 @@ fn multiplex(programs: u16, undescribed: u16, bitrate: u64) -> Vec { for program in 1..=programs { mux.frame(video_pid(program), packets, pts * 300, pts, frame % GOP == 0); } + for stray in 0..strays { + let pid = if null { NULL_PID } else { stray_pid(stray) }; + mux.header(pid, false, false); + let end = mux.out.len() + 184; + mux.out.resize(end, 0xff); + } } mux.out } @@ -201,34 +230,59 @@ fn programs(c: &mut Criterion) { group.measurement_time(Duration::from_secs(2)); group.warm_up_time(Duration::from_millis(300)); group.sample_size(10); + let mut cases = Vec::new(); for programs in [1u16, 4, 16] { for undescribed in [0u16, 1000] { for mbps in [1u64, 4] { - let input = multiplex(programs, undescribed, mbps * 1_000_000); - let (stats, _driver) = ingest(programs, &input); - assert!( - stats.streams.len() == usize::from(programs) && stats.streams.values().all(|s| s.units == FPS), - "every program delivers every frame" - ); - group.throughput(Throughput::Bytes(input.len() as u64)); + let extra = Extra { + undescribed, + strays: 0, + null: false, + }; let name = match undescribed { 0 => format!("{programs}"), _ => format!("{programs}+{undescribed}"), }; - group.bench_function(BenchmarkId::new(name, format!("{mbps}mbps")), |b| { - b.iter_batched( - setup, - |(mut programs, driver)| { - for chunk in input.chunks(188 * 7) { - programs.decode(chunk).unwrap(); - } - (programs, driver) - }, - BatchSize::PerIteration, - ) - }); + cases.push((programs, extra, mbps, name)); } } + for strays in [16u16, 256] { + for (lane, null) in [("null", true), ("stray", false)] { + let extra = Extra { + undescribed: 0, + strays, + null, + }; + cases.push((programs, extra, 1, format!("{programs}-{lane}{strays}"))); + } + } + } + for (programs, extra, mbps, name) in cases { + let input = multiplex(programs, extra, mbps * 1_000_000); + let (stats, _driver) = ingest(programs, &input); + let (media, strays): (Vec<_>, Vec<_>) = stats.streams.values().partition(|s| !s.track.is_empty()); + assert!( + media.len() == usize::from(programs) && media.iter().all(|s| s.units == FPS), + "every program delivers every frame" + ); + let counted = if extra.null { 0 } else { usize::from(extra.strays) }; + assert!( + strays.len() == counted && strays.iter().all(|s| s.refused == FPS), + "every stray packet is counted" + ); + group.throughput(Throughput::Bytes(input.len() as u64)); + group.bench_function(BenchmarkId::new(name, format!("{mbps}mbps")), |b| { + b.iter_batched( + setup, + |(mut programs, driver)| { + for chunk in input.chunks(188 * 7) { + programs.decode(chunk).unwrap(); + } + (programs, driver) + }, + BatchSize::PerIteration, + ) + }); } group.finish(); } diff --git a/rs/moq-mux/src/container/ts/catalog.rs b/rs/moq-mux/src/container/ts/catalog.rs index 63cfff2eab5..b72691befe7 100644 --- a/rs/moq-mux/src/container/ts/catalog.rs +++ b/rs/moq-mux/src/container/ts/catalog.rs @@ -367,7 +367,8 @@ impl Verbatim { /// Each frame is one complete PES packet, start code through payload, with its PTS, DTS /// and every other header field already on the catalog clock. Its timestamp is the /// packet's placement: the program clock at its first byte, which export sends it at. -/// A member beside [`Verbatim`] rather than a [`Framing`], so a consumer that predates it +/// Each frame is a group of its own, except MPEG-1/2 video's, grouped from one sequence header +/// to the next. A member beside [`Verbatim`] rather than a [`Framing`], so a consumer that predates it /// ignores the track instead of refusing the catalog. #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] #[serde(rename_all = "camelCase")] diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 6f557c3dca7..71767f63e21 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -706,6 +706,9 @@ impl Track { if video && frame.keyframe { self.keyframe = Some(frame.timestamp); } + // Video, MPEG-1/2 carried whole among it, resumes only at a keyframe, since its other + // pictures reference earlier ones. + let sync = frame.keyframe || self.kind.class() != super::stats::Class::Video; // Replaying anchors on the first frame, and the tracks arrive together rather than each // sent ahead by its send-ahead: judged as arriving that much earlier, as a live source // sends it, every track's first frames are on the clock whichever anchors it. @@ -724,7 +727,7 @@ impl Track { restart, skip, cut, - sync: frame.keyframe || !video, + sync, leading, item: Queued { frame, @@ -912,7 +915,11 @@ impl Kind { /// alike at both edges. fn class(&self) -> super::stats::Class { match self { - Kind::Video(_) => super::stats::Class::Video, + // MPEG-1/2 video rides whole, but paces the program as decoded video does. + Kind::Video(_) + | Kind::PesPacket { + stream_type: 0x01 | 0x02, + } => super::stats::Class::Video, Kind::Aac { .. } | Kind::Opus { .. } | Kind::Mp2 { .. } | Kind::Ac3 | Kind::Eac3 => { super::stats::Class::Audio } @@ -921,10 +928,14 @@ impl Kind { } } -/// The program tables: the PAT, fixed for the program's life, and the PMT, which lists the -/// streams on the air. +/// The program tables: the PAT, fixed for the program's life but for its network entry, and the +/// PMT, which lists the streams on the air. struct Psi { pat: Bytes, + /// The source's `transport_stream_id`, so the rebuilt PAT stays consistent with the carried SI. + transport_stream_id: u16, + /// Whether the PAT lists the network entry, pointing at the NIT the export carries. + network: bool, /// What the PMT says now. pmt: psi::Pmt, /// PID the PMT rides on: the source's original (preserved from the service @@ -952,6 +963,35 @@ impl Psi { } } +/// The DVB NIT's PID (ETSI EN 300 468 5.1.3), the network PID the PAT's program 0 names. +const NIT_PID: u16 = 0x0010; +/// The `table_id` of the NIT describing the delivering network. +const NIT_ACTUAL: u8 = 0x40; + +/// The PAT for the exported program at `version`, listing the network PID as program 0 first when +/// the export carries the NIT (ISO/IEC 13818-1 2.4.4.3), which a receiver otherwise cannot find. +fn encode_pat( + transport_stream_id: u16, + program_number: u16, + pmt_pid: u16, + network: bool, + version: VersionNumber, +) -> anyhow::Result { + let network = network.then(|| ProgramAssociation { + program_num: 0, + program_map_pid: Pid::new(NIT_PID).expect("a valid PID"), + }); + let program = ProgramAssociation { + program_num: program_number, + program_map_pid: Pid::new(pmt_pid)?, + }; + psi::encode_pat(&Pat { + transport_stream_id, + version_number: version, + table: network.into_iter().chain([program]).collect(), + }) +} + /// Per-frame PES descriptor (everything but the payload bytes). struct PesUnit { pid: u16, @@ -2304,14 +2344,8 @@ where // carried SI; synthesize a minimal identity otherwise. let pmt_pid = self.pmt_pid(); let transport_stream_id = self.program.as_ref().map(|s| s.transport_stream_id).unwrap_or(1); - let pat = Pat { - transport_stream_id, - version_number: self.version, - table: vec![ProgramAssociation { - program_num: pmt.program_number, - program_map_pid: Pid::new(pmt_pid)?, - }], - }; + let network = self.carries_nit(); + let pat = encode_pat(transport_stream_id, pmt.program_number, pmt_pid, network, self.version)?; self.schedule.set_tables(pmt_pid); self.schedule.set_clock(pcr_pid); @@ -2319,7 +2353,9 @@ where self.liveness.register(track.pid); } self.psi = Some(Psi { - pat: super::psi::encode_pat(&pat)?, + pat, + transport_stream_id, + network, pmt, pmt_pid, pushed: None, @@ -2330,6 +2366,11 @@ where Ok(()) } + /// Whether the catalog carries a NIT on its PID, which the PAT then points a receiver at. + fn carries_nit(&self) -> bool { + self.si.contains_key(&(NIT_PID, NIT_ACTUAL)) + } + /// The PID to carry the clock: the first video track's, else the first audio track's, else /// one used by nothing else, since sparse data cannot carry the clock's cadence. fn pick_clock(&self) -> anyhow::Result { @@ -2536,9 +2577,12 @@ where if !std::mem::take(&mut self.psi_due) { return Ok(()); } + // A NIT acquired or lost after the tables went out revises the PAT's network entry, at + // the next version beside the PMT, so the PAT's version never runs ahead of the PMT's. + let network = self.carries_nit(); let psi = self.psi.as_mut().context("PSI not built")?; let version = match &psi.pushed { - Some((version, pmt)) if *pmt == psi.pmt => *version, + Some((version, pmt)) if *pmt == psi.pmt && psi.network == network => *version, Some((version, _)) => { let mut version = *version; version.increment(); @@ -2546,6 +2590,16 @@ where } None => self.version, }; + if psi.network != network { + psi.pat = encode_pat( + psi.transport_stream_id, + psi.pmt.program_number, + psi.pmt_pid, + network, + version, + )?; + psi.network = network; + } let added: Vec = psi .pmt .streams @@ -2712,10 +2766,11 @@ where }; let pid = track.pid; let place = frame.timestamp.as_nanos(); + // Its first access unit decodes at the DTS, which a reordered picture carries ahead of its PTS. let decode = pes .as_ref() - .and_then(|pes| pes.pts) - .map(|pts| unwrap_pts(pts, frame.timestamp)); + .and_then(|pes| pes.dts.or(pes.pts)) + .map(|ticks| unwrap_pts(ticks, frame.timestamp)); let dolby = Dolby::described(&track.descriptors) .filter(|_| matches!(track.kind, Kind::PesPacket { stream_type: 0x06 })); let frames = match (&pes, decode, dolby) { @@ -3289,6 +3344,8 @@ const PES_DTS_LEN: usize = 5; struct PesPacket { /// Its PTS in 90 kHz ticks, as the 33-bit field carries it. pts: Option, + /// Its DTS, likewise, where it differs from the PTS. + dts: Option, /// Where its payload begins, past the header. payload: usize, } @@ -3312,6 +3369,7 @@ impl PesPacket { } Ok(Self { pts: pes.pts, + dts: pes.dts, payload: pes.header.len(), }) } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index e85cd7097a3..c282176cd5a 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -102,7 +102,7 @@ async fn drain(consumer: moq_net::broadcast::Consumer) -> BytesMut { } /// `drain` for an exporter built with an explicit catalog extension. -async fn drain_with(exporter: Export) -> BytesMut +pub(super) async fn drain_with(exporter: Export) -> BytesMut where S::Ext: tscat::Catalog, { @@ -1079,6 +1079,97 @@ async fn export_trims_leading_pictures_after_a_late_drop() { assert_eq!(export.dropped(), 3, "the late picture and the two it orphaned"); } +/// MPEG-2 video carried whole resumes after a late picture at its next sequence header, not +/// at the next picture, which references the dropped one. +#[tokio::test(start_paused = true)] +async fn export_resumes_whole_mpeg2_video_at_a_sequence_header() { + 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().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut aac = AudioConfig::new(AAC { profile: 2 }, 48_000, 2); + aac.container = Container::Legacy; + let track = broadcast + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + catalog.modify().unwrap().audio.renditions.insert("audio".into(), aac); + let mut audio = Producer::new(track, HangContainer::Legacy(Kind::Audio)); + let track = broadcast + .create_track("video.ts", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let mut entry = tscat::Track::new(PLACED_PID); + entry.pes_packet = Some(tscat::PesPacket::new(0x02)); + catalog + .modify() + .unwrap() + .ext + .mpegts + .tracks + .insert("video.ts".into(), entry); + let mut video = Producer::new(track, HangContainer::Legacy(Kind::Data)); + let mut export = open_ts( + crate::source::announced(&consumer).await, + crate::catalog::CatalogFormat::Hang, + ) + .await + .unwrap() + .with_delay(Duration::from_millis(500)); + + // (arrival, PTS) in milliseconds, and whether a sequence header opens a group. The 120 ms + // picture stalls past its deadline; the two pictures after it still reference it. + let pictures = [ + (0, 0, true), + (40, 40, false), + (80, 80, false), + (1000, 120, false), + (2000, 2000, false), + (2040, 2040, false), + (2080, 2080, true), + (2120, 2120, false), + ]; + let start = tokio::time::Instant::now(); + let (mut out, mut sample) = (Vec::new(), 0); + for (index, (at, pts, sequence)) in pictures.into_iter().enumerate() { + // The audio arrives on time throughout, so only the video is late. + while sample * AUDIO_US <= at * 1_000 { + tokio::time::sleep_until(start + Duration::from_micros(sample * AUDIO_US)).await; + out.extend(poll_frames(&mut export)); + audio + .write(Frame { + timestamp: Timestamp::from_micros(sample * AUDIO_US).unwrap(), + duration: None, + payload: Bytes::from(vec![sample as u8; 180]), + keyframe: sample.is_multiple_of(AUDIO_GROUP), + }) + .unwrap(); + sample += 1; + } + tokio::time::sleep_until(start + Duration::from_millis(at)).await; + out.extend(poll_frames(&mut export)); + let mut payload = if sequence { vec![0, 0, 1, 0xb3] } else { Vec::new() }; + payload.extend_from_slice(&[0, 0, 1, 0x00, index as u8]); + video + .write(Frame { + timestamp: Timestamp::from_millis(pts).unwrap(), + duration: None, + keyframe: sequence, + payload: Bytes::from(pes_with(0xe0, 0x80, 0x80, &pes_clock(2, pts * 90), 0, &payload)), + }) + .unwrap(); + out.extend(poll_frames(&mut export)); + } + audio.finish().unwrap(); + video.finish().unwrap(); + out.extend(drain_frames(&mut export).await); + + // Each picture's last byte is its index. + let sent: Vec = placed_out(&out).iter().map(|pes| *pes.bytes.last().unwrap()).collect(); + assert_eq!(sent, [0, 1, 2, 6, 7]); +} + /// `(PTS, DTS)` of every video PES start, in transport (decode) order. Read raw, so a capture /// that starts before its PMT still counts. fn video_pes_timing(ts: &[u8]) -> Vec<(u64, u64)> { @@ -3319,15 +3410,19 @@ async fn service_layer_survives_roundtrip() { .await; assert_packet_aligned(&ts); - // The rebuilt PAT preserves the transport/service identity and PMT PID. + // The rebuilt PAT preserves the transport/service identity and PMT PID, and points at the + // carried NIT through the network entry. let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref())); let mut checked_pat = false; while let Some(packet) = reader.read_ts_packet().unwrap() { if let Some(TsPayload::Pat(pat)) = packet.payload { assert_eq!(pat.transport_stream_id, 1, "TSID preserved in the rebuilt PAT"); - assert_eq!(pat.table.len(), 1); - assert_eq!(pat.table[0].program_num, 1, "service number preserved"); - assert_eq!(pat.table[0].program_map_pid.as_u16(), 0x1000, "PMT PID preserved"); + let table: Vec<_> = pat + .table + .iter() + .map(|entry| (entry.program_num, entry.program_map_pid.as_u16())) + .collect(); + assert_eq!(table, [(0, 0x0010), (1, 0x1000)], "network entry, then the service"); checked_pat = true; break; } @@ -3352,6 +3447,14 @@ async fn service_layer_survives_roundtrip() { let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve()); import2.decode(&BytesMut::from(ts.as_ref())).unwrap(); import2.finish().unwrap(); + // Its row, if any, is the SI track's, never an unclaimed PID's. + if let Some(nit) = import2.stats().streams.get(&0x0010) { + assert_eq!( + (nit.track.as_str(), nit.refused), + (".si", 0), + "the network entry names captured SI, not a PID to count: {nit:?}" + ); + } let snapshot2 = catalog2.snapshot(); let program2 = snapshot2 diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 493e3ebeebe..f00ed568bdb 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -41,10 +41,11 @@ mod video; /// Demuxes an MPEG-TS byte stream into a MoQ broadcast. /// -/// Supports H.264 (stream type 0x1B), H.265 (0x24), ADTS AAC (0x0F), MP2 -/// (0x03/0x04), AC-3 (0x81), and E-AC-3 (0x87). LATM/LOAS AAC (0x11) is not -/// ADTS-framed and is dropped. Each codec stream is fed to its importer, which -/// manages the track, catalog config, and keyframe-based group boundaries. +/// Decodes H.264 (stream type 0x1B), H.265 (0x24), ADTS AAC (0x0F), MP2 +/// (0x03/0x04), AC-3 (0x81), and E-AC-3 (0x87). Each codec stream is fed to its importer, which +/// manages the track, catalog config, and keyframe-based group boundaries. MPEG-1/2 video +/// (0x01/0x02) is not decoded but still paces the program: its PTS anchors the timebase while +/// its PES ride whole as below. /// /// Video access units can span PES packets, and one PES can contain several. Its PTS /// stamps the AU whose required start-code prefix begins there, once. Later AUs need @@ -59,10 +60,18 @@ mod video; /// /// Elementary streams we don't decode are carried verbatim, one MoQ track per /// PID, when the catalog `E` carries the [`mpegts`](super::Mpegts) section: section-framed -/// streams (SCTE-35, marked by a program-level 'CUEI' registration descriptor, and other -/// private sections) are reassembled as sections, and any other stream travels as whole PES -/// packets ([`PesPacket`](super::PesPacket)). With a base `Catalog<()>` they're logged and -/// dropped instead. +/// streams (private sections 0x05, the ISO 13818-6 DSM-CC types 0x0A to 0x0D, and SCTE-35 0x86 +/// marked by a program-level 'CUEI' registration descriptor) are reassembled as sections, and +/// any other stream travels as whole PES packets ([`PesPacket`](super::PesPacket)), LATM/LOAS +/// AAC (0x11) included. A 0x86 stream without 'CUEI', which may be DTS audio or SCTE-35, is not +/// carried. With a base `Catalog<()>` no undecoded stream is carried. +/// +/// Every PID is accounted for. A listed stream that is not carried counts once in +/// [`stats::Stream::refused`] with a warning, and so does each TS-scrambled packet on a carried +/// one. Once the PMT is in, a packet on a PID no table claims (the null PID aside) counts in its +/// own row: see [`stats::Snapshot::streams`]. That holds when the PAT lists only the imported +/// program; a [`with_program`](Self::with_program) import of a multiplex cannot tell another +/// program's PIDs from unclaimed ones, so it leaves them to [`Programs`](super::Programs). /// /// A whole PES or section is timestamped with its placement: the program clock at its first byte, /// interpolated between the PCRs either side of it (ISO 13818-1 2.4.2.2) and shifted onto the @@ -213,6 +222,16 @@ pub struct Import { /// The program number chosen by [`with_program`](Self::with_program). `None` imports the /// multiplex's only program and refuses a PAT that lists more than one. program: Option, + /// The packets on PIDs no table claims, counted once [`alone`](Self::alone) and + /// [`described`](Self::described). Under [`Programs`](super::Programs) this importer reads only + /// the PIDs it claims, so `Programs` counts them instead, once for the multiplex. + unclaimed: super::unclaimed::Unclaimed, + /// The last PAT lists only the imported program, so a PID it does not claim is no other's. + alone: bool, + /// A PMT has described the program, so a PID it does not claim is not one still to come. + described: bool, + /// Carried PIDs already warned about a TS-scrambled packet. + scrambled: HashSet, } impl Import { @@ -280,6 +299,10 @@ impl Import { last_pts: None, pts: PtsUnwrap::default(), program: None, + unclaimed: Default::default(), + alone: false, + described: false, + scrambled: HashSet::new(), } } @@ -379,6 +402,21 @@ impl Import { pids } + /// Whether this program reads `pid`: one it [`claims`](Self::claims), or one claimed whatever + /// the PMT says. + fn claims_pid(&self, pid: u16) -> bool { + super::unclaimed::Unclaimed::fixed(pid, self.supports_mpegts) + || self.sections.contains_key(&pid) + || Pid::new(pid).is_ok_and(|pid| { + self.pmt_sections.contains_key(&pid) || self.streams.contains_key(&pid) || self.pcr_pid == Some(pid) + }) + } + + /// The program clock: PCR ticks since its first PCR, or `None` before one. + pub(super) fn now(&self) -> Option { + self.liveness.now() + } + /// Whether [`claims`](Self::claims) may have changed since the last call. pub(super) fn claims_changed(&mut self) -> bool { std::mem::take(&mut self.claims_changed) @@ -399,6 +437,10 @@ impl Import { continuation, .. } = *packet; + // A packet flagged corrupt may carry another PID's number. + if self.alone && self.described && pkt[1] & 0x80 == 0 && !self.claims_pid(pid) { + self.unclaimed.packet(pid); + } // Every packet is graded, but only the media PIDs and the clock route on the // verdict: PSI and section PIDs keep their reassemblers' own. self.health.packet(pkt, continuation, self.liveness.now()); @@ -467,6 +509,20 @@ impl Import { } } } + // TS-level scrambling hides a payload from all but a descrambler, so the semantic lane + // cannot carry it where the raw lane can. Its adaptation field stays clear: the clock + // above has read it. + if pkt[1] & 0x80 == 0 + && pkt[3] & 0xc0 != 0 + && let Ok(pid) = Pid::new(pid) + && (self.sections.contains_key(&pid.as_u16()) + || self + .streams + .get(&pid) + .is_some_and(|stream| !matches!(stream, Stream::Ignored))) + { + return self.scrambled(pid); + } if let Some(section) = self.sections.get_mut(&pid) { let unplaced = section.packet(pkt, offset, &self.placement)?; let (units, dropped) = section.release(&self.timebase, &self.pts, self.placement.generation)?; @@ -523,10 +579,10 @@ impl Import { } else if self.pmt_sections.contains_key(&pid) { self.pmt_packet(pid, packet)?; } else { - // PIDs we don't decode and don't carry (`Stream::Ignored`: a base catalog's - // undecoded streams, or an ambiguous 0x86 PID without CUEI) are dropped, as is - // every PID before the PSI that declares it: a live capture joins mid-stream, - // so PES arrive before their PMT. + // PIDs we don't carry (`Stream::Ignored`, counted when refused) are dropped, as is + // every PID no table claims (counted by `claims` above) and every PID before the + // PSI that declares it: a live capture joins mid-stream, so PES arrive before + // their PMT. match self.streams.get(&pid) { None | Some(Stream::Ignored) => {} Some(_) => self.pes_packet(pid, pkt, offset)?, @@ -663,9 +719,8 @@ impl Import { if self.supports_mpegts { self.es_descriptors.insert(es.pid.as_u16(), es.descriptors.clone()); } - // Section-framed private data is reassembled as sections, never PES-parsed. - // Everything else routes through ensure_stream (a decoded codec, PES-framed - // verbatim, or dropped). + // Section-framed data is reassembled as sections, never PES-parsed. Everything else + // routes through ensure_stream (a decoded codec, whole PES, or refused). if section_framed(es.stream_type, cuei) { self.ensure_section(es.pid, es.stream_type, &es.descriptors)?; } else { @@ -685,19 +740,13 @@ impl Import { } } self.media_clock = pmt.streams.iter().any(|es| { - matches!( - self.streams.get(&es.pid), - Some( - Stream::H264 { .. } - | Stream::H265 { .. } - | Stream::Aac(_) | Stream::Opus(_) - | Stream::Legacy(_) - | Stream::Clock - ) - ) + self.streams.get(&es.pid).is_some_and(|stream| { + stream.video() || matches!(stream, Stream::Aac(_) | Stream::Opus(_) | Stream::Legacy(_)) + }) }); let pids: Vec = pmt.streams.iter().map(|es| es.pid.as_u16()).collect(); self.health.pmt_streams(&pids); + self.described = true; Ok(()) } @@ -785,7 +834,7 @@ impl Import { // or it would keep intercepting the PID and the new stream would never get data. // This only fires on a genuine remap: section PIDs otherwise route to // `ensure_section`, never here. - if let Some(mut section) = self.sections.remove(&pid.as_u16()) { + if let Some(mut section) = self.retire_section(pid) { section.finish()?; self.pending.remove(&pid); self.headers.remove(&pid); @@ -802,7 +851,7 @@ impl Import { self.unclocked.insert(pid, carried); return Ok(()); } - Some(Stream::Ignored) + Some(Stream::Ignored | Stream::Clock) if self .unclocked .get(&pid) @@ -841,7 +890,7 @@ impl Import { } } // Only ADTS-framed AAC (0x0F). 0x11 is LATM/LOAS, which uses a different - // framing and syncword, so it falls through to the ignored arm below. + // framing and syncword, so it rides whole like any other undecoded stream. Some(StreamType::AdtsAac) => Stream::Aac(Box::new(AacStream { import: None, asc: bytes::Bytes::new(), @@ -878,54 +927,35 @@ impl Import { import: opus::Import::new(track, self.reserve(), config)?, })) } - Err(err) => { - tracing::warn!( - pid = pid.as_u16(), - error = %err, - "unsupported Opus channel configuration, dropping" - ); - Stream::Ignored - } + Err(err) => self.uncarried( + pid, + stream_type, + &format!("unsupported Opus channel configuration: {err:#}"), + ), } } - Some(StreamType::Mpeg1Video | StreamType::Mpeg2Video) => Stream::Clock, - // A codec we don't decode, named or not. Carry it as whole PES packets when the - // catalog supports the `mpegts` section. 0x86 is excluded: it's ambiguous (DTS - // audio, or a non-conformant SCTE-35 mux without CUEI, which is sections a PES - // parse would abort on), so drop it rather than risk PES-parsing sections. - _ => { - if self.supports_mpegts && stream_type != StreamType::Dts8ChannelLosslessAudio as u8 { - if self.pcr_pid.is_none() { - // Nothing could place its PES, so no track is advertised. - tracing::warn!( - stream_type, - pid = pid.as_u16(), - "refused a private PES stream in a program with no PCR to place it by" - ); - *self.refused.entry(pid.as_u16()).or_default() += 1; - self.unclocked.insert(pid, stream_type); - Stream::Ignored - } else { - let descriptors = descriptors.to_vec(); - match PesPacketStream::new( - self.broadcast.clone(), - self.catalog.clone(), - pid.as_u16(), - stream_type, - descriptors, - ) { - Ok(stream) => Stream::PesPacket(Box::new(stream)), - Err(err) => { - tracing::warn!(?err, pid = pid.as_u16(), "failed to create a PES stream, dropping"); - Stream::Ignored - } - } - } - } else { - tracing::warn!(stream_type, pid = pid.as_u16(), "unsupported TS stream type, dropping"); - Stream::Ignored - } + // 0x86 is ambiguous: DTS audio, or a non-conformant SCTE-35 mux without CUEI, which is + // sections a PES parse would abort on. Refuse it rather than guess. + _ if stream_type == StreamType::Dts8ChannelLosslessAudio as u8 => self.uncarried( + pid, + stream_type, + "0x86 without a CUEI registration may be DTS or SCTE-35", + ), + // A codec we don't decode, named or not, MPEG-1/2 video included, rides as whole PES + // packets when the catalog supports the `mpegts` section. + _ if !self.supports_mpegts => self.uncarried(pid, stream_type, "the catalog has no mpegts section"), + _ if self.pcr_pid.is_none() => { + // Nothing could place its PES, so no track is advertised. + self.unclocked.insert(pid, stream_type); + self.uncarried(pid, stream_type, "the program has no PCR to place it by") } + _ => Stream::PesPacket(Box::new(PesPacketStream::new( + self.broadcast.clone(), + self.catalog.clone(), + pid.as_u16(), + stream_type, + descriptors.to_vec(), + )?)), }; // Clock is not a decodable track, so it doesn't initialize the importer. @@ -941,6 +971,17 @@ impl Import { Ok(()) } + /// The route of a listed stream this import does not carry, counted once as refused. MPEG-1/2 + /// video still paces the program by its PTS. + fn uncarried(&mut self, pid: Pid, stream_type: u8, why: &str) -> Stream { + *self.refused.entry(pid.as_u16()).or_default() += 1; + tracing::warn!(pid = pid.as_u16(), stream_type, why, "refused a listed stream"); + match mpeg_video(stream_type) { + true => Stream::Clock, + false => Stream::Ignored, + } + } + fn legacy_stream(&self, pid: Pid, descriptor: &'static legacy::Descriptor) -> Stream { Stream::Legacy(Box::new(LegacyStream { descriptor, @@ -977,15 +1018,12 @@ impl Import { self.restarts.push(pid.as_u16()); if !self.supports_mpegts { // Always route to Ignored, replacing any prior codec on this PID (a later PMT - // can reassign it), so a private section is never PES-parsed. Warn once. - let previous = self.retire_stream(pid); - self.streams.insert(pid, Stream::Ignored); - if !matches!(previous, Some(Stream::Ignored)) { - tracing::warn!( - pid = pid.as_u16(), - "private section stream detected without `mpegts` catalog support; dropping" - ); - } + // can reassign it), so a private section is never PES-parsed. Counted once. + let stream = match self.retire_stream(pid) { + Some(Stream::Ignored) => Stream::Ignored, + _ => self.uncarried(pid, stream_type, "the catalog has no mpegts section"), + }; + self.streams.insert(pid, stream); return Ok(()); } // A prior PMT may have routed this PID to Ignored; drop it so the PID has one route. @@ -1071,6 +1109,7 @@ impl Import { if let Some(stream) = self.streams.get_mut(&pid) && let Stream::PesPacket(stream) = &mut *stream { + stream.end(); stream.settle(&self.placement); let (published, dropped) = stream.release(&self.timebase, &self.pts, generation)?; units += published; @@ -1107,6 +1146,15 @@ impl Import { Ok(()) } + /// Remove the section route on `pid`, keeping the row of the track it published. + fn retire_section(&mut self, pid: Pid) -> Option> { + let section = self.sections.remove(&pid.as_u16())?; + self.retired_stats + .entry(pid.as_u16()) + .or_insert_with(|| stats::Stream::new(".ts", stats::Class::Data)); + Some(section) + } + /// Open the PES whose header `pes` begins where `span` says, if any PCR came before it. fn handle_pes_start(&mut self, pid: Pid, pes: PesStart, span: Option) -> anyhow::Result<()> { // Finish the preceding PES before accepting another header. @@ -1133,11 +1181,11 @@ impl Import { } let stream = &self.streams[&pid]; - let is_video = matches!(stream, Stream::H264 { .. } | Stream::H265 { .. } | Stream::Clock); + let is_video = stream.video(); let is_clock = matches!(stream, Stream::Clock); // Verbatim streams with no cadence (SCTE-35, DVB subtitles) are left ungraded // rather than watched against an invented one. - if pes.pts.is_some() && !matches!(stream, Stream::PesPacket(_) | Stream::Ignored) { + if pes.pts.is_some() && (is_video || !matches!(stream, Stream::PesPacket(_) | Stream::Ignored)) { self.health.pts(pid.as_u16(), self.liveness.now()); } // Resolve the wrap at PES start so every elementary stream and private section shares @@ -1289,10 +1337,7 @@ impl Import { return Ok(()); }; - let batched = self - .streams - .values() - .any(|stream| matches!(stream, Stream::H264 { .. } | Stream::H265 { .. } | Stream::Clock)); + let batched = self.streams.values().any(Stream::video); let Some(stream) = self.streams.get_mut(&pid) else { return Ok(()); }; @@ -1427,22 +1472,52 @@ impl Import { fn refuse(&mut self, pid: Pid, why: &str) -> anyhow::Result<()> { self.pending.remove(&pid); self.headers.remove(&pid); - if let Some(section) = self.sections.remove(&pid.as_u16()) { + if let Some(section) = self.retire_section(pid) { let name = section.entry.name.clone(); section.abort(moq_net::Error::Unsupported); self.retire_track(Some(name))?; } - if let Some(stream) = self.retire_stream(pid) { - let name = stream.track_name(); - stream.abort(moq_net::Error::Unsupported); - self.retire_track(name)?; - } - self.streams.insert(pid, Stream::Ignored); + // MPEG-1/2 video still paces the program once its PES are refused. + let clock = match self.retire_stream(pid) { + Some(stream) => { + let video = stream.video(); + let name = stream.track_name(); + stream.abort(moq_net::Error::Unsupported); + self.retire_track(name)?; + video + } + None => false, + }; + self.streams + .insert(pid, if clock { Stream::Clock } else { Stream::Ignored }); *self.refused.entry(pid.as_u16()).or_default() += 1; tracing::warn!(pid = pid.as_u16(), why, "refused an undecoded stream"); Ok(()) } + /// A TS-scrambled packet on the carried `pid`: refused, and the unit it interrupts lost with + /// it. The route stays, so the stream resumes once its packets are clear again. + fn scrambled(&mut self, pid: Pid) -> anyhow::Result<()> { + self.pending.remove(&pid); + self.headers.remove(&pid); + if let Some(stream) = self.streams.get_mut(&pid) { + stream.lost()?; + } + // A section's continuity counter never sees the packet, so after 16 the next clear one + // would look contiguous and complete the partial with bytes from another section. + if let Some(section) = self.sections.get_mut(&pid.as_u16()) { + section.reassembler = SectionReassembler::default(); + } + *self.refused.entry(pid.as_u16()).or_default() += 1; + if self.scrambled.insert(pid.as_u16()) { + tracing::warn!( + pid = pid.as_u16(), + "refusing TS-scrambled packets; carry the multiplex whole to keep them" + ); + } + Ok(()) + } + /// The packet chain on `pid` was cut: salvage the truncated PES, and treat the bytes /// after it as [lost](Stream::lost). fn broken(&mut self, pid: Pid) -> anyhow::Result<()> { @@ -1491,7 +1566,11 @@ impl Import { for pid in self.streams.keys().copied().collect::>() { self.salvage(pid)?; } - // Whole PES the old clock began go out on its timeline, ahead of the marker. + // Whole PES the old clock began go out on its timeline, ahead of the marker, so none waits + // for a PES on the new one to complete a start code. + for stream in self.streams.values_mut() { + stream.end(); + } self.settle_placement()?; anyhow::ensure!( self.sections.values().all(|section| section.held.is_empty()), @@ -1633,6 +1712,8 @@ impl Import { self.pmt_sections.entry(entry.pmt_pid).or_default(); self.claims_changed = true; self.health.pmt_pids(&[entry.pmt_pid.as_u16()], self.liveness.now()); + // Alongside another program, a PID could be that program's. + self.alone = programs.len() == 1; if self.identity.is_none() { self.identity = Some(identity); self.record_program_identity(pat.transport_stream_id, entry)?; @@ -1697,6 +1778,9 @@ impl Import { for pid in pids { self.flush(pid)?; } + for stream in self.streams.values_mut() { + stream.end(); + } self.settle_placement()?; anyhow::ensure!( self.sections.values().all(|section| section.held.is_empty()), @@ -1754,7 +1838,7 @@ impl Import { // A stream refused before it had a track has no row of its own either, nor does an // SI PID. for (&pid, &unplaced) in &self.unplaced { - let track = if catalog::SI_PIDS.contains(&pid) { ".si" } else { ".ts" }; + let track = if catalog::SI_PIDS.contains(&pid) { ".si" } else { "" }; streams .entry(pid) .or_insert_with(|| stats::Stream::new(track, stats::Class::Data)) @@ -1763,9 +1847,10 @@ impl Import { for (&pid, &refused) in &self.refused { streams .entry(pid) - .or_insert_with(|| stats::Stream::new(".ts", stats::Class::Data)) + .or_insert_with(|| stats::Stream::new("", stats::Class::Data)) .refused = refused; } + self.unclaimed.report(&mut streams, |pid| self.claims_pid(pid)); for (pid, stats) in &mut streams { (stats.units, stats.quiet) = self.liveness.stream(*pid); } @@ -1836,10 +1921,12 @@ struct Listing { } /// Whether a stream of `stream_type` carries sections rather than PES: private sections (0x05), -/// and SCTE-35 (0x86) in a program whose descriptors carry the 'CUEI' registration. 0x86 is also -/// a DTS audio type, so without it the stream is not read as cues. +/// the ISO 13818-6 DSM-CC types, each of which rides DSM-CC sections (0x0A multiprotocol +/// encapsulation, 0x0B U-N messages and carousels, 0x0C stream descriptors, 0x0D any DSM-CC +/// section), and SCTE-35 (0x86) in a program whose descriptors carry the 'CUEI' registration. +/// 0x86 is also a DTS audio type, so without it the stream is not read as cues. fn section_framed(stream_type: u8, cuei: bool) -> bool { - stream_type == 0x05 || (cuei && stream_type == StreamType::Dts8ChannelLosslessAudio as u8) + matches!(stream_type, 0x05 | 0x0a..=0x0d) || (cuei && stream_type == StreamType::Dts8ChannelLosslessAudio as u8) } // Match the raw TS accumulator: half a MoQ group bounds one program's staged input. @@ -2208,14 +2295,40 @@ fn adjust_splice(section: &mut [u8], offset: Offset) { /// catalog clock and every other byte as received, timestamped with its placement on the program /// clock. A PES without a PTS keeps none. See [`Import`] for when one is held, dropped, or refuses /// its stream. +/// +/// Each PES is a group of its own, except MPEG-1/2 video's: a PES in which a sequence header +/// begins opens a group, where a decoder can start, and the pictures after it join that group. +/// Those before the first sequence header, or after a loss until the next, are dropped, since +/// they depend on pictures a subscriber never gets. A video PES ending in part of a start code +/// waits for the next PES on the stream, or for the stream to end, to tell whether a sequence +/// header begins in it. struct PesPacketStream { track: crate::container::Producer, /// Clears this track's catalog entry on drop; its name retires the timeline when the route ends. entry: VerbatimEntry, /// The PMT `stream_type`, which says whether an unbounded PES is video. stream_type: u8, - /// Whole PES waiting for the PCR after them or the timebase's offset, in input order. - held: VecDeque, + /// Whole PES waiting for the PCR after them or the timebase's offset, in input order, each + /// with where it falls in the groups. + held: VecDeque<(Held, Join)>, + /// The stream lost a unit since the last PES held, so the next joins no group. + lost: bool, + /// The PES at the back of `held` ends in part of a start code, so whether it opens a group + /// waits for the PES after it; [`Self::release`] stops before it meanwhile. + straddles: bool, +} + +/// Where a whole PES falls in its track's groups. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Join { + /// It opens a group: every PES of a stream other than MPEG-1/2 video, and one of that video + /// carrying a sequence header. + Opens, + /// It joins the group open before it. + Continues, + /// It follows a loss, so the group before it ends and it is dropped with the PES after it + /// until one opens a group. + Resyncs, } /// A whole PES or section waiting to be placed. @@ -2261,38 +2374,74 @@ impl PesPacketStream { entry, stream_type, held: VecDeque::new(), + lost: false, + straddles: false, }) } /// Hold one complete PES until it can be placed: [`Self::release`] publishes it. fn hold(&mut self, pending: Pending) { let span = pending.span.expect("a whole PES is spanned at its start"); - self.held.push_back(Held { + let join = match mpeg_video(self.stream_type) { + false => Join::Opens, + true => { + let payload = es_payload(&pending.data); + if std::mem::take(&mut self.straddles) { + let (before, join) = self.held.back_mut().expect("a straddling PES is held"); + if straddled_header(es_payload(&before.payload), payload) { + *join = Join::Opens; + } + } + let join = match () { + _ if memchr::memmem::find(payload, &SEQUENCE_HEADER).is_some() => Join::Opens, + _ if self.lost => Join::Resyncs, + _ => Join::Continues, + }; + self.straddles = join != Join::Opens && partial_start_code(payload); + join + } + }; + self.lost = false; + let held = Held { payload: pending.data, span, _retained: pending.retained, - }); + }; + self.held.push_back((held, join)); + } + + /// The unit in progress was lost: an MPEG-1/2 picture after it may reference it, and no + /// sequence header continues from the PES before it. + fn lost(&mut self) { + self.lost = true; + self.straddles = false; + } + + /// No PES follows the last held on its clock, so a start code its end began completes nothing. + fn end(&mut self) { + self.straddles = false; } /// The PCR `after` arrived on timebase `generation`: every PES held since the last is /// bracketed. fn bracket(&mut self, generation: u64, after: Sample) { - for held in &mut self.held { + for (held, _) in &mut self.held { held.span.bracket(generation, after); } } /// The timebase of `placement` ends: extrapolate the PES held past its last PCR. fn settle(&mut self, placement: &Placement) { - for held in &mut self.held { + for (held, _) in &mut self.held { held.span.settle(placement); } } - /// Publish the placed PES at the front, each in its own group, once `timebase` has its + /// Publish the placed PES at the front, grouped as [`Join`] says, once `timebase` has its /// offset, and drop those no PCR on their own timebase can place: one ended before them, now /// that `generation` is the timebase running. `pts` is the program's PTS history, which - /// picks each PCR's wrap. Returns how many went each way. + /// picks each PCR's wrap. Returns how many went each way; a picture no group takes is + /// neither. fn release( &mut self, timebase: &crate::catalog::Timebase, @@ -2306,23 +2455,33 @@ impl PesPacketStream { while let Some(timestamp) = self .held .front() - .and_then(|held| held.timestamp(pts, offset, generation)) + .filter(|_| !(self.straddles && self.held.len() == 1)) + .and_then(|(held, _)| held.timestamp(pts, offset, generation)) { - let mut held = self.held.pop_front().expect("a front"); + let (mut held, join) = self.held.pop_front().expect("a front"); let Some(timestamp) = timestamp else { unplaced += 1; continue; }; rebase(&mut held.payload, offset); + if join == Join::Resyncs { + self.track.cut(None)?; + } let frame = crate::container::Frame { timestamp, duration: None, payload: bytes::Bytes::from(held.payload), - keyframe: true, + keyframe: join == Join::Opens, }; - self.track.write(frame)?; - self.track.cut(None)?; - published += 1; + match self.track.write(frame) { + Ok(()) => published += 1, + // A picture with no group open: before the first sequence header, or after a loss. + Err(crate::Error::MissingKeyframe(_)) => continue, + Err(err) => return Err(err.into()), + } + if !mpeg_video(self.stream_type) { + self.track.cut(None)?; + } } Ok((published, unplaced)) } @@ -2930,6 +3089,39 @@ fn unit_error(err: crate::Error) -> anyhow::Error { } } +/// Whether `stream_type` is MPEG-1 or MPEG-2 video, which is carried whole but still paces the +/// program like decoded video. +fn mpeg_video(stream_type: u8) -> bool { + matches!(stream_type, 0x01 | 0x02) +} + +/// An MPEG-1/2 `sequence_header_code` (ISO 13818-2 6.2.2.1), where a decoder can start. Start +/// codes never occur inside coded data, so any match is one. +const SEQUENCE_HEADER: [u8; 4] = [0x00, 0x00, 0x01, 0xb3]; + +/// The elementary stream bytes of the whole PES `pes`, after its header. +fn es_payload(pes: &[u8]) -> &[u8] { + let header = match pes.get(3) { + Some(&stream_id) if psi::has_optional_header(stream_id) => 9 + usize::from(pes.get(8).copied().unwrap_or(0)), + _ => 6, + }; + pes.get(header..).unwrap_or_default() +} + +/// Whether `payload` ends in the first one to three bytes of a start code (`00`, `00 00`, or +/// `00 00 01`), which the next PES may complete. +fn partial_start_code(payload: &[u8]) -> bool { + matches!(payload, [.., 0x00] | [.., 0x00, 0x00, 0x01]) +} + +/// Whether a sequence header begins in `before`'s payload and ends in `after`'s, the next PES +/// on the stream. +fn straddled_header(before: &[u8], after: &[u8]) -> bool { + let before = &before[before.len().saturating_sub(SEQUENCE_HEADER.len() - 1)..]; + let after = &after[..after.len().min(SEQUENCE_HEADER.len() - 1)]; + memchr::memmem::find(&[before, after].concat(), &SEQUENCE_HEADER).is_some() +} + /// One elementary stream's codec importer. enum Stream { H264 { @@ -2945,10 +3137,10 @@ enum Stream { Aac(Box>), Opus(Box), Legacy(Box>), - /// A codec we don't decode, carried as whole PES (DTS audio, private PES, ...). + /// A codec we don't decode, carried as whole PES (MPEG-1/2 video, LATM AAC, private PES, ...). PesPacket(Box>), - /// MPEG-1/2 video we don't decode, kept only to advance the media clock. - /// `is_video` counts it, so never reuse this variant for audio or data. + /// MPEG-1/2 video this import refused to carry, kept only to advance the media clock. + /// [`Self::video`] counts it, so never reuse this variant for audio or data. Clock, Ignored, } @@ -2985,6 +3177,15 @@ impl Stream { } } + /// Whether this is the program's video, decoded or not, whose pictures pace the program. + fn video(&self) -> bool { + match self { + Stream::H264 { .. } | Stream::H265 { .. } | Stream::Clock => true, + Stream::PesPacket(stream) => mpeg_video(stream.stream_type), + _ => false, + } + } + /// Whether nothing waits to be placed on the program clock. fn placed(&self) -> bool { match self { @@ -3039,11 +3240,19 @@ impl Stream { match self { Stream::H264 { import, .. } => import.cut(None)?, Stream::H265 { import, .. } => import.cut(None)?, + Stream::PesPacket(stream) => stream.lost(), _ => {} } Ok(()) } + /// No PES follows the last on this clock or input: see [`PesPacketStream::end`]. + fn end(&mut self) { + if let Stream::PesPacket(stream) = self { + stream.end(); + } + } + /// The program clock restarted: mark the break on this track. fn discontinuity(&mut self, published: bool) -> anyhow::Result<()> { match self { @@ -3155,7 +3364,10 @@ impl Stream { Stream::H264 { .. } => stats::Stream::new(".avc3", stats::Class::Video), Stream::H265 { .. } => stats::Stream::new(".hev1", stats::Class::Video), Stream::Opus(_) => stats::Stream::new(".opus", stats::Class::Audio), - Stream::PesPacket(_) => stats::Stream::new(".ts", stats::Class::Data), + Stream::PesPacket(stream) => match mpeg_video(stream.stream_type) { + true => stats::Stream::new(".ts", stats::Class::Video), + false => stats::Stream::new(".ts", stats::Class::Data), + }, Stream::Clock => stats::Stream::new("", stats::Class::Video), Stream::Ignored => return None, }) @@ -4569,7 +4781,10 @@ pub(super) mod test { .unwrap(); assert!(matches!(import.streams.get(&pid), Some(super::Stream::Ignored))); - assert_eq!(import.stats(), before, "a PMT remap must not reset lifetime counters"); + // A base catalog cannot carry the section route, which counts as refused. + let mut expected = before; + expected.streams.get_mut(&pid.as_u16()).unwrap().refused = 1; + assert_eq!(import.stats(), expected, "a PMT remap must not reset lifetime counters"); } /// Build a payload-only TS packet (PID 0x0021, afc 0b01). `body` is the bytes @@ -4846,7 +5061,7 @@ pub(super) mod test { import.finish().unwrap(); assert_eq!( - catalog.snapshot().ext.mpegts.tracks.len(), + cue_tracks(&catalog.snapshot().ext.mpegts), 1, "expected one scte35 rendition" ); @@ -4907,7 +5122,7 @@ pub(super) mod test { first.video.renditions.is_empty() && first.audio.renditions.is_empty(), "only the extension lists a track" ); - assert_eq!(first.ext.mpegts.tracks.len(), 1, "the cue track is listed"); + assert_eq!(cue_tracks(&first.ext.mpegts), 1, "the cue track is listed"); } // The base catalog (`Catalog<()>`) can't carry cues, so a detected CUEI PID routes to @@ -5003,14 +5218,14 @@ pub(super) mod test { "upgrade drops the stale Ignored route" ); assert_eq!( - catalog.snapshot().ext.mpegts.tracks.len(), + cue_tracks(&catalog.snapshot().ext.mpegts), 1, "upgrade advertises the cue track" ); // The importer clears its verbatim entries from the catalog when it drops, so read the // track name while it is still registered. - let name = catalog.snapshot().ext.mpegts.tracks.keys().next().unwrap().clone(); + let name = track_on(&catalog.snapshot().ext.mpegts, SECTION_PID); import.finish().unwrap(); let track = consumer .track(&name) @@ -5060,14 +5275,15 @@ pub(super) mod test { true, )); bytes.extend_from_slice(&pcr_packet(VIDEO, 24_300_000)); - bytes.extend_from_slice(&pes_packet(VIDEO, 90_000)); + // A sequence header, so the picture opens a group rather than waiting for one. + bytes.extend_from_slice(&video_pes(VIDEO, 0, 90_000, None, &[0, 0, 1, 0xb3, 0, 0, 1, 0])); bytes.extend_from_slice(&packet(true, 0, 0, &CUE)); bytes.extend_from_slice(&pcr_packet(VIDEO, 27_000_000)); import.decode(&bytes).unwrap(); let stats = import.stats(); assert_eq!(stats.streams[&VIDEO].class, super::stats::Class::Video); - assert_eq!(stats.streams[&VIDEO].track, ""); + assert_eq!(stats.streams[&VIDEO].track, ".ts"); assert!( stats.streams[&VIDEO].units >= 1, "the video PID delivered before it stalled" @@ -5092,6 +5308,17 @@ pub(super) mod test { }); } + /// The section tracks the `mpegts` section lists, beside any whole-PES video clocking them. + fn cue_tracks(mpegts: &super::catalog::Mpegts) -> usize { + mpegts.tracks.values().filter(|track| track.verbatim.is_some()).count() + } + + /// The track the `mpegts` section carries `pid` on. + fn track_on(mpegts: &super::catalog::Mpegts, pid: u16) -> String { + let found = mpegts.tracks.iter().find(|(_, track)| track.pid == pid); + found.expect("a track on the PID").0.clone() + } + /// A PUSI TS packet on `pid` carrying a minimal PES with `pts` (90 kHz) and a /// 1-byte dummy payload, for streams we observe only for their PTS. fn pes_packet(pid: u16, pts: u64) -> Vec { @@ -6952,7 +7179,7 @@ pub(super) mod test { let expected = 1_000_000 + (at - (2 * 188 + 10)) * 100_000 / (3 * 188); import.finish().unwrap(); - let name = catalog.snapshot().ext.mpegts.tracks.keys().next().unwrap().clone(); + let name = track_on(&catalog.snapshot().ext.mpegts, 0x21); let track = consumer .track(&name) .unwrap() @@ -7007,7 +7234,7 @@ pub(super) mod test { let last = data.len() as u64 + 10; data.extend(pcr_packet(PCR_PID, 31_050_000)); let (consumer, catalog, _import) = import_all(&data).unwrap(); - let name = catalog.snapshot().ext.mpegts.tracks.keys().next().unwrap().clone(); + let name = track_on(&catalog.snapshot().ext.mpegts, 0x21); let frames = read_track(&consumer, &name, crate::container::Kind::Data).await; let times: Vec<_> = frames.iter().map(|frame| frame.timestamp.as_micros() as u64).collect(); let clock = |at: u64, before: u64, ticks: u64, after: u64| { @@ -7042,7 +7269,7 @@ pub(super) mod test { data.extend(packet(true, 1, 0, &CUE)); let (consumer, catalog, import) = import_all(&data).unwrap(); assert_eq!(import.stats().streams[&0x21].unplaced, 2); - let name = catalog.snapshot().ext.mpegts.tracks.keys().next().unwrap().clone(); + let name = track_on(&catalog.snapshot().ext.mpegts, 0x21); assert!( read_track(&consumer, &name, crate::container::Kind::Data) .await @@ -7089,7 +7316,13 @@ pub(super) mod test { "video kept importing past the dropped section PID" ); assert!( - catalog.snapshot().ext.mpegts.tracks.is_empty(), + !catalog + .snapshot() + .ext + .mpegts + .tracks + .values() + .any(|track| track.pid == SECTION_PID), "a 0x86 PID without CUEI must not be cataloged" ); } @@ -7658,7 +7891,7 @@ pub(super) mod test { import.decode(pes_packet(PCR_PID, 90_000).as_slice()).unwrap(); import.decode(pcr_packet(PCR_PID, 27_000_000).as_slice()).unwrap(); import.finish().unwrap(); - let name = catalog.snapshot().ext.mpegts.tracks.keys().next().unwrap().clone(); + let name = track_on(&catalog.snapshot().ext.mpegts, 0x21); let (frames, breaks) = read_breaks(&consumer, &name, crate::container::Kind::Data).await; assert_eq!(frames.len(), 3); assert_eq!( @@ -8240,7 +8473,7 @@ pub(super) mod test { import .decode(&synth_pmt( &[ - (StreamType::Mpeg2Video, 0x50), + (StreamType::AdtsAac, 0x50), (StreamType::Dts8ChannelLosslessAudio, 0x21), ], true, @@ -9585,7 +9818,7 @@ pub(super) mod test { let mut updates = catalog.consume().unwrap(); let mut import = super::Import::new(broadcast, catalog.reserve()); import - .decode(&synth_pmt(&[(StreamType::Mpeg2Video, PCR_PID)], false)) + .decode(&synth_pmt(&[(StreamType::H264, PCR_PID)], false)) .unwrap(); let waiter = kio::Waiter::noop(); // Each epoch carries 0.96 s of media, then a flagged PCR restarts the clock. @@ -10211,7 +10444,7 @@ pub(super) mod test { (257, 0x06, "verbatim", "pesPacket"), (258, 0x05, "verbatim", "section"), (259, 0x86, "dropped", "none"), - (260, 0x02, "dropped", "clock only"), + (260, 0x02, "verbatim", "pesPacket"), (261, 0x0f, "track", "decoded"), ] { if !streams.iter().any(|line| { @@ -10329,7 +10562,7 @@ pub(super) mod test { assert!(catalog.snapshot().ext.mpegts.tracks.is_empty()); assert!(matches!(import.streams[&pid], super::Stream::Ignored)); import.handle_pmt(pmt(Some(Pid::new(PCR_PID).unwrap()))).unwrap(); - let name = catalog.snapshot().ext.mpegts.tracks.keys().next().unwrap().clone(); + let name = track_on(&catalog.snapshot().ext.mpegts, 0x21); let data = [ pcr_packet(PCR_PID, 27_000_000), pes_packet(PCR_PID, 99_000), @@ -10358,7 +10591,7 @@ pub(super) mod test { ); assert!(catalog.snapshot().ext.mpegts.tracks.is_empty()); import.handle_pmt(pmt(Some(Pid::new(PCR_PID).unwrap()))).unwrap(); - let next = catalog.snapshot().ext.mpegts.tracks.keys().next().unwrap().clone(); + let next = track_on(&catalog.snapshot().ext.mpegts, 0x21); assert_ne!(next, name, "restoring a stream publishes a new track"); import.finish().unwrap(); } diff --git a/rs/moq-mux/src/container/ts/import_test.rs b/rs/moq-mux/src/container/ts/import_test.rs index 301adc2bad3..beef3734bd5 100644 --- a/rs/moq-mux/src/container/ts/import_test.rs +++ b/rs/moq-mux/src/container/ts/import_test.rs @@ -959,15 +959,26 @@ fn import_reports_a_silent_scte35_pid() { assert_stalls(data, 0x14d, ".ts", 0x100, 2.5, 3.5); } -/// Every elementary stream the importer carries reports a row, decoded, clock-only, and -/// verbatim alike, and every row has delivered. An SI PID has a row only for the sections -/// it sent before the program clock could place them, as ffmpeg leads with its SDT. +/// Every PID reports a row: each elementary stream the importer carries, decoded, whole PES +/// (MPEG-2 video among them) and sections alike, has delivered, and each PID no table claims +/// (an ATSC capture's CAT, PSIP base PID and its private PIDs) counts what it dropped. An SI +/// PID has a row only for the sections it sent before the program clock could place them, as +/// ffmpeg leads with its SDT. #[test] fn import_reports_every_elementary_stream() { for (data, rows) in [ ( &include_bytes!("test_data/kyrion_mpeg2av_ac3.ts")[..], - &[(0x100, ""), (0x101, ".ac3"), (0x102, ".mp2"), (0x14d, ".ts")][..], + &[ + (0x0001, ""), + (0x100, ".ts"), + (0x101, ".ac3"), + (0x102, ".mp2"), + (0x14d, ".ts"), + (0x1fd0, ""), + (0x1fd1, ""), + (0x1ffb, ""), + ][..], ), ( &include_bytes!("test_data/scte35/bbb5s.ts")[..], @@ -981,6 +992,11 @@ fn import_reports_every_elementary_stream() { assert_eq!((sdt.units, sdt.unplaced), (0, 1), "{sdt:?}"); } for (pid, stream) in &stats.streams { + if stream.track.is_empty() { + assert_eq!(stream.class, super::stats::Class::Data, "{pid:#x}: {stream:?}"); + assert!(stream.refused > 0 && stream.units == 0, "{pid:#x}: {stream:?}"); + continue; + } assert!(stream.units > 0, "{pid:#x} delivered nothing: {stream:?}"); assert!(stream.quiet.is_some(), "{pid:#x} has no program clock: {stream:?}"); } diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 7a329dc5b6c..ac1956e9509 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -37,6 +37,7 @@ mod psi; mod release; mod schedule; mod si; +mod unclaimed; // The `mpegts` catalog section (per-track PID + descriptors plus verbatim carriage // of undecoded elementary streams) and the `Catalog` capability, re-exported flat so @@ -73,3 +74,5 @@ mod recovery_test; mod release_test; #[cfg(test)] mod revision_test; +#[cfg(test)] +mod routes_test; diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 462550276c2..141a353c089 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -7,6 +7,7 @@ use mpeg2ts::ts::Pid; use super::demux::{Demux, Packet}; use super::import::SectionReassembler; +use super::unclaimed::Unclaimed; use super::{Ext, Import, catalog::SI_PIDS, psi, stats}; use crate::catalog; @@ -21,7 +22,11 @@ use crate::catalog; /// lists. A program a later PAT adds is not published; removing a started one ends the import. /// /// The multiplex is parsed once: framing, continuity, the PAT and the SI sections are shared, -/// and each program reads only the packets on the PIDs its PAT entry and PMT claim. +/// and each program reads only the packets on the PIDs its PAT entry and PMT claim. A packet on a +/// PID no program reads, the PAT's PMT PIDs, the SI PIDs and the null PID aside, is counted once +/// for the multiplex (see [`stats::Snapshot::streams`]). Counting waits until every program the +/// PAT lists has started, or 0.5 s of program clock has passed without its PMT (TR 101 290's +/// `PMT_error_2`), since until then a PID may be a program's still to come. pub struct Programs { origin: moq_net::origin::Producer, name: moq_net::PathOwned, @@ -41,8 +46,19 @@ pub struct Programs { readers: Vec>, /// The programs the current [`decode`](Self::decode) call routed a packet to. touched: Vec, + /// Every PMT PID the latest PAT lists, which the PAT claims whether or not a program reads it. + pmt_pids: HashSet, + /// The programs the PAT lists that no PMT has started, holding back the unclaimed count. + holdouts: HashSet, + /// The first program's clock past which every holdout's PMT has lapsed, once it has a clock. + deadline: Option, + /// The packets on PIDs no program reads. + unclaimed: Unclaimed, } +/// `PMT_error_2`: a PMT recurs at least every 0.5 s of program clock. +const PMT_LIMIT: u64 = super::passthrough::PCR_HZ / 2; + /// The programs waiting for their PMT on one PID. #[derive(Default)] struct Waiting { @@ -104,6 +120,10 @@ impl Programs { programs: Vec::new(), readers: vec![Vec::new(); 1 << 13], touched: Vec::new(), + pmt_pids: HashSet::new(), + holdouts: HashSet::new(), + deadline: None, + unclaimed: Unclaimed::default(), } } @@ -118,6 +138,10 @@ impl Programs { fn decode_with(&mut self, demux: &mut Demux, data: &[u8]) -> anyhow::Result<()> { demux.push(data); while let Some(packet) = demux.next() { + // The holdouts have no clock of their own until their PMT names one. + if self.deadline.is_none() && !self.holdouts.is_empty() { + self.deadline = self.now().map(|now| now + PMT_LIMIT); + } if let Some(pat) = &packet.pat { self.wait(pat); } else if !self.waiting.is_empty() { @@ -128,7 +152,10 @@ impl Programs { let readers = match everyone { true => None, false => match self.readers[usize::from(packet.pid)].as_slice() { - [] => continue, + [] => { + self.unclaimed(&packet); + continue; + } readers => Some(readers), }, }; @@ -191,6 +218,7 @@ impl Programs { for (&number, &pid) in &listed { self.waiting.entry(pid).or_default().programs.insert(number); } + self.holdouts = listed.keys().copied().collect(); } else { let mut moved = Vec::new(); self.waiting.retain(|pid, waiting| { @@ -207,10 +235,43 @@ impl Programs { for (number, pid) in moved { self.waiting.entry(pid).or_default().programs.insert(number); } + // A program the PAT adds holds the count back afresh. + self.holdouts.retain(|number| listed.contains_key(number)); + let started: HashSet = self.programs.iter().map(|program| program.number).collect(); + for &number in listed.keys() { + if !started.contains(&number) && self.holdouts.insert(number) { + self.deadline = None; + } + } } + self.pmt_pids = listed.values().copied().collect(); self.pat = Some(pat.clone()); } + /// The first program's clock, which times the holdouts' lapse. + fn now(&self) -> Option { + self.programs.first().and_then(|program| program.import.now()) + } + + /// Count a packet on a PID no program reads, unless the PAT claims it or a holdout's PMT may + /// still claim it. + fn unclaimed(&mut self, packet: &Packet) { + // A packet flagged corrupt may carry another PID's number. + if packet.pkt[1] & 0x80 != 0 || Unclaimed::fixed(packet.pid, true) || self.pmt_pids.contains(&packet.pid) { + return; + } + let acquired = match self.holdouts.is_empty() { + true => self.pat.is_some(), + false => self + .deadline + .zip(self.now()) + .is_some_and(|(deadline, now)| now > deadline), + }; + if acquired { + self.unclaimed.packet(packet.pid); + } + } + /// Read a packet on a waiting PMT PID, and start each program whose PMT it completes with a /// stream listed. /// @@ -279,6 +340,7 @@ impl Programs { /// Create the broadcast of program `number`, its importer seeded with the PAT and carrying /// on the PMT `sections` began on `pmt_pid`. fn start(&mut self, number: u16, pmt_pid: u16, sections: SectionReassembler) -> anyhow::Result<()> { + self.holdouts.remove(&number); let pat = self.pat.as_ref().expect("a program waits on its PAT"); let name = program_broadcast(self.name.as_str(), number); let broadcast = self @@ -334,7 +396,8 @@ impl Programs { /// The multiplex is graded once. Programs sharing a PMT PID read the same PMT sections, so /// a dropped section counts once however many read it, and programs sharing a PCR PID /// count its faults once too. PMT sections dropped before any program read their PID are - /// counted here. + /// counted here, and so are the packets on PIDs no program reads, in the row of the program + /// that reads the PID whenever one does. pub fn stats(&self) -> stats::Snapshot { let mut stats = stats::Snapshot::default(); let mut crc_errors = BTreeMap::::new(); @@ -347,6 +410,9 @@ impl Programs { } errors.max(program.import.errors()); } + self.unclaimed.report(&mut stats.streams, |pid| { + Unclaimed::fixed(pid, true) || self.pmt_pids.contains(&pid) || !self.readers[usize::from(pid)].is_empty() + }); stats.crc_error = crc_errors.values().chain(self.crc_errors.values()).sum::() + self.demux.pat_crc_error(); errors.report(&mut stats); stats diff --git a/rs/moq-mux/src/container/ts/routes_test.rs b/rs/moq-mux/src/container/ts/routes_test.rs new file mode 100644 index 00000000000..7ee3197b082 --- /dev/null +++ b/rs/moq-mux/src/container/ts/routes_test.rs @@ -0,0 +1,545 @@ +//! Every PID of an imported program takes one declared route: decoded, whole PES, section, +//! captured PSI/SI, or not carried and counted. + +use std::time::Duration; + +use super::catalog::Ext; +use super::fixture::{Flags, MS, Mux}; +use super::stats::{Class, Stream}; +use super::{Import, Programs, psi}; + +const PMT: u16 = 0x20; +const CLOCK: u16 = 0x1ff; +const VIDEO: u16 = 0x100; +const DATA: u16 = 0x101; +const UNLISTED: u16 = 0x300; +/// The PCR leads each picture's PTS by this many 90 kHz ticks. +const LEAD: u64 = 9_000; +/// `registration_descriptor` naming SMPTE RP 217 KLV ('KLVA'). +const KLVA: &[u8] = &[0x05, 0x04, b'K', b'L', b'V', b'A']; + +/// A 33-bit PTS field behind the '0010' prefix. +fn pts_field(ticks: u64) -> [u8; 5] { + [ + 0x21 | ((ticks >> 29) & 0x0e) as u8, + (ticks >> 22) as u8, + ((ticks >> 14) & 0xfe) as u8 | 1, + (ticks >> 7) as u8, + ((ticks << 1) & 0xfe) as u8 | 1, + ] +} + +/// An MPEG-2 video PES presenting at `pts`, its length left unbounded as video may: a sequence +/// header start code when `sequence`, a picture start code, then `len` bytes of `fill`. +fn picture(pts: u64, sequence: bool, fill: u8, len: usize) -> Vec { + let mut pes = vec![0, 0, 1, 0xe0, 0, 0, 0x80, 0x80, 5]; + pes.extend_from_slice(&pts_field(pts)); + if sequence { + pes.extend_from_slice(&[0, 0, 1, 0xb3]); + } + pes.extend_from_slice(&[0, 0, 1, 0x00]); + pes.resize(pes.len() + len, fill); + pes +} + +/// An asynchronous KLV PES (private_stream_1, no PTS) of `len` bytes of `fill`. +fn klv(fill: u8, len: usize) -> Vec { + let mut pes = vec![0, 0, 1, 0xbd]; + pes.extend_from_slice(&((3 + len) as u16).to_be_bytes()); + pes.extend_from_slice(&[0x84, 0x00, 0x00]); + pes.resize(pes.len() + len, fill); + pes +} + +/// A bounded audio PES presenting at `pts`, carrying `payload`. +fn audio(pts: u64, payload: &[u8]) -> Vec { + let mut pes = vec![0, 0, 1, 0xc0]; + pes.extend_from_slice(&((8 + payload.len()) as u16).to_be_bytes()); + pes.extend_from_slice(&[0x84, 0x80, 5]); + pes.extend_from_slice(&pts_field(pts)); + pes.extend_from_slice(payload); + pes +} + +/// A `pid` packet whose payload is TS-scrambled. +fn scrambled(mux: &mut Mux, pid: u16) { + let at = mux.out.len(); + mux.fill(pid, 1); + mux.out[at + 3] |= 0x80; +} + +/// An importer of one program, onto a catalog with the `mpegts` section. +struct Imported { + consumer: moq_net::broadcast::Consumer, + catalog: crate::catalog::Producer, + import: Import, +} + +impl Imported { + fn new() -> Self { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let import = Import::new(broadcast, catalog.reserve()); + Self { + consumer, + catalog, + import, + } + } + + /// Import all of `data`. + fn all(data: &[u8]) -> Self { + let mut imported = Self::new(); + imported.import.decode(data).unwrap(); + imported.import.finish().unwrap(); + imported + } + + /// The `mpegts` catalog entry for `pid`: its track name and record. + fn entry(&self, pid: u16) -> Option<(String, super::catalog::Track)> { + let mpegts = self.catalog.snapshot().ext.mpegts; + mpegts.tracks.into_iter().find(|(_, track)| track.pid == pid) + } + + /// The stats row for `pid`. + fn row(&self, pid: u16) -> Stream { + self.import.stats().streams.get(&pid).cloned().unwrap_or_default() + } + + /// Every frame on the track carrying `pid`. + async fn frames(&self, pid: u16) -> Vec { + let (name, _) = self.entry(pid).expect("a track on the PID"); + let subscription = moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(3600)); + let track = self + .consumer + .track(&name) + .unwrap() + .subscribe(subscription) + .await + .unwrap(); + let mut reader = crate::container::Consumer::new( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + ); + let mut frames = Vec::new(); + while let Ok(Ok(Some(frame))) = tokio::time::timeout(Duration::from_millis(50), reader.read_frame()).await { + frames.push(frame); + } + frames + } +} + +/// MPEG-2 video rides whole while its PTS still paces the program, and a TS round trip keeps +/// every PES byte of it. A sequence header opens each group, so the picture joined mid-GOP is +/// dropped, and so is the picture after one lost to scrambling, until the next sequence header. +#[tokio::test(start_paused = true)] +async fn mpeg2_video_rides_whole_and_survives_a_round_trip() { + let mut mux = Mux::default(); + mux.pat(&[(1, PMT)]).pmt(PMT, 1, CLOCK, &[(0x02, VIDEO)], 0); + let start = 2_000 * MS; + let mut pictures = Vec::new(); + for i in 0..9u64 { + let ticks = start + i * 40 * MS; + mux.packet(CLOCK, Flags::pcr(ticks)); + // Picture 6 is still open, its length unbounded, when the scrambled packet loses it. + if i == 7 { + scrambled(&mut mux, VIDEO); + } + let pes = picture( + ticks / 300 + LEAD, + matches!(i, 1 | 5 | 8), + i as u8, + 300 + i as usize * 50, + ); + mux.whole(VIDEO, &pes); + pictures.push(pes); + } + mux.packet(CLOCK, Flags::pcr(start + 9 * 40 * MS)); + // What a subscriber gets, and which pictures open a group. + let carried: Vec<_> = [1, 2, 3, 4, 5, 8].map(|i| pictures[i].clone()).into(); + let opens = [true, false, false, false, true, true]; + + let imported = Imported::all(&mux.bytes()); + let (_, entry) = imported.entry(VIDEO).expect("the video is in the catalog"); + assert_eq!(entry.pes_packet.as_ref().map(|pes| pes.stream_type), Some(0x02)); + let snapshot = imported.catalog.snapshot(); + assert!( + snapshot.video.renditions.is_empty(), + "no rendition a player cannot decode" + ); + let frames = imported.frames(VIDEO).await; + let payloads: Vec<_> = frames.iter().map(|frame| frame.payload.to_vec()).collect(); + assert_eq!(payloads, carried); + let keyframes: Vec<_> = frames.iter().map(|frame| frame.keyframe).collect(); + assert_eq!(keyframes, opens); + assert!(frames.windows(2).all(|pair| pair[0].timestamp < pair[1].timestamp)); + let row = imported.row(VIDEO); + assert_eq!((row.track.as_str(), row.class), (".ts", Class::Video)); + assert_eq!((row.units, row.refused, row.damaged), (6, 1, 0)); + + let ts = super::export_test::drain_with( + super::export_test::open_ts( + crate::source::announced(&imported.consumer).await, + crate::catalog::CatalogFormat::Hang, + ) + .await + .unwrap(), + ) + .await; + let again = Imported::all(&ts); + let (_, entry) = again.entry(VIDEO).expect("the export lists the video"); + assert_eq!(entry.pes_packet.as_ref().map(|pes| pes.stream_type), Some(0x02)); + let frames = again.frames(VIDEO).await; + let payloads: Vec<_> = frames.iter().map(|frame| frame.payload.to_vec()).collect(); + assert_eq!(payloads, carried, "the round trip keeps every PES byte"); + let keyframes: Vec<_> = frames.iter().map(|frame| frame.keyframe).collect(); + assert_eq!(keyframes, opens, "and its groups"); +} + +/// A sequence header split across two video PES, at each of its three inner boundaries, opens +/// its group at the PES it begins in, while a picture start code split the same way opens none. +/// A whole PES ending in part of a start code with no whole PES after it on its clock still goes +/// out, at a clock break and at the end of input. +#[tokio::test(start_paused = true)] +async fn mpeg2_sequence_header_split_across_pes_opens_a_group() { + let mut mux = Mux::default(); + mux.pat(&[(1, PMT)]).pmt(PMT, 1, CLOCK, &[(0x02, VIDEO)], 0); + let start = 2_000 * MS; + // Each PES's payload after the picture start code it opens with, its last bytes completing + // in the next PES: a sequence header at `00 00 | 01 b3`, `00 | 00 01 b3`, and `00 00 01 | b3`, + // and a picture start code at `00 00 | 01 00`. A clock break cuts the sixth, its length + // unbounded, so the fifth is the last whole PES on the old clock. The seventh, carrying a + // whole sequence header, ends the input. + let tails: [(&[u8], &[u8]); 7] = [ + (&[], &[0, 0]), + (&[1, 0xb3], &[0]), + (&[0, 1, 0xb3], &[]), + (&[], &[0, 0, 1]), + (&[0xb3], &[0, 0]), + (&[1, 0], &[0]), + (&[0, 0, 1, 0xb3], &[0, 0]), + ]; + let restart = 9_000 * MS; + let mut pictures = Vec::new(); + for (i, (head, tail)) in tails.into_iter().enumerate() { + let ticks = match i { + 6 => restart, + _ => start + i as u64 * 40 * MS, + }; + match i { + 6 => mux.packet(CLOCK, Flags::pcr(ticks).discontinuity()), + _ => mux.packet(CLOCK, Flags::pcr(ticks)), + }; + let mut pes = vec![0, 0, 1, 0xe0, 0, 0, 0x80, 0x80, 5]; + pes.extend_from_slice(&pts_field(ticks / 300 + LEAD)); + pes.extend_from_slice(head); + pes.extend_from_slice(&[0, 0, 1, 0x00]); + pes.resize(pes.len() + 200, 0x55); + pes.extend_from_slice(tail); + mux.whole(VIDEO, &pes); + pictures.push(pes); + } + mux.packet(CLOCK, Flags::pcr(restart + 40 * MS)); + + let imported = Imported::all(&mux.bytes()); + let frames = imported.frames(VIDEO).await; + // The PES on the new clock has its PTS rebased, so compare around it. + let unstamped = |pes: &[u8]| [&pes[..9], &pes[14..]].concat(); + let payloads: Vec<_> = frames.iter().map(|frame| unstamped(&frame.payload)).collect(); + pictures.remove(5); + let pictures: Vec<_> = pictures.iter().map(|pes| unstamped(pes)).collect(); + assert_eq!( + payloads, pictures, + "every whole PES goes out, at the clock break and the end of input" + ); + let keyframes: Vec<_> = frames.iter().map(|frame| frame.keyframe).collect(); + assert_eq!(keyframes, [true, true, false, true, false, true]); +} + +/// Each ISO 13818-6 DSM-CC type rides DSM-CC sections, so each is reassembled as sections +/// rather than failing a PES start code. +#[tokio::test(start_paused = true)] +async fn dsmcc_types_ride_as_sections() { + let types = [ + (0x0a, 0x110, 0x3a), + (0x0b, 0x111, 0x3b), + (0x0c, 0x112, 0x3d), + (0x0d, 0x113, 0x3e), + ]; + let mut mux = Mux::default(); + let streams: Vec<_> = types.iter().map(|&(stream_type, pid, _)| (stream_type, pid)).collect(); + mux.pat(&[(1, PMT)]).pmt(PMT, 1, CLOCK, &streams, 0); + mux.packet(CLOCK, Flags::pcr(0)); + let mut sections = Vec::new(); + for &(_, pid, table_id) in &types { + let section = psi::section(table_id, pid, 0, 0, 0, &[pid as u8; 40]); + mux.section(pid, §ion); + sections.push(section); + } + mux.packet(CLOCK, Flags::pcr(40 * MS)); + + let imported = Imported::all(&mux.bytes()); + for (&(stream_type, pid, _), section) in types.iter().zip(sections) { + let (_, entry) = imported.entry(pid).expect("a section track"); + assert_eq!(entry.verbatim.as_ref().map(|v| v.stream_type), Some(stream_type)); + let frames = imported.frames(pid).await; + assert_eq!(frames.len(), 1, "{stream_type:#04x}"); + assert_eq!(frames[0].payload.as_ref(), section); + let row = imported.row(pid); + assert_eq!((row.units, row.damaged, row.refused), (1, 0, 0), "{stream_type:#04x}"); + } +} + +/// Scrambled packets lose the section they interrupt, even when 16 of them wrap the continuity +/// counter so the clear packet after them looks contiguous with the partial. +#[tokio::test(start_paused = true)] +async fn scrambled_packets_lose_a_partial_section() { + const DSMCC: u16 = 0x110; + let mut mux = Mux::default(); + mux.pat(&[(1, PMT)]).pmt(PMT, 1, CLOCK, &[(0x0b, DSMCC)], 0); + mux.packet(CLOCK, Flags::pcr(0)); + let at = mux.out.len(); + mux.section(DSMCC, &psi::section(0x3b, 1, 0, 0, 0, &[0x11; 300])); + let tail = mux.out.split_off(at + 188); + for cc in 1..=16u8 { + let mut pkt = vec![0x47, (DSMCC >> 8) as u8, DSMCC as u8, 0x90 | (cc & 0x0f)]; + pkt.resize(188, 0xee); + mux.out.extend_from_slice(&pkt); + } + mux.out.extend_from_slice(&tail); + let clear = psi::section(0x3b, 2, 0, 0, 0, &[0x22; 40]); + mux.section(DSMCC, &clear); + mux.packet(CLOCK, Flags::pcr(40 * MS)); + + let imported = Imported::all(&mux.bytes()); + let frames = imported.frames(DSMCC).await; + let payloads: Vec<_> = frames.iter().map(|frame| frame.payload.to_vec()).collect(); + assert_eq!(payloads, [clear]); + let row = imported.row(DSMCC); + assert_eq!((row.units, row.refused, row.damaged), (1, 16, 0)); +} + +/// A 0x86 stream without a CUEI registration is refused once and never PES-parsed, while LATM +/// AAC, which nothing here decodes, rides whole. +#[tokio::test(start_paused = true)] +async fn ambiguous_streams_are_refused_and_latm_rides_whole() { + const DTS: u16 = 0x120; + const LATM: u16 = 0x121; + let mut mux = Mux::default(); + mux.pat(&[(1, PMT)]).pmt(PMT, 1, CLOCK, &[(0x86, DTS), (0x11, LATM)], 0); + mux.packet(CLOCK, Flags::pcr(0)); + let latm = audio(LEAD, &[0x56, 0xe0, 0x10, 0xaa, 0xbb]); + mux.whole(LATM, &latm); + mux.whole(DTS, &audio(LEAD, &[0x7f, 0xfe, 0x80, 0x01])); + mux.packet(CLOCK, Flags::pcr(40 * MS)); + + let imported = Imported::all(&mux.bytes()); + assert!(imported.entry(DTS).is_none()); + let row = imported.row(DTS); + assert_eq!((row.track.as_str(), row.class), ("", Class::Data)); + assert_eq!((row.refused, row.damaged, row.units), (1, 0, 0)); + + let (_, entry) = imported.entry(LATM).expect("LATM is carried"); + assert_eq!(entry.pes_packet.as_ref().map(|pes| pes.stream_type), Some(0x11)); + let frames = imported.frames(LATM).await; + assert_eq!(frames.len(), 1); + assert_eq!(frames[0].payload.as_ref(), latm); +} + +/// Once the program's PMT is in, a packet on a PID nothing claims counts in its own row: the CAT, +/// an ATSC PSIP base PID, and a PID no table lists. Packets before the PAT or the PMT are +/// acquisition, and the null PID, the captured SI and the tables themselves are never counted. +/// A TS-scrambled packet on a carried stream is refused rather than damaged. +#[tokio::test(start_paused = true)] +async fn unclaimed_pids_count_once_the_pmt_is_in() { + const CAT: u16 = 0x0001; + const PSIP: u16 = 0x1ffb; + let mut mux = Mux::default(); + // A mid-stream join: media and strays before the PAT, then before the PMT. + mux.fill(DATA, 2).fill(UNLISTED, 2); + mux.pat(&[(1, PMT)]); + mux.fill(DATA, 1).fill(UNLISTED, 1); + mux.pmt_with(PMT, 1, CLOCK, &[(0x06, DATA, KLVA)], 0); + mux.packet(CLOCK, Flags::pcr(0)); + mux.section(CAT, &psi::section(0x01, 0xffff, 0, 0, 0, &[])); + mux.fill(PSIP, 2).fill(UNLISTED, 3).fill(0x1fff, 2).fill(0x11, 1); + mux.pat(&[(1, PMT)]).pmt_with(PMT, 1, CLOCK, &[(0x06, DATA, KLVA)], 0); + scrambled(&mut mux, DATA); + let pes = klv(0x5a, 64); + mux.whole(DATA, &pes); + mux.packet(CLOCK, Flags::pcr(40 * MS)); + + let imported = Imported::all(&mux.bytes()); + let stats = imported.import.stats(); + let rows: Vec = stats.streams.keys().copied().collect(); + assert_eq!(rows, [CAT, DATA, UNLISTED, PSIP]); + assert_eq!( + stats.streams[&UNLISTED], + Stream { + class: Class::Data, + refused: 3, + ..Default::default() + } + ); + for (pid, refused) in [(CAT, 1), (PSIP, 2)] { + let row = &stats.streams[&pid]; + assert_eq!((row.track.as_str(), row.class, row.refused), ("", Class::Data, refused)); + } + let data = &stats.streams[&DATA]; + assert_eq!( + (data.track.as_str(), data.units, data.refused, data.damaged), + (".ts", 1, 1, 0) + ); + assert_eq!(imported.frames(DATA).await[0].payload.as_ref(), pes); +} + +/// Two programs, the second's media and a stray PID ahead of its PMT. Under `--program all` +/// nothing counts until both PMTs are in, and a stray PID counts once for the multiplex. A +/// `--program 1` import cannot tell program 2's PIDs from strays, so it counts neither. +#[tokio::test(start_paused = true)] +async fn an_mpts_counts_strays_once_both_pmts_are_in() { + const PMT2: u16 = 0x30; + const DATA2: u16 = 0x201; + let mut mux = Mux::default(); + mux.pat(&[(1, PMT), (2, PMT2)]); + mux.pmt_with(PMT, 1, 0x1f1, &[(0x06, DATA, KLVA)], 0); + mux.packet(0x1f1, Flags::pcr(0)).packet(0x1f2, Flags::pcr(0)); + mux.whole(DATA2, &klv(2, 32)); + mux.fill(UNLISTED, 2); + mux.pmt_with(PMT2, 2, 0x1f2, &[(0x06, DATA2, KLVA)], 0); + mux.whole(DATA, &klv(1, 32)); + mux.whole(DATA2, &klv(2, 32)); + mux.fill(UNLISTED, 4); + mux.packet(0x1f1, Flags::pcr(40 * MS)) + .packet(0x1f2, Flags::pcr(40 * MS)); + let data = mux.bytes(); + + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", crate::catalog::Config::default()); + programs.decode(&data).unwrap(); + let stats = programs.stats(); + assert_eq!( + stats.streams[&UNLISTED], + Stream { + class: Class::Data, + refused: 4, + ..Default::default() + } + ); + for pid in [DATA, DATA2] { + assert_eq!(stats.streams[&pid].refused, 0, "{pid:#x}"); + assert_eq!(stats.streams[&pid].track, ".ts", "{pid:#x}"); + } + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut one = Import::new(broadcast, catalog.reserve()).with_program(1); + one.decode(&data).unwrap(); + let rows: Vec = one.stats().streams.keys().copied().collect(); + assert_eq!(rows, [DATA], "only program 1's own PIDs"); +} + +/// A PMT section on another program's PMT PID claims nothing: program 2's PMT arriving on +/// program 1's PID never takes program 2's claim on a stray. +#[tokio::test(start_paused = true)] +async fn a_pmt_on_another_programs_pid_claims_nothing() { + const PMT2: u16 = 0x30; + const DATA2: u16 = 0x201; + let mut mux = Mux::default(); + mux.pat(&[(1, PMT), (2, PMT2)]); + mux.pmt_with(PMT, 1, CLOCK, &[(0x06, DATA, KLVA)], 0); + mux.pmt_with(PMT2, 2, CLOCK, &[(0x06, DATA2, KLVA)], 0); + mux.packet(CLOCK, Flags::pcr(0)); + mux.fill(UNLISTED, 2); + mux.pmt_at(1, PMT, 2, CLOCK, &[(0x06, UNLISTED, KLVA)], 0); + mux.fill(UNLISTED, 3); + + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", crate::catalog::Config::default()); + programs.decode(&mux.bytes()).unwrap(); + assert_eq!(programs.stats().streams[&UNLISTED].refused, 5); +} + +/// A PMT the PAT lists but that never arrives holds the count back for TR 101 290's 0.5 s of +/// program clock, timed on a program whose clock runs, and no longer. +#[tokio::test(start_paused = true)] +async fn a_pmt_that_never_arrives_stops_holding_the_count() { + let mut mux = Mux::default(); + mux.pat(&[(1, PMT), (2, 0x30)]); + mux.pmt_with(PMT, 1, CLOCK, &[(0x06, DATA, KLVA)], 0); + for ms in [0, 200, 400, 500] { + mux.packet(CLOCK, Flags::pcr(ms * MS)).fill(UNLISTED, 1); + } + mux.packet(CLOCK, Flags::pcr(501 * MS)).fill(UNLISTED, 2); + + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", crate::catalog::Config::default()); + programs.decode(&mux.bytes()).unwrap(); + assert_eq!(programs.stats().streams[&UNLISTED].refused, 2); +} + +/// One PID through its whole life, packets flowing throughout: unlisted, added by a PMT +/// revision, removed by the next, then added again. It keeps one row, whose counters never go +/// back, whichever owner holds it: the import while carried, the dropped-PID count otherwise, +/// which `Programs` holds for `--program all`. +#[tokio::test(start_paused = true)] +async fn a_pid_keeps_one_row_through_its_lifecycle() { + let mut steps = Vec::new(); + let mut mux = Mux::default(); + mux.pat(&[(1, PMT)]).pmt_at(0, PMT, 1, CLOCK, &[(0x06, DATA, KLVA)], 0); + mux.packet(CLOCK, Flags::pcr(0)); + mux.fill(UNLISTED, 3); + steps.push(mux.out.len()); + mux.pmt_at(1, PMT, 1, CLOCK, &[(0x06, DATA, KLVA), (0x06, UNLISTED, KLVA)], 0); + mux.packet(CLOCK, Flags::pcr(40 * MS)); + mux.whole(UNLISTED, &klv(1, 32)); + mux.whole(UNLISTED, &klv(2, 32)); + // A later PCR places them. + mux.packet(CLOCK, Flags::pcr(60 * MS)); + steps.push(mux.out.len()); + mux.pmt_at(2, PMT, 1, CLOCK, &[(0x06, DATA, KLVA)], 0); + mux.fill(UNLISTED, 4); + steps.push(mux.out.len()); + mux.pmt_at(3, PMT, 1, CLOCK, &[(0x06, DATA, KLVA), (0x06, UNLISTED, KLVA)], 0); + mux.packet(CLOCK, Flags::pcr(80 * MS)); + mux.whole(UNLISTED, &klv(3, 32)); + mux.packet(CLOCK, Flags::pcr(100 * MS)); + steps.push(mux.out.len()); + let data = mux.bytes(); + // (track, units, refused) after each step. + let expected = [("", 0, 3), (".ts", 2, 3), ("", 2, 7), (".ts", 3, 7)]; + + let mut one = Imported::new(); + let mut programs = Programs::new( + crate::source::produce_origin(), + "event.hang", + crate::catalog::Config::default(), + ); + let mut at = 0; + for (end, (track, units, refused)) in steps.into_iter().zip(expected) { + one.import.decode(&data[at..end]).unwrap(); + programs.decode(&data[at..end]).unwrap(); + at = end; + for (owner, row) in [ + ("import", one.row(UNLISTED)), + ("programs", programs.stats().streams[&UNLISTED].clone()), + ] { + let class = row.class; + assert_eq!( + (row.track.as_str(), row.units, row.refused, row.damaged), + (track, units, refused, 0), + "{owner} at {end}" + ); + assert_eq!(class, Class::Data, "{owner} at {end}"); + } + } +} diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index f0c2424cb47..f54ea6efb8f 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -747,6 +747,8 @@ impl Schedule { /// bytes (the section pointer field) just before it. It /// decodes at `decode`, or each of `frames` at its own time, as [`Self::push_frames`] /// lists them, which only the PID's decoder buffer reads: the unit misses no deadline. + /// It yields to units due before it decodes, so a large one, such as a whole picture, + /// shares the multiplex rather than holding it from placement until its last byte. pub fn push_placed( &mut self, pid: u16, @@ -756,7 +758,10 @@ impl Schedule { keyframe: bool, frames: Vec<(u128, usize)>, ) -> anyhow::Result<()> { - let timing = Timing { by: place, decode }; + let timing = Timing { + by: decode.max(place), + decode, + }; self.queue(pid, timing, Some((place, prefix)), packets, keyframe, frames) } @@ -780,9 +785,15 @@ impl Schedule { holds, leads: pid, }; + // They go out ahead of the first byte of a placed unit, which a decode time well past + // its placement would otherwise let them hold the lane until, past later tables' due. + let lead = Timing { + by: place.map_or(timing.by, |(place, _)| place.min(timing.by)), + ..timing + }; // The unit they lead keeps its own start. let last = self.last; - self.enqueue(pmt, timing, tables.packets, Some(program)); + self.enqueue(pmt, lead, tables.packets, Some(program)); self.last = last; } if packets.is_empty() { @@ -1602,6 +1613,71 @@ mod tests { out } + /// A placed picture five slots long, decoding a second out, shares the multiplex with a + /// unit due while it is still going out, rather than holding the rate until its last byte. + #[test] + fn a_placed_pes_yields_to_units_due_before_it_decodes() { + let mut schedule = Schedule::new(Duration::from_millis(25)); + schedule.set_rate(Some(RATE)); + schedule.set_buffer(1, VIDEO); + let mut picture = unit(7, 200); + picture[1] |= 0x40; + schedule + .push_placed(7, (ms(50), ms(1_000)), 0, picture, true, Vec::new()) + .unwrap(); + schedule.push(1, timing(100), unit(1, 30), true).unwrap(); + let slots = drain(&mut schedule); + let last = slots + .iter() + .filter(|(_, pids, _)| pids.contains_key(&1)) + .map(|&(index, ..)| index) + .max(); + assert!( + last.is_some_and(|last| last <= slot(ms(100))), + "the unit due at 100 ms goes out by slot {last:?}" + ); + let sent: usize = slots.iter().filter_map(|(_, pids, _)| pids.get(&7)).sum(); + assert_eq!(sent, 200); + } + + /// Tables leading a placed picture that decodes a second out go out by its placement, so + /// the tables leading a unit due sooner, which ride the same lane after them, are not held + /// past that unit's deadline. + #[test] + fn tables_leading_a_placed_pes_go_out_by_its_placement() { + let mut schedule = Schedule::new(Duration::from_millis(25)); + schedule.set_rate(Some(RATE)); + schedule.set_tables(100); + schedule.set_clock(1); + schedule.set_buffer(1, VIDEO); + let tables = || Tables { + packets: [unit(0, 1), unit(100, 1)].concat(), + clock: 1, + rate: Some(RATE), + added: Vec::new(), + }; + let mut picture = unit(7, 200); + picture[1] |= 0x40; + schedule.lead(tables()); + schedule + .push_placed(7, (ms(50), ms(1_000)), 0, picture, true, Vec::new()) + .unwrap(); + schedule.lead(tables()); + schedule.push(1, timing(100), unit(1, 30), true).unwrap(); + let slots = drain(&mut schedule); + let last = slots + .iter() + .filter(|(_, pids, _)| pids.contains_key(&1)) + .map(|&(index, ..)| index) + .max(); + assert!( + last.is_some_and(|last| last <= slot(ms(100))), + "the unit due at 100 ms goes out by slot {last:?}" + ); + let tables: usize = slots.iter().filter_map(|(_, pids, _)| pids.get(&100)).sum(); + assert_eq!(tables, 2); + } + /// The pointer byte goes just before placement, so it cannot cost a section another packet. #[test] fn placed_first_byte_accounts_for_the_section_pointer() { diff --git a/rs/moq-mux/src/container/ts/stats.rs b/rs/moq-mux/src/container/ts/stats.rs index 82e575e284e..e59d3c43eef 100644 --- a/rs/moq-mux/src/container/ts/stats.rs +++ b/rs/moq-mux/src/container/ts/stats.rs @@ -50,9 +50,15 @@ pub struct Report { #[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. - /// An SI PID has a `.si` row only once it counts [`Stream::unplaced`] sections. + /// One row per PID, ordered by PID: each elementary stream the PMT lists, carried or + /// refused, and each PID no table claims that the importer dropped packets on. A dropped + /// PID's row names no track, is [`Class::Data`], and counts its packets in + /// [`Stream::refused`]. A PID keeps one row for its life: a PMT that claims it, releases it, + /// or claims it again moves the row with its counters, and while dropped it names no track. + /// Packets before every listed PMT has arrived, or lapsed past 0.5 s, are acquisition and + /// not counted; so are null packets. An import of one program from a multiplex that lists + /// others counts no PID no table claims, since another program's PMT may. An SI PID has a + /// `.si` row only once it counts [`Stream::unplaced`] sections. // Keys read from strings: a flattened section loses JSON's integer-key parsing. #[serde_as(as = "BTreeMap")] #[serde(skip_serializing_if = "BTreeMap::is_empty")] @@ -124,8 +130,8 @@ impl Snapshot { #[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. + /// or empty for a PID that publishes no track: one dropped, or MPEG-1/2 video refused and + /// read only for its clock. 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`]. @@ -167,11 +173,15 @@ pub struct Stream { /// 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. + /// What this PID lost to a route the import does not carry. Once for each time its stream + /// was refused: a listed stream the import cannot carry (an ambiguous 0x86 without a CUEI + /// registration, an Opus configuration it cannot map, any undecoded stream without the + /// `mpegts` catalog section or in a program with no PCR to place it by), or a whole-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, whose track ends and leaves the catalog. Once for each + /// TS-scrambled packet on a carried stream, which the raw lane could carry. And once for + /// each packet on a PID no table claims ([`Snapshot::streams`]). Export stays zero. #[serde(skip_serializing_if = "is_zero")] pub refused: u64, /// This PID's share of [`Snapshot::continuity_count_error`]. @@ -215,10 +225,10 @@ impl Stream { pub enum Class { /// Continuous audio. Audio = 1, - /// Continuous video, including MPEG-1/2 video read only for its clock. + /// Continuous video, including MPEG-1/2 video carried whole or read only for its clock. 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 + /// Undecoded or sparse data (SCTE-35, ID3, private sections, private PES), a PCR PID that + /// carries no elementary stream, or a dropped PID. 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] diff --git a/rs/moq-mux/src/container/ts/unclaimed.rs b/rs/moq-mux/src/container/ts/unclaimed.rs new file mode 100644 index 00000000000..5805c56a2c7 --- /dev/null +++ b/rs/moq-mux/src/container/ts/unclaimed.rs @@ -0,0 +1,58 @@ +//! The packets an import drops on PIDs no table claims. +//! +//! A PID is claimed by the PAT (PID 0 and each listed program's PMT PID), by a route an import +//! keeps (a PMT, an elementary stream, a clock), or by the SI the import captures. The null PID +//! is stuffing, rebuilt on export, and never counted. Each owner decides when counting starts: +//! until every listed PMT is in, a PID may belong to a program whose PMT is still on its way. + +use std::collections::BTreeMap; + +use mpeg2ts::ts::Pid; + +use super::catalog::SI_PIDS; +use super::stats; + +/// TS packets that carry nothing: stuffing. +const NULL_PID: u16 = 0x1fff; + +/// Packets dropped on each PID no table claims, kept across the PID's claims since counters +/// never go backward. The 13-bit PID namespace bounds it. +#[derive(Default)] +pub(super) struct Unclaimed { + dropped: BTreeMap, +} + +impl Unclaimed { + /// Whether `pid` is claimed whatever the PMTs say: the null PID, the PAT, and, when `si` is + /// captured, the standalone SI PIDs. + pub fn fixed(pid: u16, si: bool) -> bool { + pid == NULL_PID || pid == Pid::PAT || (si && SI_PIDS.contains(&pid)) + } + + /// Count a packet dropped on `pid`, warning on its first. + pub fn packet(&mut self, pid: u16) { + let dropped = self.dropped.entry(pid).or_default(); + if *dropped == 0 { + tracing::warn!( + pid, + "dropping a PID no imported table claims; carry the multiplex whole to keep it" + ); + } + *dropped += 1; + } + + /// Fold the dropped packets into `streams`, counted in `refused`. A PID `claimed` says no + /// table claims now names no track; one claimed again keeps its count in its route's row. + pub fn report(&self, streams: &mut BTreeMap, claimed: impl Fn(u16) -> bool) { + for (&pid, &dropped) in &self.dropped { + let row = streams + .entry(pid) + .or_insert_with(|| stats::Stream::new("", stats::Class::Data)); + row.refused += dropped; + if !claimed(pid) { + row.track.clear(); + row.class = stats::Class::Data; + } + } + } +} diff --git a/test/justfile b/test/justfile index b0544b043af..b71f700909b 100644 --- a/test/justfile +++ b/test/justfile @@ -176,7 +176,8 @@ drill-sensitivity *args: # (use a real capture), `--live` (grade PCR release timing and byte position off # the pipe instead, which a capture cannot carry; nightly runs this arm), # `--open-gop` (an open-GOP clip whose orphan leading pictures are trimmed), `--hrd` -# (1080p video filling a broadcast-sized CPB), `--delay 2s` (the exporter's). See +# (1080p video filling a broadcast-sized CPB), `--mpeg2` (MPEG-2 video carried whole, +# decoded from the capture), `--delay 2s` (the exporter's). See # ts/README.md. # Check the subscriber's MPEG-TS output for IRD compliance. diff --git a/test/ts/README.md b/test/ts/README.md index 9e68b47f684..efdb7afaf97 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -44,6 +44,7 @@ just test ts --leak # 1080p video with no HRD, filling its level's 2 just test ts --full # the same with two audio PIDs, filling its multiplex with no nulls just test ts --fragment-pes # legal headers split across packets, with FFmpeg/TSDuck grading just test ts --klv # KLV and DVB AC-3 carried as whole PES, graded against the source +just test ts --mpeg2 # MPEG-2 video carried as whole PES, decoded from the capture just test ts --revision # the PMT revised mid-stream, a KLV stream joining and leaving just test ts --delay 2s # pass the exporter's --delay just test ts --hrd --rate 7000000 # congest the subscriber's leg below the media rate @@ -69,6 +70,11 @@ source placement on the program clock, and none after its PTS; and FFmpeg readin each as KLV with the same PES and PTS. `holes.py` and the T-STD model grade the AC-3, whose buffer drains each sync frame at its own decode time. +`--mpeg2` publishes Main Profile at Main Level MPEG-2 video beside MPEG audio. The +importer does not decode MPEG-2 video, so it rides its track whole while its PTS +still paces the program. `mpeg2.py` decodes source and capture with FFmpeg: past +the GOP the subscriber joins in, every picture is the source's next one, through +the source's last. `holes.py` and the T-STD model grade the rest. `--revision` has `revision.py` add asynchronous KLV to the default clip's null slots from a third of its program clock to two thirds, and revise the PMT at each end: the next `version_number` lists the stream before its first packet, and the @@ -155,8 +161,8 @@ decoder (2.4.2; Rec. ITU-T H.222.0, whose 10/2014 edition is a free download), fed on the stream's own PCR clock: ```text -video (AVC 2.14.3.1, HEVC 2.17.2) TB --Rx--> MB --Rbx (leak)--> EB --DTS--> decoder -audio (2.4.2.3) TB --Rx--> B ----------------PTS--> decoder +video (H.262 2.4.2.3, AVC 2.14.3.1, HEVC 2.17.2) TB --Rx--> MB --Rbx (leak)--> EB --DTS--> decoder +audio (2.4.2.3) TB --Rx--> B ----------------PTS--> decoder ``` It fails a stream where TB, MB, EB or B overflows, where TB stays occupied for a @@ -167,7 +173,7 @@ underflowed access unit finished arriving, and the longest any access unit waite TSDuck supplies the transport analysis, not this decoder-buffer model. Commercial analyzers advertise T-STD checks; no suitable maintained open-source replacement -was established for this harness. Its parameters are transcribed from the specs: H.264 Table A-1 and H.265 Table A.8 for the level, the +was established for this harness. Its parameters are transcribed from the specs: H.262 Tables 8-13 and 8-14, H.264 Table A-1 and H.265 Table A.8 for the level, the ADTS and "other audio" rates and sizes in H.222.0 2.4.2.3, and ATSC A/52 and A/53 Part 5 for AC-3 and E-AC-3. TSDuck does the parsing: `tstables` decodes the PMT and `tsp -P pes --avc-access-unit` the SPS, AVC and HEVC alike. @@ -182,6 +188,14 @@ level defaults too. An `AVC_timing_and_HRD_descriptor` or transfer to the HRD's own schedule, which is not modelled, so such a stream is refused. +MPEG-2 video takes Rx = 1.2 Rmax from its level, Rbx = Rmax at Low and Main level +and min(1.05 Res, Rmax) at High-1440 and High, EB from the sequence header's +`vbv_buffer_size`, and MB from what the level's VBV leaves over, as H.222.0 2.4.2.3 +specifies for the leak method. A stream whose `STD_descriptor` clears `leak_valid` +while its pictures code a `vbv_delay` other than 0xFFFF takes the `vbv_delay` +method instead, which is not modelled, so it is refused. Its picture, GOP or +sequence header opens each access unit. + Opus is graded against ADTS's buffers for the same channel count: the Opus-in-TS draft gives Rx (2 Mb/s for 1-2 channels, matching ADTS) but leaves the buffer size unset, so that size is borrowed rather than specified. A stream with no parameters @@ -915,7 +929,7 @@ browser interop. It runs nightly on `main`, and release PRs touching a workspace crate `run.sh` builds (kept in sync by `sh/gh/ts-paths.py`), the harness, or the toolchain run it too: `just test ts-controls`, `just test ts-pair-controls`, the default and 2 Mb/s round trips, HRD, leak, full, headroom, -open-GOP, KLV (strict), PMT revision, passthrough (generated, Kyrion and MPTS), EIT, and T-STD controls. Each step runs even after another test fails. +open-GOP, KLV (strict), MPEG-2 video, PMT revision, passthrough (generated, Kyrion and MPTS), EIT, and T-STD controls. Each step runs even after another test fails. All recipes use the same pinned Nix FFmpeg and TSDuck locally and in CI. Nightly and manual runs also invoke `just test ts --pair`, diff --git a/test/ts/compliance.py b/test/ts/compliance.py index 11c048b2fbe..549928768f5 100755 --- a/test/ts/compliance.py +++ b/test/ts/compliance.py @@ -438,6 +438,11 @@ def check_service_descriptors(analysis: dict) -> Check: # H.265 Table A.9 CpbNalFactor for Main, Main 10 and Main Still Picture (profile_idc 1-3). HEVC_NAL_FACTOR = 1100 +# H.262 Tables 8-13 and 8-14: level (the low nibble of profile_and_level_indication) -> +# (Rmax bit/s, VBV_max bits), the same for every profile that has the level. +MPEG2_LEVELS = {10: (4_000_000, 475_136), 8: (15_000_000, 1_835_008), 6: (60_000_000, 7_340_032), 4: (80_000_000, 9_781_248)} +MPEG2_LEVEL_NAMES = {10: "Low", 8: "Main", 6: "High-1440", 4: "High"} + # H.222.0 2.4.2.3: ADTS (Rx, BSn) by channels, the LFE not counted. ADTS_BUFFERS = ((2, 2_000_000, 3584), (8, 5_529_600, 8976), (12, 8_294_400, 12804), (48, 33_177_600, 51216)) # channel_configuration -> full-bandwidth channels (5.1 and 7.1 drop the LFE). @@ -464,6 +469,7 @@ def check_service_descriptors(analysis: dict) -> Check: } # stream_type -> model, for the types identified by stream_type alone. STREAM_KINDS = { + 0x02: "mpeg2-video", 0x1B: "avc", 0x24: "hevc", 0x0F: "adts", @@ -558,6 +564,7 @@ def read_programs(ts_path: str) -> list[tuple[int, list[Stream]]]: names = {d["#name"] for d in descriptors} names |= {f"registration:{d['format_identifier']}" for d in descriptors if "format_identifier" in d} names |= {"hrd_management_valid" for d in descriptors if d.get("hrd_management_valid") is True} + names |= {"leak_invalid" for d in descriptors if d["#name"] == "STD_descriptor" and d.get("leak_valid") is False} names |= {"temporal_poc_clock" for d in descriptors if d.get("fixed_frame_rate") is True and d.get("temporal_poc") is True} names |= { @@ -666,6 +673,54 @@ def _hrd(sps: dict[str, str], prefix: str, value: str) -> tuple[float, float] | return rate, size +def mpeg2_params(es: bytes, descriptors: set[str]) -> Params: + """H.222.0 2.4.2.3 buffers for H.262 video from its first sequence header and extension. + + Graded by the leak method, which H.222.0 applies unless an STD_descriptor clears leak_valid + and the pictures code a vbv_delay other than 0xFFFF; that vbv_delay method is refused. + Rbx is Rmax at Low and Main level, and min(1.05 Res, Rmax) at High-1440 and High. + """ + head = es.find(b"\x00\x00\x01\xb3") + ext = es.find(b"\x00\x00\x01\xb5", head) + picture = es.find(b"\x00\x00\x01\x00", head) + if head < 0 or ext < 0 or len(es) < ext + 9 or es[ext + 4] >> 4 != 1: + raise Refused("no sequence header and sequence extension, which an H.262 stream starts with") + if picture < 0 or len(es) < picture + 8: + raise Refused("no picture header after the sequence header") + # vbv_delay: 16 bits after the 10-bit temporal_reference and 3-bit picture_coding_type. H.262 + # codes it 0xFFFF in every picture of a sequence or in none, so the first decides. + vbv_delay = (es[picture + 5] & 0x07) << 13 | es[picture + 6] << 5 | es[picture + 7] >> 3 + if "leak_invalid" in descriptors and vbv_delay != 0xFFFF: + raise Refused("an STD_descriptor selects the vbv_delay method, which this model does not implement") + # vbv_buffer_size_value: 10 bits after bit_rate_value and its marker; the extension's 8 bits + # follow its bit_rate_extension and marker. + vbv_low = (es[head + 10] & 0x1F) << 5 | es[head + 11] >> 3 + vbv_high = es[ext + 8] + # profile_and_level_indication follows the 4-bit extension_start_code_identifier. + indication = (es[ext + 4] & 0x0F) << 4 | es[ext + 5] >> 4 + level = indication & 0x0F + if indication & 0x80 or level not in MPEG2_LEVELS: + raise Refused(f"H.262 profile_and_level_indication {indication:#04x} has no Table 8-13 entry") + rmax, vbv_max = MPEG2_LEVELS[level] + vbv = (vbv_high << 10 | vbv_low) * 16 * 1024 + if not vbv: + raise Refused("H.262 vbv_buffer_size 0 sizes no EB") + # bit_rate_value's 18 bits after frame_rate_code, and the extension's 12-bit + # bit_rate_extension above them, in units of 400 bit/s. + bit_rate = es[head + 8] << 10 | es[head + 9] << 2 | es[head + 10] >> 6 + bit_rate |= ((es[ext + 6] & 0x1F) << 7 | es[ext + 7] >> 1) << 18 + rbx = rmax if level in (10, 8) else min(1.05 * bit_rate * 400, rmax) + overhead = (0.004 + 1 / 750) * max(rmax, 2_000_000) + return Params( + label=f"H.262 {MPEG2_LEVEL_NAMES[level]} level, vbv {vbv / 1e3:.0f} kbit", + # Rx = 1.2 * Rmax; MBS = BSmux + BSoh + VBV_max - vbv_buffer_size; EBS = vbv_buffer_size. + rx=1.2 * rmax, + mb=(overhead + vbv_max - vbv) / 8, + eb=vbv / 8, + rbx=rbx, + ) + + def avc_params(sps: dict[str, str]) -> Params: """H.222.0 2.14.3.1 buffers from the SPS: level limits, overridden by a declared NAL HRD. @@ -839,7 +894,7 @@ def stream_kind(stream: Stream) -> str | None: return "opus" if kind == 0x06 and tags & {"DVB_enhanced_AC3_descriptor", "enhanced_AC3_descriptor"}: return "eac3-dvb" - if kind in (0x01, 0x02, 0x10, 0x11, 0x1C, 0x20, 0x21, 0x42, 0xD1, 0xEA): + if kind in (0x01, 0x10, 0x11, 0x1C, 0x20, 0x21, 0x42, 0xD1, 0xEA): raise Refused(f"stream_type 0x{kind:02X} is audio/video this model has no parameters for") return None @@ -849,6 +904,8 @@ def stream_params(kind: str, stream: Stream) -> tuple[Params, object]: es = stream.es if kind == "avc": return avc_params(stream.sps), None + if kind == "mpeg2-video": + return mpeg2_params(bytes(es), stream.descriptors), None if kind == "hevc": return hevc_params(stream.sps), None if kind == "teletext": @@ -883,9 +940,16 @@ def stream_params(kind: str, stream: Stream) -> tuple[Params, object]: # NAL unit types that open a new access unit when they follow the last VCL NAL unit # of the previous one (H.264 7.4.1.2.3, H.265 7.4.2.4.4), and the VCL types. -AU_OPENERS = {"avc": {6, 7, 8, 9, 14, 15, 16, 17, 18}, "hevc": {32, 33, 34, 35, 39, 41, 42, 43, 44, *range(48, 56)}} -VCL = {"avc": set(range(1, 6)), "hevc": set(range(0, 32))} -DELIMITER = {"avc": 9, "hevc": 35} +# H.262 6.2.1 has no NAL units: a picture opens at its sequence, GOP or picture header after +# the last slice of the one before, keyed by the start code value. +AU_OPENERS = { + "avc": {6, 7, 8, 9, 14, 15, 16, 17, 18}, + "hevc": {32, 33, 34, 35, 39, 41, 42, 43, 44, *range(48, 56)}, + "mpeg2-video": {0xB3, 0xB8, 0x00}, +} +VCL = {"avc": set(range(1, 6)), "hevc": set(range(0, 32)), "mpeg2-video": set(range(0x01, 0xB0))} +# H.262 has no access unit delimiter, and needs none: its picture header is one. +DELIMITER = {"avc": 9, "hevc": 35, "mpeg2-video": 0x00} def proportional_video_clock(stream: Stream, kind: str, count: int) -> list[tuple[int, Fraction, bool, tuple[int, Fraction] | None]]: @@ -1016,7 +1080,7 @@ def video_units(stream: Stream, kind: str) -> list[AccessUnit]: after_vcl = True # the first opener starts the first whole access unit pos = es.find(b"\x00\x00\x01") while 0 <= pos < len(es) - 3: - nal = (es[pos + 3] >> 1) & 0x3F if kind == "hevc" else es[pos + 3] & 0x1F + nal = {"hevc": (es[pos + 3] >> 1) & 0x3F, "mpeg2-video": es[pos + 3]}.get(kind, es[pos + 3] & 0x1F) delimited |= nal == DELIMITER[kind] if nal in AU_OPENERS[kind] and after_vcl: # The zero_byte ahead of the start code belongs to this NAL unit (H.264 B.1.1, diff --git a/test/ts/compliance.test.py b/test/ts/compliance.test.py index d0052b78ec5..d3ba2e47080 100644 --- a/test/ts/compliance.test.py +++ b/test/ts/compliance.test.py @@ -62,6 +62,38 @@ def tables(*pmts: tuple[int, list[int]]) -> bytes: return b"".join(out) +def h262(indication: int, rate: int, vbv: int, vbv_delay: int) -> bytes: + """A sequence header, sequence extension and picture header: profile_and_level_indication, + bit_rate in bit/s, vbv_buffer_size in 16 kbit units, and the picture's vbv_delay.""" + value = rate // 400 + head = b"\x00\x00\x01\xb3\x2d\x02\x40\x23" + head += ((value & 0x3FFFF) << 14 | 1 << 13 | (vbv & 0x3FF) << 3).to_bytes(4, "big") + ext = b"\x00\x00\x01\xb5" + ext += (1 << 44 | indication << 36 | 1 << 35 | 1 << 33 | (value >> 18) << 17 | 1 << 16 | (vbv >> 10) << 8).to_bytes(6, "big") + picture = b"\x00\x00\x01\x00" + (1 << 19 | vbv_delay << 3).to_bytes(4, "big") + return head + ext + picture + + +class Mpeg2Params(unittest.TestCase): + def test_rbx_follows_the_level(self): + # Main Profile at Main Level leaks at Rmax; at High Level, at 1.05 times the stream's rate. + for indication, rbx in ((0x48, 15_000_000), (0x44, 21_000_000)): + with self.subTest(indication=indication): + params = compliance.mpeg2_params(h262(indication, 20_000_000, 112, 0xFFFF), set()) + self.assertAlmostEqual(params.rbx, rbx) + + def test_zero_vbv_buffer_is_refused(self): + with self.assertRaises(compliance.Refused): + compliance.mpeg2_params(h262(0x48, 4_000_000, 0, 0xFFFF), set()) + + def test_vbv_delay_method_is_refused(self): + es = h262(0x48, 4_000_000, 112, 0x1000) + compliance.mpeg2_params(es, set()) + compliance.mpeg2_params(h262(0x48, 4_000_000, 112, 0xFFFF), {"leak_invalid"}) + with self.assertRaises(compliance.Refused): + compliance.mpeg2_params(es, {"leak_invalid"}) + + class PmtVersions(unittest.TestCase): def check(self, *pmts: tuple[int, list[int]]) -> compliance.Check: with tempfile.NamedTemporaryFile(suffix=".ts") as handle: diff --git a/test/ts/mpeg2.py b/test/ts/mpeg2.py new file mode 100644 index 00000000000..d22503ffef5 --- /dev/null +++ b/test/ts/mpeg2.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Grade a round trip of MPEG-2 video, which the importer carries as whole PES, by decoding it. + + ./mpeg2.py source.ts capture.ts + +The importer does not decode MPEG-2 video: it carries each PES whole and the exporter writes it +back unchanged but for its PTS and DTS. So FFmpeg must read the capture's video as MPEG-2, decode +it without an error, and produce the source's pictures: once its first whole GOP starts, every +decoded picture is the source's next one, through the source's last. The subscriber joins at the +relay's latest group, so the run may start up to holes.JOIN into the source, and the pictures of +the GOP it joins in, whose references never arrived, may decode to garbage or not at all. + +Exit status is 0 when every check passes, 1 otherwise. +""" + +import argparse +import json +import subprocess +import sys +from fractions import Fraction +from pathlib import Path + +import holes + + +def codecs(path): + """The codec of each audio and video stream FFmpeg finds.""" + report = json.loads(subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", "stream=codec_type,codec_name", "-of", "json", str(path)], + check=True, capture_output=True, text=True, + ).stdout) + return sorted((s["codec_type"], s["codec_name"]) for s in report["streams"] if s["codec_type"] in ("audio", "video")) + + +def pictures(path): + """The MD5 of each decoded picture, and FFmpeg's decode errors.""" + result = subprocess.run( + ["ffmpeg", "-v", "error", "-i", str(path), "-map", "0:v:0", "-f", "framemd5", "-"], + capture_output=True, text=True, check=False, + ) + if result.returncode: + raise SystemExit(f"error: ffmpeg could not decode {path}: {result.stderr.strip()}") + digests = [line.rsplit(",", 1)[1].strip() for line in result.stdout.splitlines() if line and not line.startswith("#")] + return digests, [line for line in result.stderr.splitlines() if line.strip()] + + +def grade(source, capture, gop, rate): + """Raise SystemExit naming the first miss; return a summary otherwise.""" + if codecs(capture) != codecs(source): + raise SystemExit(f"error: the capture's streams are {codecs(capture)}, the source's {codecs(source)}") + ours, errors = pictures(source) + if errors: + raise SystemExit(f"error: the source does not decode cleanly: {errors[:3]}") + theirs, errors = pictures(capture) + known = set(ours) + # The pictures of the joined GOP before its next I picture may not be the source's. + lead = next((j for j, digest in enumerate(theirs) if digest in known), None) + if lead is None: + raise SystemExit("error: no decoded picture of the capture is the source's") + if lead > gop: + raise SystemExit(f"error: the first {lead} pictures are not the source's, past one {gop}-picture GOP") + run = theirs[lead:] + # A picture can repeat in the source, so align the run where it matches, preferring the + # alignment that ends on the source's last picture. + starts = [k for k, digest in enumerate(ours) if digest == run[0]] + matches = [k for k in starts if ours[k : k + len(run)] == run] + first = next((k for k in matches if k + len(run) == len(ours)), matches[0] if matches else starts[0]) + if run != ours[first : first + len(run)]: + miss = next(j for j, digest in enumerate(run) if first + j >= len(ours) or ours[first + j] != digest) + raise SystemExit(f"error: decoded picture {lead + miss} is not the source's picture {first + miss}") + if first + len(run) != len(ours): + raise SystemExit(f"error: the capture ends at picture {first + len(run)} of the source's {len(ours)}") + if Fraction(first, rate) > holes.JOIN: + raise SystemExit(f"error: the capture starts at picture {first}, past the {float(holes.JOIN)}s join") + # Damage past the join already shows as a picture that is not the source's, so an error is + # only allowed for missing references at the join. + if errors and not lead: + raise SystemExit(f"error: FFmpeg reports decode errors on a capture that joined on a whole GOP: {errors[:3]}") + return {"pictures": len(run), "source": len(ours), "first": first, "joined": lead} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("source", type=Path) + parser.add_argument("capture", type=Path) + parser.add_argument("--gop", type=int, default=25, help="pictures per GOP in the source") + parser.add_argument("--rate", type=int, default=25, help="pictures per second in the source") + args = parser.parse_args() + report = grade(args.source, args.capture, args.gop, args.rate) + print(f"MPEG-2 video: {report['pictures']} of {report['source']} pictures decoded as the source's, " + f"from picture {report['first']} ({report['joined']} joined pictures skipped)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test/ts/run.sh b/test/ts/run.sh index 1913d9fd9cb..bbcb4ab1849 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -26,6 +26,7 @@ # ./run.sh --fragment-pes # split generated clear PES headers across packets # ./run.sh --clockless sections # sections, pes, or audio-scte35 without video # ./run.sh --klv # add KLV and DVB AC-3, carried as whole PES, and grade them +# ./run.sh --mpeg2 # MPEG-2 video, carried as whole PES, graded by decoding it # ./run.sh --revision # revise the PMT mid-stream, adding then dropping a KLV stream # ./run.sh --delay 2s # pass the exporter's --delay # ./run.sh --rate 7000000 # bufferbloat: 7 Mb/s to the subscriber behind a deep queue (moq-shaper) @@ -92,6 +93,7 @@ FULL="" # the same with audio, in a multiplex the source fills with no n FRAGMENT_PES="" # use following null slots for fragmented PES headers CLOCKLESS="" # generated programs clocked by audio or a dedicated PCR-only PID KLV="" # inject asynchronous and synchronous KLV beside DVB AC-3, and grade every PES +MPEG2="" # publish MPEG-2 video, which the importer carries whole, and decode the capture REVISION="" # revise the PMT twice mid-stream, and grade the export's revisions LOSS="" # percent of relay -> subscriber datagrams dropped at random by moq-shaper RATE="" # bits/s moq-shaper carries from the relay to the subscriber, behind QUEUE @@ -190,6 +192,10 @@ while [[ $# -gt 0 ]]; do KLV=1 shift ;; + --mpeg2) + MPEG2=1 + shift + ;; --revision) REVISION=1 shift @@ -296,18 +302,23 @@ if [[ -n "$KLV" && -n "$SOURCE$ANALYZE_ONLY$HRD$HEADROOM$LEAK$FULL$OPEN_GOP$FRAG echo "error: --klv generates its own clip and grades the demultiplexed export's PES" >&2 exit 1 fi -if [[ -n "$REVISION" && -n "$SOURCE$ANALYZE_ONLY$HRD$HEADROOM$LEAK$FULL$OPEN_GOP$FRAGMENT_PES$PAIR$LIVE$PASSTHROUGH$MPTS$KLV$CLOCKLESS" ]]; then +if [[ -n "$REVISION" && -n "$SOURCE$ANALYZE_ONLY$HRD$HEADROOM$LEAK$FULL$OPEN_GOP$FRAGMENT_PES$PAIR$LIVE$PASSTHROUGH$MPTS$KLV$CLOCKLESS$MPEG2" ]]; then echo "error: --revision revises the generated closed-GOP clip and grades the demultiplexed export's PMT" >&2 exit 1 fi +if [[ -n "$MPEG2" && -n "$SOURCE$ANALYZE_ONLY$HRD$HEADROOM$LEAK$FULL$OPEN_GOP$FRAGMENT_PES$PAIR$LIVE$PASSTHROUGH$MPTS$KLV" ]]; then + echo "error: --mpeg2 generates its own clip and decodes the demultiplexed export" >&2 + exit 1 +fi + if [[ -n "$CLOCKLESS" ]]; then case "$CLOCKLESS" in audio-scte35 | sections | pes) ;; *) echo "error: unsupported --clockless profile: $CLOCKLESS" >&2 exit 1 ;; esac - if [[ -n "$SOURCE$ANALYZE_ONLY$HRD$HEADROOM$LEAK$FULL$OPEN_GOP$FRAGMENT_PES$PAIR$LIVE$PASSTHROUGH$MPTS$KLV$RATE$LOSS" ]]; then + if [[ -n "$SOURCE$ANALYZE_ONLY$HRD$HEADROOM$LEAK$FULL$OPEN_GOP$FRAGMENT_PES$PAIR$LIVE$PASSTHROUGH$MPTS$KLV$MPEG2$RATE$LOSS" ]]; then echo "error: --clockless generates and grades an unimpaired program without video" >&2 exit 1 fi @@ -561,6 +572,17 @@ elif [[ -n "$KLV" ]]; then -c:a:0 aac -b:a:0 128k -c:a:1 ac3 -b:a:1 192k \ -f mpegts -mpegts_flags system_b -muxrate "$BITRATE" -pcr_period 20 "$HARNESS_RUN/media.ts" python3 "$DIR/klv.py" inject "$HARNESS_RUN/media.ts" "$SRC_TS" --report "$HARNESS_RUN/klv-injected.json" +elif [[ -n "$MPEG2" ]]; then + echo "### generating ~${DURATION}s MPEG-2 video and MPEG audio clip" + # Main Profile at Main Level with its whole VBV buffer, the shape of an SD contribution + # feed. Variable rate, so the encoder codes vbv_delay 0xFFFF and the T-STD leak method applies. + ffmpeg -y -hide_banner -loglevel error \ + -f lavfi -i "testsrc=size=720x576:rate=25" \ + -f lavfi -i "sine=frequency=1000:sample_rate=48000" \ + -t "$DURATION" \ + -c:v mpeg2video -g 25 -bf 2 -b:v 6M -maxrate 8M -bufsize 1835008 \ + -c:a mp2 -b:a 192k \ + -f mpegts -muxrate "$BITRATE" -pcr_period 20 "$SRC_TS" else echo "### generating ~${DURATION}s broadcast-like ${OPEN_GOP:+open-GOP }clip with ffmpeg" # CBR with a 20 ms PCR, like a contribution feed. Not cosmetic: `regulate` @@ -914,7 +936,7 @@ fi # pair. Other arms retain their specific structural/model oracles until their # acquisition and decoder-priming contracts are covered by fixtures. A path # narrower than the media sheds groups, which only holes.py grades. -if [[ -z "$SOURCE$OPEN_GOP$HRD$HEADROOM$LEAK$FULL$KLV$CLOCKLESS$RATE" ]]; then +if [[ -z "$SOURCE$OPEN_GOP$HRD$HEADROOM$LEAK$FULL$KLV$MPEG2$CLOCKLESS$RATE" ]]; then if ! python3 "$DIR/decode.py" "$SRC_TS" "$SUB_TS" --output "$HARNESS_RUN/decode"; then # The exporter counts what it dropped (late), trimmed (before its output began), # and lost upstream (gaps), and logs each loss, so the failure says which it was. @@ -933,7 +955,7 @@ fi # grade only the access units that arrived. Under --rate the path is narrower than the media, # so the export has to shed: only whole groups, and never the last, which a stall or a # latency that grew with the queue would cost. -if [[ -n "$HRD$HEADROOM$LEAK$FULL$KLV" || (-n "$RATE" && -z "$SOURCE") ]]; then +if [[ -n "$HRD$HEADROOM$LEAK$FULL$KLV$MPEG2" || (-n "$RATE" && -z "$SOURCE") ]]; then echo if ! python3 "$DIR/holes.py" "$SUB_TS" "$SRC_TS" ${RATE:+--shed}; then echo >&2 @@ -990,6 +1012,16 @@ if [[ -n "$KLV" ]]; then fi fi +if [[ -n "$MPEG2" ]]; then + echo + if ! python3 "$DIR/mpeg2.py" "$SRC_TS" "$SUB_TS"; then + echo >&2 + echo "error: the MPEG-2 video did not survive the round trip (see round-trip logs below)" >&2 + dump_logs + exit 1 + fi +fi + # The export announces each layout the source's PMT revised to, at the next version. if [[ -n "$REVISION" ]]; then echo