diff --git a/doc/bin/cli.md b/doc/bin/cli.md index d236b883323..45cfa0c072d 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -104,11 +104,22 @@ profile or channel layout ADTS cannot label is refused rather than mislabeled. A constant-rate MPEG-TS source records its multiplex rate in the catalog (`mpegts.muxRate`, measured off the PCR clock, null stuffing included), and -`export ts` pads its output with null packets back to that rate so an IRD or -groomer receives a constant-rate stream. `--mux-rate 5000000` pads to an explicit -rate instead, including for a broadcast that recorded none. Media is never delayed -or dropped to fit: a source that sustains more than the rate overruns it, and a -VBR source records nothing, so export without either stays unpadded. +`export ts` transmits its output at that rate: every 25 ms between clock packets +carries the bytes the rate implies, media first and null packets after, so an IRD +or groomer recovering the clock from packet arrival can lock. `--mux-rate 5000000` +sets an explicit rate instead, including for a broadcast that recorded none. A VBR +source records nothing, so export without either stays VBR and unpadded. The +importer measures the rate over its first 2 s, and an export that starts with the +broadcast is VBR until then. + +A keyframe is usually several intervals' worth of bytes, so it is spread over the +intervals before it decodes, and the output trails the media by a buffer delay. +The delay grows to fit the largest burst seen, up to `--max-age`, and is latency +on top of the export's own. A burst that needs more goes out above the rate before +it decodes, with a warning, and later intervals repay it from their padding: media +is never delayed past its decode time or dropped. A 60 s cut of a 9.95 Mb/s news +channel reaches the 500 ms default and has 75 % of intervals on the rate; +`--max-age 1s` settles at a 710 ms delay with 99 % on it. fMP4 export writes one fragment per publisher group on each track. Audio follows the publisher's cuts; video normally follows GOPs. Closing a group flushes it @@ -328,7 +339,9 @@ For `export ts`, `--max-age` also bounds how long the muxer holds a leading track for a lagging one. Frames go out in media-time order across all tracks, not arrival order, so two exporters of one broadcast emit them in one order. A track quiet for longer is muxed around until it catches up; a sparse track -(SCTE-35) costs that wait once per cue. `--max-age 0` keeps arrival order. +(SCTE-35) costs that wait once per cue. `--max-age 0` keeps arrival order. With +a mux rate it is also how far the output may trail the media to spread a keyframe +over the rate. A stdout export ends with the broadcast. `export ts --linger 10s` waits that long for the broadcast to come back instead: a publisher that restarts within diff --git a/quest/m1/README.md b/quest/m1/README.md index 12f1f067c0a..1ee815c462d 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -135,7 +135,6 @@ QUIC studies there on that rule. - [Catalog warmup](/quest/m1/catalog-warmup.md) - `warmup` on video and audio renditions, in the catalog and the draft - [Audio warmup](/quest/m1/audio-warmup.md) - a viewer joining an Opus rendition mid-stream never hears the unconverged first 80 ms - [#3021](/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md) - moq-gst picks the broadcast wall epoch; a restarted source is a new epoch, not a forward re-anchor -- [TS byte schedule](/quest/m1/ts-export-byte-schedule.md) - moq export ts places PCRs and padding on the byte grid `mpegts.muxRate` implies, so a receiver can clock off arrival - [Release profile](/quest/m1/release-profile.md) - every release build gets fat LTO, one codegen unit, and stripping from the workspace profile instead of three script exports - [Size report](/quest/m1/size-report.md) - a nightly job reports every shipped artifact's size, native and JS, and alerts when one grows - [Publish lazy file source](/quest/m1/publish-lazy-file.md) - a camera or screen `` stops downloading mediabunny's ~99 KB gzip diff --git a/quest/m1/ts-export-byte-schedule.md b/quest/m1/ts-export-byte-schedule.md deleted file mode 100644 index 1130acbd2d2..00000000000 --- a/quest/m1/ts-export-byte-schedule.md +++ /dev/null @@ -1,30 +0,0 @@ -# [L] moq export ts: PCRs sit on the byte grid the mux rate implies - -## Goal - -When the catalog carries `mpegts.muxRate`, `moq export ts` emits a -constant-rate stream: the bytes between consecutive PCRs are what the declared -rate implies for that interval, so a receiver recovering its clock from packet -arrival can lock. Today the average rate is right (#3831) but the bytes clump: -188 B to 870 KB between PCRs where 31 KB is needed, and only 3.3 % of intervals -within 1 % of nominal (#3925). Streams without a mux rate stay VBR as today. A -UDP sink is out of scope; delivery stays with an external tool. - -## Plan - -- Padding and PCR placement become one scheduling decision in - `rs/moq-mux/src/container/ts/export.rs` (`advance`, `emit`, `stuff`): a PCR's - value follows its byte position at the mux rate, and a keyframe burst is - spread over the slots before its DTS instead of landing between two PCRs. -- That makes the mux run ahead of decode by a buffer delay. The delay grows to - fit the largest burst seen and is capped by `--max-age`, which already bounds - the pacer's lead; no new flag. Say what happens when a burst does not fit the - cap (fail loud, or drop to VBR with a warning) and record the choice here. -- The CLI `Delivery` pacer releases one slice per PCR interval; confirm it - still writes on the schedule the PCRs describe. -- `test/ts/pcr-timing.py`'s `pcr-schedule` grades the bytes between - consecutive PCRs, report-only in the TS recipe. Once the schedule lands, gate - it there with `--schedule-pct-min`, against a fixture whose keyframes outgrow - a PCR slot; the generated clip mostly pads and passes already. -- `doc/bin/cli.md`: say that export pads to `mpegts.muxRate` on a constant-rate - schedule and what latency that adds. diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 733a4eaff1b..92815cb1cdd 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -973,9 +973,10 @@ pub struct Transport { #[usage(flatten)] pub container: Container, - /// Pad the output with null packets to this constant rate, in bits per second. + /// Transmit the output at this constant rate, in bits per second, padding with + /// null packets and trailing the media by up to `--max-age` to spread keyframes. /// Defaults to the multiplex rate the catalog recorded from a constant-rate - /// source (`mpegts.muxRate`); without either the output is unpadded. + /// source (`mpegts.muxRate`); without either the output is VBR and unpadded. #[usage(long)] pub mux_rate: Option, } diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index e07f1b0d39e..90c4dc2a59b 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -7,7 +7,9 @@ //! each [`Frame`] is one slot's clock packet plus the bytes belonging to it, //! stamped at the slot boundary, so the clock a receiver recovers from byte //! position agrees with the values, and a pacing caller releases each slot at -//! the instant it asserts. Video is carried as Annex-B, audio as ADTS AAC. +//! the instant it asserts. With a multiplex rate every slot carries exactly the +//! bytes that rate implies, media first and null stuffing after +//! ([`Export::transmit`]). Video is carried as Annex-B, audio as ADTS AAC. //! //! Video flows through [`ExportSource`], which normalizes every H.264/H.265 //! source to length-prefixed NALU plus a resolved avcC/hvcC (parsing in-band @@ -179,12 +181,12 @@ pub struct Export { stall: Option, /// Wakes [`Self::pick_next_track`] once the stall has lasted `max_age`. hold: Option>>, - /// The rate to pad the output to with null packets, in bits per second: the - /// builder override when set, else the catalog's recorded multiplex rate, else - /// none and the output is unpadded ([`Self::stuff`]). + /// The constant rate to transmit the output at, in bits per second: the builder + /// override when set, else the catalog's recorded multiplex rate, else none and + /// the output is VBR and unpadded ([`Self::emit`]). mux_rate: Option, mux_rate_override: Option, - stuffing: Stuffing, + schedule: Schedule, /// Tune-in point: the first video keyframe's timestamp, captured when the program /// tables are built. Non-video frames before it are dropped so the keyframe leads /// the stream. @@ -408,18 +410,101 @@ enum Kind { }, } -/// The null stuffing owed to the multiplex rate ([`Export::stuff`]). +/// The constant-rate transmission of an export with a multiplex rate +/// ([`Export::schedule`]): closed spans queue here and go out a grid slot at a +/// time, each slot carrying the rate's allowance. #[derive(Default)] -struct Stuffing { +struct Schedule { + /// Closed spans not yet transmitted, oldest first. + backlog: VecDeque, + /// Untransmitted packets across the backlog. + packets: usize, + /// The grid slot the next output frame opens, once the clock has started. + next: Option, + /// How far the transmitted clock trails the media, in nanoseconds: a slot goes + /// out once the media has run this far past its end. It grows to fit the + /// largest burst seen, up to `max_age`, and never shrinks. + delay: u128, + /// The media time the slots are transmitted against, in nanoseconds. + horizon: u128, + /// The largest delay a burst has asked for beyond `max_age`, so each new worst + /// is reported once rather than on every keyframe. + refused: u128, /// Packets the rate has allowed minus packets sent, in [`STUFFING_UNIT`]s. - /// Negative while the media alone exceeds the rate. + /// Negative while a span sent above the rate is being repaid. balance: i64, - /// Packets sent since the last clock packet, that packet included. - since_pcr: u64, /// Whether the debt cap has been hit and reported in the current overrun. overrun: bool, } +impl Schedule { + /// Drop the backlog and stop the clock, keeping the delay the stream has shown + /// it needs. + fn restart(&mut self) { + self.backlog.clear(); + self.packets = 0; + self.next = None; + self.horizon = 0; + self.balance = 0; + self.overrun = false; + } + + /// Queue a closed span; `keyframes` are byte offsets into it. + fn push(&mut self, bytes: Vec, keyframes: Vec, deadline: u128) { + let packets = bytes.len() / TsPacket::SIZE; + if packets == 0 { + return; + } + self.packets += packets; + self.backlog.push_back(Chunk { + bytes, + sent: 0, + keyframes: keyframes.into_iter().map(|at| at / TsPacket::SIZE).collect(), + deadline, + queued: self.horizon.max(deadline), + }); + } + + /// The delay under which a span decoding at `deadline` and queued behind the + /// backlog would reach a receiver in time at `rate`. + /// + /// A queue drained at a constant rate finishes a span at the latest of, for each + /// span ahead of it, when that one was queued plus the time to send it and + /// everything after. The clock runs the delay behind the media, so that finish + /// has to be no later than the deadline plus the delay; the slot the last byte + /// lands in has to end by then too. + fn needs(&self, deadline: u128, rate: u64) -> u128 { + let mut packets = 0; + let mut delay = 0; + for chunk in self.backlog.iter().rev() { + packets += chunk.remaining(); + let finish = chunk.queued + transmit_time(packets, rate) + PCR_INTERVAL.as_nanos(); + delay = delay.max(finish.saturating_sub(deadline)); + } + delay + } +} + +/// One closed span waiting in the [`Schedule`] backlog. +struct Chunk { + bytes: Vec, + /// Packets already transmitted from the front. + sent: usize, + /// Packet indices into `bytes` where a keyframe begins. + keyframes: Vec, + /// The span's earliest media timestamp, in nanoseconds: every byte of it has + /// to be transmitted by then. + deadline: u128, + /// The media time it was queued at, in nanoseconds. + queued: u128, +} + +impl Chunk { + fn remaining(&self) -> usize { + self.bytes.len() / TsPacket::SIZE - self.sent + } +} + /// The program tables plus the resolved PID layout. struct Psi { pat: Pat, @@ -697,13 +782,13 @@ impl Export { video_start: None, mux_rate: None, mux_rate_override: None, - stuffing: Stuffing::default(), + schedule: Schedule::default(), }) } - /// Pad the output with null packets to `mux_rate` bits per second, whatever the - /// catalog records. Without this the catalog's `mpegts.muxRate` decides, and a - /// catalog without one leaves the output unpadded. + /// Transmit the output at a constant `mux_rate` bits per second, padded with null + /// packets, whatever the catalog records. Without this the catalog's + /// `mpegts.muxRate` decides, and a catalog without one leaves the output VBR. /// /// Zero and absurd rates are refused with a warning, leaving the output /// unpadded rather than allocating unbounded nulls. @@ -717,7 +802,8 @@ impl Export { } /// Set the max age for each per-track source, which also bounds how long the - /// interleave holds a leading track for a lagging one. + /// interleave holds a leading track for a lagging one, and how far a + /// constant-rate output may trail the media to spread a burst. /// /// See [`Consumer`](crate::container::Consumer) for the per-track skip behavior. /// Frames are muxed in media-time order across every track, waiting up to this @@ -831,7 +917,7 @@ impl Export { if changed { let joined = self.tracks[&name].epoch == self.epoch; if joined { - if !self.pending.is_empty() { + if !self.pending.is_empty() || self.schedule.packets > 0 { // A boundary ends valid media rather than reneging it. // Return that tail under the old generation before adopting the new one. self.emit(None)?; @@ -1001,8 +1087,13 @@ impl Export { }), }; if self.mux_rate != mux_rate { + // The backlog was laid out for the old rate, so it goes out at that rate + // and the new one takes over from where it ends. + if let Some(rate) = self.mux_rate { + self.flush(rate)?; + } self.mux_rate = mux_rate; - self.stuffing = Stuffing::default(); + self.schedule.restart(); } // Reconcile the SI subscriptions with the catalog's map. Entries may appear @@ -1311,7 +1402,7 @@ impl Export { } self.video_start = None; self.pcr_discontinuity = true; - self.stuffing = Stuffing::default(); + self.schedule.restart(); for track in self.tracks.values_mut() { track.last_dts = None; track.timeline = None; @@ -1704,7 +1795,8 @@ impl Export { /// grid slots *preceding* the span, which are only known once the span has /// closed. That is the mux buffer: the exporter runs two spans behind the media /// clock, by a constant amount, so a caller pacing on the stamps still releases - /// every slot at the interval its own PCR value asserts. + /// every slot at the interval its own PCR value asserts. A constant-rate output + /// runs further behind, by the delay its bursts need ([`Self::schedule`]). fn advance(&mut self, ts: Timestamp) -> anyhow::Result<()> { let Some(watermark) = self.watermark else { self.watermark = Some(ts); @@ -1720,7 +1812,8 @@ impl Export { /// Lay the open span's bytes across the grid slots that run up to its decode /// time, one [`Frame`] per slot. `span` is how long the span ran, or `None` at - /// end of stream, where nothing measured it. + /// end of stream, where nothing measured it. With a multiplex rate the span + /// joins the constant-rate schedule instead ([`Self::schedule`]). /// /// Each frame opens with the clock packets whose slot boundary it starts at, /// then carries the share of the bytes its slice of the interval earns. Three @@ -1743,6 +1836,9 @@ impl Export { /// decode-clock reserve of any track so every PES unit, whichever rendition it /// belongs to, decodes at or after the clock that precedes it. fn emit(&mut self, span: Option) -> anyhow::Result<()> { + if let Some(rate) = self.mux_rate { + return self.schedule(rate, span.is_none()); + } self.span_counters = None; let bytes = std::mem::take(&mut self.pending); let keyframes = std::mem::take(&mut self.keyframes); @@ -1788,10 +1884,9 @@ impl Export { let pcr_pid = self.psi.as_ref().context("PSI not built")?.pcr_pid; let mut payload = Vec::new(); for index in first..=open { - self.stuff(index, &mut payload); let before = counter_before(&bytes, 0, pcr_pid, self.pcr_cc); let clock = self.pcr_at(index, before)?; - self.send(&mut payload, &clock); + payload.extend_from_slice(&clock); } let mut cut = 0; let mut at = from; @@ -1806,18 +1901,16 @@ impl Export { } else { packets }; - self.send(&mut payload, &bytes[cut * TsPacket::SIZE..next * TsPacket::SIZE]); - self.stuff(index, &mut payload); + payload.extend_from_slice(&bytes[cut * TsPacket::SIZE..next * TsPacket::SIZE]); self.push(at, payload, &keyframes, cut, next); cut = next; at = boundary; let before = counter_before(&bytes, cut * TsPacket::SIZE, pcr_pid, self.pcr_cc); let clock = self.pcr_at(index, before)?; - payload = Vec::new(); - self.send(&mut payload, &clock); + payload = clock; } - self.send(&mut payload, &bytes[cut * TsPacket::SIZE..]); + payload.extend_from_slice(&bytes[cut * TsPacket::SIZE..]); self.push(at, payload, &keyframes, cut, packets); if let Some(cc) = counter_before(&bytes, bytes.len(), pcr_pid, None) { self.pcr_cc = Some(cc); @@ -1825,63 +1918,213 @@ impl Export { Ok(()) } - /// Append packets to the open output frame, counting them against the rate. - fn send(&mut self, payload: &mut Vec, packets: &[u8]) { - payload.extend_from_slice(packets); - self.stuffing.since_pcr += (packets.len() / TsPacket::SIZE) as u64; + /// Queue the open span on the constant-rate schedule, then transmit every slot + /// the media has run [`Schedule::delay`] past, or the whole backlog when + /// `flush`ing (end of stream, a boundary). + /// + /// The rate fixes how many bytes each slot carries, so the byte position of a + /// clock packet agrees with its value at the mux rate and a receiver can recover + /// the clock from arrival. What the rate cannot do is carry a keyframe in the + /// slot before it decodes: at 10 Mb/s a 25 ms slot holds 31 kB, and a keyframe + /// is several times that. So the transmitted clock trails the media by a delay, + /// and a burst goes out across the slots before its decode time instead of + /// heaped between two clock packets. The delay grows to fit the largest burst + /// seen, which only holds the output back and so needs no new time base; it is + /// capped by `max_age`, the latency the caller has already said it will hold. A + /// burst the cap cannot fit is not delayed past its decode time or dropped: it + /// goes out above the rate ([`Self::transmit`]), with a warning. + fn schedule(&mut self, rate: u64, flush: bool) -> anyhow::Result<()> { + self.span_counters = None; + let bytes = std::mem::take(&mut self.pending); + let keyframes = std::mem::take(&mut self.keyframes); + let horizon = self.watermark.map_or(0, Timestamp::as_nanos); + self.schedule.horizon = self.schedule.horizon.max(horizon); + if let Some(low) = self.low.take() { + self.enqueue(rate, bytes, keyframes, low.as_nanos()); + } + if flush { + return self.flush(rate); + } + let Some(mut next) = self.schedule.next else { + return Ok(()); + }; + // Slots before this one have ended at least the delay before the media did. + let due = horizon.saturating_sub(self.schedule.delay) / PCR_INTERVAL.as_nanos(); + while next < due { + // As in [`Self::emit`]: past the backfill cap the media gapped, and a dense + // clock history for a span that carried no bytes only stalls a pacer. + if self.schedule.packets == 0 && due - next > PCR_BACKFILL { + next = due - PCR_BACKFILL; + } + self.transmit(next, rate, false)?; + next += 1; + } + Ok(()) + } + + /// Transmit the whole backlog at `rate`. + fn flush(&mut self, rate: u64) -> anyhow::Result<()> { + while self.schedule.packets > 0 { + let next = self.schedule.next.context("backlog without a clock")?; + self.transmit(next, rate, true)?; + } + Ok(()) } - /// Settle the interval that closes at grid slot `index`, the one the clock packet - /// about to be written for it ends, appending the null packets that bring it up - /// to the multiplex rate. Nothing without a rate. + /// Add a closed span to the backlog, raising the delay if it would reach its + /// `deadline` late. + fn enqueue(&mut self, rate: u64, bytes: Vec, keyframes: Vec, deadline: u128) { + let interval = PCR_INTERVAL.as_nanos(); + let cap = self.max_age.as_nanos(); + let schedule = &mut self.schedule; + schedule.push(bytes, keyframes, deadline); + let backlog = transmit_time(schedule.packets, rate); + // The last byte lands inside the slot the backlog ends in, and that slot has to + // end by the deadline. + let lead = backlog + interval; + let next = match schedule.next { + Some(next) => next, + // Carry on from the last clock packet (a VBR stretch before the rate was + // known), or start early enough to carry this span in by its deadline. + None => { + let next = match self.last_pcr { + Some(last) => last + 1, + None => deadline.saturating_sub(lead.min(cap)) / interval, + }; + schedule.next = Some(next); + schedule.delay = schedule.delay.max(deadline.saturating_sub(next * interval).min(cap)); + next + } + }; + if next * interval + lead <= deadline { + return; + } + // Measured from when the backlog was queued rather than added to the delay in + // effect, which the clock may not have fallen back to yet. + let want = schedule.needs(deadline, rate); + if want <= schedule.delay { + return; + } + let delay = want.min(cap); + if delay > schedule.delay { + tracing::info!( + from = ?Duration::from_nanos(schedule.delay as u64), + to = ?Duration::from_nanos(delay as u64), + "raising the MPEG-TS mux buffer delay" + ); + schedule.delay = delay; + } + if want > cap && want > schedule.refused { + tracing::warn!( + needs = ?Duration::from_nanos(want as u64), + max_age = ?self.max_age, + mux_rate = rate, + "burst needs more mux buffer than max_age; sending it above the multiplex rate" + ); + schedule.refused = want; + } + } + + /// Transmit grid slot `index` as one [`Frame`]: its clock packet, then the + /// backlog's share of the rate, then null stuffing up to the rate. /// /// The rate credits each slot its exact fractional allowance and every packet /// sent debits one, so the remainder carries across slots and the long-run count - /// is exact. Media is never delayed or dropped to fit: a slot that already - /// exceeds its allowance gets no nulls and carries the debt forward, and a source - /// that sustains more than the rate simply overruns it. The debt is capped at one - /// second of packets and reported once per overrun, or the rate would never - /// recover after a long burst. Credit is capped the same way, matching the clock - /// backfill: past it the media gapped, and a second of stuffing marks that without - /// filling the whole gap. - fn stuff(&mut self, index: u128, payload: &mut Vec) { - let since = std::mem::take(&mut self.stuffing.since_pcr); - let (Some(rate), Some(last)) = (self.mux_rate, self.last_pcr) else { - return; - }; + /// is exact. Media is never delayed past its decode time or dropped to fit: a + /// span due before the next slot ends goes out whole in this one, above the + /// allowance if it has to, and a source that sustains more than the rate simply + /// overruns it. The excess is repaid from later stuffing rather than from later + /// media, which would only make more of it late. The debt is capped at one second + /// of packets and reported once per overrun, or the rate would never recover + /// after a long burst. A `flush` sends at least one packet per slot, so a rate + /// below two packets a slot still drains. + fn transmit(&mut self, index: u128, rate: u64, flush: bool) -> anyhow::Result<()> { + let pcr_pid = self.psi.as_ref().context("PSI not built")?.pcr_pid; // The rate is sanitized where it enters (the builder and the catalog), so - // this holds; refuse to pad rather than wrap if it ever does not. - let Ok(rate) = i64::try_from(rate) else { - return; - }; - let slots = index.saturating_sub(last).min(PCR_BACKFILL) as i64; - let stuffing = &mut self.stuffing; - stuffing.balance = stuffing - .balance - .saturating_add(slots.saturating_mul(rate)) - .saturating_sub((since as i64).saturating_mul(STUFFING_UNIT)); + // this holds. + let rate = i64::try_from(rate).context("multiplex rate out of range")?; + // Nothing on the clock PID has gone out yet: repeat the counter one behind + // the first packet queued on it. + let before = self.pcr_cc.or_else(|| { + self.schedule + .backlog + .iter() + .find_map(|chunk| counter_before(&chunk.bytes[chunk.sent * TsPacket::SIZE..], 0, pcr_pid, None)) + }); + let mut payload = self.pcr_at(index, before)?; + + let schedule = &mut self.schedule; + let budget = schedule.balance.max(0).saturating_add(rate); + schedule.balance = schedule.balance.saturating_add(rate); + // Everything up to the last span that would miss its deadline in the next slot. + let deadline = (index + 2) * PCR_INTERVAL.as_nanos(); + let mut due = 0; + let mut queued = 0; + for chunk in &schedule.backlog { + queued += chunk.remaining(); + if chunk.deadline < deadline { + due = queued; + } + } + let allowance = usize::try_from(budget / STUFFING_UNIT - 1).unwrap_or(0); + let mut take = allowance.min(schedule.packets).max(due); + if flush { + take = take.max(1).min(schedule.packets); + } + + let mut keyframe = false; + let mut left = take; + while left > 0 { + let Some(chunk) = schedule.backlog.front_mut() else { + break; + }; + let n = left.min(chunk.remaining()); + let (from, to) = (chunk.sent, chunk.sent + n); + keyframe |= chunk.keyframes.iter().any(|&at| at >= from && at < to); + let packets = &chunk.bytes[from * TsPacket::SIZE..to * TsPacket::SIZE]; + if let Some(cc) = counter_before(packets, packets.len(), pcr_pid, None) { + self.pcr_cc = Some(cc); + } + payload.extend_from_slice(packets); + chunk.sent = to; + schedule.packets -= n; + left -= n; + if chunk.remaining() == 0 { + schedule.backlog.pop_front(); + } + } + + let sent = 1 + (take - left) as i64; + schedule.balance = schedule.balance.saturating_sub(sent.saturating_mul(STUFFING_UNIT)); let floor = rate.saturating_mul(-SLOTS_PER_SECOND); - if stuffing.balance < floor { - if !std::mem::replace(&mut stuffing.overrun, true) { + if schedule.balance < floor { + if !std::mem::replace(&mut schedule.overrun, true) { tracing::warn!(mux_rate = rate, "MPEG-TS output exceeds the multiplex rate"); } - stuffing.balance = floor; - } else if stuffing.balance >= 0 { - stuffing.overrun = false; - } - let nulls = (stuffing.balance / STUFFING_UNIT).max(0); - // Never emit more than the slots just credited could allow: a bound on one - // call's allocation whatever the balance holds. A no-op on a rate that came - // through [`sanitize_mux_rate`], whose balance carries less than one packet. - // (A manual `div_ceil`: all terms are non-negative, and the toolchain's - // signed `div_ceil` is still unstable.) - let ceiling = slots.saturating_mul(rate).saturating_add(STUFFING_UNIT - 1) / STUFFING_UNIT; - let nulls = nulls.min(ceiling); - stuffing.balance -= nulls * STUFFING_UNIT; + schedule.balance = floor; + } else if schedule.balance >= 0 { + schedule.overrun = false; + } + // Never more than one slot's credit, a bound on the allocation whatever the + // balance holds. (A manual `div_ceil`: all terms are non-negative, and the + // toolchain's signed `div_ceil` is still unstable.) + let ceiling = rate.saturating_add(STUFFING_UNIT - 1) / STUFFING_UNIT; + let nulls = (schedule.balance / STUFFING_UNIT).clamp(0, ceiling); + schedule.balance -= nulls * STUFFING_UNIT; payload.reserve(nulls as usize * TsPacket::SIZE); for _ in 0..nulls { payload.extend_from_slice(&NULL_PACKET); } + + schedule.next = Some(index + 1); + self.clock = Some(slot_stamp(index + 1)?); + self.queue.push_back(Frame { + timestamp: slot_stamp(index)?, + duration: None, + payload: Bytes::from(payload), + keyframe, + }); + Ok(()) } /// Queue one output frame, unless it would be empty. `from`..`to` are the packet @@ -2249,6 +2492,14 @@ fn slot_ticks(index: u128, interval: Duration) -> u64 { (index * interval.as_nanos() * 90_000 / 1_000_000_000) as u64 } +/// How long `packets` of media take to transmit at `rate` bits per second, in +/// nanoseconds, given that every grid slot also carries its clock packet. +fn transmit_time(packets: usize, rate: u64) -> u128 { + let unit = STUFFING_UNIT as u128; + let media = u128::from(rate).saturating_sub(unit).max(1); + (packets as u128 * unit * PCR_INTERVAL.as_nanos()).div_ceil(media) +} + /// External byte size of an adaptation field (manual mirror of the crate's /// private `external_size`); only PCR is ever set. fn adaptation_size(af: &AdaptationField) -> usize { diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index cadb57a7f4a..ed5647feb13 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -5103,6 +5103,337 @@ async fn export_stuffing_keeps_the_fractional_remainder() { assert!(clocked.nulls > 0, "no null stuffing was emitted"); } +/// The multiplex rate of the [`export_bursty`] fixture: 16.6 packets per 25 ms slot. +const BURSTY_RATE: u64 = 1_000_000; + +/// A live H.264 feed at 25 fps whose keyframes outgrow a PCR slot many times over, +/// the shape of a broadcast contribution feed (#3925): `keyframe(n)` bytes on the +/// `n`th keyframe, one a second, and 1.5 kB frames between, about 0.6 Mb/s in all +/// against the 1 Mb/s the export is padded to. Written a frame at a time with the +/// export polled after each, so a small `max_age` sees a live edge rather than a +/// recording to skip through. +async fn export_bursty(max_age: Duration, keyframe: impl Fn(u64) -> usize) -> Vec { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let track = broadcast + .create_track( + broadcast.unique_name(".h264"), + hang::container::track_info(hang::catalog::PRIORITY.video), + ) + .unwrap(); + let mut cfg = VideoConfig::new(H264 { + profile: 0x42, + constraints: 0xc0, + level: 0x1f, + inline: true, + }); + cfg.container = Container::Legacy; + catalog + .modify() + .unwrap() + .video + .renditions + .insert(track.name().to_string(), cfg); + let mut video = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data)); + + let mut export = Export::new(crate::source::announced(&consumer)) + .await + .unwrap() + .with_max_age(max_age) + .with_mux_rate(BURSTY_RATE); + assert!(drain_frames(&mut export).await.is_empty()); + + let mut out = Vec::new(); + for i in 0..125u64 { + let key = i.is_multiple_of(25); + let mut nal = vec![if key { 0x65u8 } else { 0x41 }]; + nal.resize(if key { keyframe(i / 25) } else { 1_500 }, 0xAB); + if key && i > 0 { + video.cut(None).unwrap(); + } + video + .write(Frame { + // Clear of the start, so the clock the schedule starts ahead of the first + // keyframe does not wrap below zero. + timestamp: Timestamp::from_micros(10_000_000 + i * 40_000).unwrap(), + duration: None, + payload: if key { + annexb(&[SPS, PPS, &nal]) + } else { + annexb(&[&nal]) + }, + keyframe: key, + }) + .unwrap(); + out.extend(poll_frames(&mut export)); + } + video.finish().unwrap(); + out.extend(drain_frames(&mut export).await); + out +} + +/// A packet index paired with a 90 kHz time. +type Mark = (usize, u64); + +/// What a receiver clocking off arrival sees of `frames`: every PCR as (packet index, +/// 90 kHz value), and every video access unit as (the index of its last packet, the +/// 90 kHz time it decodes at). +fn arrivals(frames: &[Frame]) -> (Vec, Vec) { + let ts: Vec = frames.iter().flat_map(|f| f.payload.iter().copied()).collect(); + assert_packet_aligned(&ts); + let stamp = |b: &[u8]| { + (u64::from(b[0] & 0x0e) << 29) + | (u64::from(b[1]) << 22) + | (u64::from(b[2] & 0xfe) << 14) + | (u64::from(b[3]) << 7) + | (u64::from(b[4]) >> 1) + }; + let mut pcrs = Vec::new(); + let mut units: Vec = Vec::new(); + let mut video = None; + for (index, packet) in ts.chunks(188).enumerate() { + let pid = (u16::from(packet[1] & 0x1f) << 8) | u16::from(packet[2]); + let adaptation = packet[3] & 0x20 != 0; + if adaptation && packet[4] >= 7 && packet[5] & 0x10 != 0 { + assert_eq!(packet[5] & 0x80, 0, "a signalled discontinuity at packet {index}"); + pcrs.push((index, pcr_base(packet))); + } + if packet[3] & 0x10 == 0 { + continue; + } + let start = if adaptation { 5 + usize::from(packet[4]) } else { 4 }; + let pes = &packet[start..]; + if packet[1] & 0x40 != 0 && pes.starts_with(&[0, 0, 1, 0xe0]) { + video = Some(pid); + // PTS_DTS_flags: 0b10 is PTS only, 0b11 is PTS then DTS. + let decode = if pes[7] >> 6 == 0b11 { &pes[14..19] } else { &pes[9..14] }; + units.push((index, stamp(decode))); + } else if video == Some(pid) + && let Some(unit) = units.last_mut() + { + unit.0 = index; + } + } + (pcrs, units) +} + +/// The 33-bit PCR base of a clock packet. +fn pcr_base(packet: &[u8]) -> u64 { + (u64::from(packet[6]) << 25) + | (u64::from(packet[7]) << 17) + | (u64::from(packet[8]) << 9) + | (u64::from(packet[9]) << 1) + | u64::from(packet[10] >> 7) +} + +/// Assert every access unit's last byte reaches a receiver clocking off arrival by +/// the time it decodes, the T-STD's condition on a constant-rate stream. +fn assert_arrives_before_decode(pcrs: &[Mark], units: &[Mark]) { + let mut checked = 0; + for &(last, decode) in units { + let Some(pair) = pcrs.windows(2).find(|w| w[0].0 <= last && last < w[1].0) else { + continue; + }; + let ((a, va), (b, vb)) = (pair[0], pair[1]); + let arrival = va + (last + 1 - a) as u64 * (vb - va) / (b - a) as u64; + assert!( + arrival <= decode, + "an access unit ending at packet {last} arrives at {arrival}, after it decodes at {decode}" + ); + checked += 1; + } + assert!(checked > 100, "only {checked} access units were checked"); +} + +/// #3925: with a multiplex rate, every PCR interval carries the bytes the rate implies, +/// keyframes included. Padding each slot up to the rate got the average right, but a +/// keyframe landed whole between the two clock packets before its decode time: 40 kB +/// against a 3 kB slot here, 870 kB against 31 kB on a broadcast capture. +#[tokio::test(start_paused = true)] +async fn export_spreads_keyframes_over_the_mux_rate() { + let frames = export_bursty(Duration::from_secs(1), |_| 40_000).await; + let (pcrs, units) = arrivals(&frames); + assert!(pcrs.len() > 150, "expected the full feed, got {} PCRs", pcrs.len()); + + let nominal = BURSTY_RATE as f64 * PCR_INTERVAL.as_secs_f64() / (188.0 * 8.0); + for (i, w) in pcrs.windows(2).enumerate() { + assert_eq!(w[1].1 - w[0].1, 2_250, "value step at {i}"); + let packets = (w[1].0 - w[0].0) as f64; + assert!( + (packets - nominal).abs() <= 1.0, + "interval {i} carries {packets} packets, not the {nominal:.2} the rate implies" + ); + } + assert_arrives_before_decode(&pcrs, &units); +} + +/// The delay grows to fit the largest burst seen: a keyframe twice any before it +/// goes out above the rate once, and every later one of that size is on schedule. +#[tokio::test(start_paused = true)] +async fn export_mux_delay_grows_to_the_largest_burst() { + let frames = export_bursty(Duration::from_secs(1), |n| if n == 0 { 20_000 } else { 40_000 }).await; + let (pcrs, units) = arrivals(&frames); + assert_arrives_before_decode(&pcrs, &units); + + let nominal = BURSTY_RATE as f64 * PCR_INTERVAL.as_secs_f64() / (188.0 * 8.0); + let over: Vec = pcrs + .windows(2) + .enumerate() + .filter(|(_, w)| (w[1].0 - w[0].0) as f64 > nominal + 1.0) + .map(|(i, _)| i) + .collect(); + assert!( + !over.is_empty(), + "the first larger keyframe had no room and must overrun" + ); + // The second keyframe decodes 1 s (40 slots) after the first; everything past the + // slots carrying it keeps the rate. + let settled = over[0] + 40; + assert!( + over.iter().all(|&i| i < settled), + "intervals {over:?} overran after the delay grew" + ); +} + +/// A burst that needs more delay than `max_age` allows is neither late nor dropped: it +/// goes out above the rate, in the slots before it decodes. +#[tokio::test(start_paused = true)] +async fn export_burst_beyond_max_age_overruns_the_rate() { + let frames = export_bursty(Duration::from_millis(100), |_| 40_000).await; + let (pcrs, units) = arrivals(&frames); + assert_arrives_before_decode(&pcrs, &units); + + let nominal = BURSTY_RATE as f64 * PCR_INTERVAL.as_secs_f64() / (188.0 * 8.0); + let over = pcrs + .windows(2) + .filter(|w| (w[1].0 - w[0].0) as f64 > nominal + 1.0) + .count(); + assert!( + over >= 5, + "every keyframe must overrun a 100 ms delay, {over} intervals did" + ); + let video: usize = units.len(); + assert_eq!(video, 125, "every access unit is carried"); +} + +/// A dropped malformed section still opens a span, one that carries no packets. If it +/// is the last thing muxed before a marker, the boundary queues it behind everything +/// the flush sends, and a backlog waited on until it empties would never empty: the +/// export spins inside `poll_next` without yielding, which no async timeout can +/// interrupt, so the export runs on its own thread against a wall-clock deadline. +#[test] +fn export_marker_after_a_dropped_section_does_not_spin() { + let (done, finished) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .start_paused(true) + .build() + .unwrap(); + done.send(runtime.block_on(marker_after_a_dropped_section())).unwrap(); + }); + let (discontinuity, frames) = finished + .recv_timeout(Duration::from_secs(30)) + .expect("the export hung at the marker"); + assert_eq!(discontinuity, 1, "the marker was observed"); + let ts: Vec = frames.iter().flat_map(|f| f.payload.iter().copied()).collect(); + let (video, _) = collect_pes_pts(&ts); + for i in 0..10u64 { + let pts = (11_000_000 + i * 40_000) * 90 / 1000; + assert!(video.contains(&pts), "video after the marker is missing PTS {pts}"); + } +} + +async fn marker_after_a_dropped_section() -> (u64, Vec) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + + let track = broadcast + .create_track( + broadcast.unique_name(".h264"), + hang::container::track_info(hang::catalog::PRIORITY.video), + ) + .unwrap(); + let mut cfg = VideoConfig::new(H264 { + profile: 0x42, + constraints: 0xc0, + level: 0x1f, + inline: true, + }); + cfg.container = Container::Legacy; + let scte = broadcast + .unique_track(".scte35", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + { + let mut guard = catalog.modify().unwrap(); + guard.video.renditions.insert(track.name().to_string(), cfg); + guard.ext.mpegts.tracks.insert( + scte.name().to_string(), + tscat::Track { + pid: 0x102, + descriptors: Vec::new(), + verbatim: Some(tscat::Verbatim::new(0x86, tscat::Framing::Section)), + }, + ); + } + let mut video = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data)); + let mut cues = Producer::new(scte, HangContainer::Legacy(crate::container::Kind::Data)); + + let mut export = Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang) + .await + .unwrap() + .with_max_age(Duration::from_millis(500)) + .with_mux_rate(BURSTY_RATE); + let mut out = drain_frames(&mut export).await; + + // A keyframe several slots long, so the schedule holds a delay and still has + // packets queued when the marker arrives. + let frame = |i: u64, at: u64| { + let key = i == 0; + let mut nal = vec![if key { 0x65u8 } else { 0x41 }]; + nal.resize(if key { 20_000 } else { 1_500 }, 0xAB); + Frame { + timestamp: Timestamp::from_micros(at).unwrap(), + duration: None, + payload: if key { + annexb(&[SPS, PPS, &nal]) + } else { + annexb(&[&nal]) + }, + keyframe: key, + } + }; + // Few enough frames that the keyframe is still queued at the marker. + for i in 0..4u64 { + video.write(frame(i, 10_000_000 + i * 40_000)).unwrap(); + } + // After the last frame, so it opens the span the marker closes. Its + // section_length promises 32 bytes and one follows. + cues.write(Frame { + timestamp: Timestamp::from_micros(10_140_000).unwrap(), + duration: None, + payload: Bytes::from_static(&[0xfc, 0x30, 0x20, 0x00]), + keyframe: true, + }) + .unwrap(); + out.extend(drain_frames(&mut export).await); + + video.discontinuity().unwrap(); + for i in 0..10u64 { + video.write(frame(i, 11_000_000 + i * 40_000)).unwrap(); + } + video.finish().unwrap(); + cues.finish().unwrap(); + out.extend(drain_frames(&mut export).await); + (export.discontinuity(), out) +} + // The decode clock follows the stream's reordering: its reserve comes from the catalog `jitter`, // else the depth the SPS declares, else the reordering muxed so far, and it keeps following all // three after the program tables are written. diff --git a/test/ts/README.md b/test/ts/README.md index 4fb5950b9d8..c2058919830 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -7,7 +7,7 @@ runs [TSDuck](https://tsduck.io) plus a custom analyzer over it. This is a diagnostic gate, not just a pass/fail: the exporter ([`rs/moq-mux/src/container/ts/export.rs`](../../rs/moq-mux/src/container/ts/export.rs)) -is VBR, inserts no null packets, and paces PCR once per media frame, so several +is VBR and inserts no null packets unless the catalog declares a mux rate, so several broadcast-shape checks are expected to flag. The report quantifies exactly where and by how much. @@ -41,8 +41,8 @@ runs (see [CI](#ci)). The default arm runs `pcr-timing.py` over its capture too, after `compliance.py`, for the one thing the IRD model does not grade: whether the bytes between consecutive PCRs are the ones the mux rate implies -([`pcr-schedule`](#byte-schedule)). It is a shape check, so it reports without -gating unless `--strict`. +([`pcr-schedule`](#byte-schedule)). It gates on the generated clip and reports +without gating on a `--source` capture unless `--strict`. The live arm passes only when the grader's verdict *and* the publisher's exit status are clean. The grader can only speak for what reached it, and the sample floor @@ -223,15 +223,18 @@ the exporter's unpadded first half-second pulled it ~3 % low and read every padd interval after it as off schedule. `run.sh` passes the generated clip's rate (`--bitrate`), and lets the grader estimate for `--source`. -The generated clip is a weak fixture for this check. It compresses to almost -nothing, so padding dominates and no keyframe outgrows its slot: measured against -the exporter this check was written for, 92-97 % of intervals were on schedule -over three 20 s runs, the misses being near-empty intervals from the unpadded -start. A real constant-rate capture is the case that -discriminates. A 60 s cut of a 9.95 Mb/s broadcast clip round-tripped through the -same harness came back with a median of 1,316 B between PCRs against 31,081 B -nominal and 3.1 % of intervals within tolerance, while its aggregate rate was -within 16 b/s of nominal: +The generated clip is detailed enough that each keyframe is 110-140 kB, several +25 ms slots at 10 Mb/s, while the average frame fits one. An exporter that pads +without scheduling heaps each keyframe between two PCRs, and scored 34 % on it; the +scheduled exporter keeps ~92 %, the misses being the ~2 s before the importer has +measured the rate, when the output is still VBR. `run.sh` gates on it with +`--schedule-pct-min`, allowing those 4 s off schedule (80 % over the default 20 s). + +A real constant-rate capture is graded without a gate, since its rate is estimated. +A 60 s cut of a 9.95 Mb/s broadcast clip round-tripped through the same harness +came back, before the exporter scheduled its bytes, with a median of 1,316 B +between PCRs against 31,081 B nominal and 3.1 % of intervals within tolerance, +while its aggregate rate was within 16 b/s of nominal: ```bash just test ts --source cap.ts --duration 60 # reports the schedule, estimating the rate diff --git a/test/ts/run.sh b/test/ts/run.sh index 0858b89dadc..ab2f590e2ed 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -248,12 +248,16 @@ else # recovery-point SEI. Fixed B placement in a 24-frame GOP puts three B pictures # right after each one in decode order, presented before it: leading pictures. [[ -n "$OPEN_GOP" ]] && X264="keyint=24:min-keyint=24:scenecut=0:open-gop=1:bframes=3:b-adapt=0" + # Detailed, slowly moving content under a VBV, so each keyframe is several PCR + # slots of bytes at the mux rate (~110-140 kB against ~31 kB per 25 ms at 10 Mb/s) + # while the average frame fits one, the shape of a broadcast feed. A flat test + # pattern compresses to almost nothing, pads, and keeps any schedule. ffmpeg -y -hide_banner -loglevel error \ - -f lavfi -i "testsrc=size=1280x720:rate=25" \ + -f lavfi -i "mandelbrot=size=1280x720:rate=25" \ -f lavfi -i "sine=frequency=1000:sample_rate=48000" \ -t "$DURATION" \ -c:v libx264 -profile:v high -preset veryfast -pix_fmt yuv420p \ - -x264-params "$X264" -b:v 8M \ + -x264-params "$X264" -b:v 6M -maxrate 8M -bufsize 2M \ -c:a aac -b:a 128k \ -f mpegts -muxrate "$BITRATE" -pcr_period 20 -pes_payload_size 0 "$SRC_TS" fi @@ -521,9 +525,16 @@ fi # compliance.py grades rate in aggregate and over fixed windows, neither of which says # whether the bytes between consecutive PCRs are the ones the mux rate implies, so the # capture goes through pcr-timing.py as well. Its hard checks gate here as they do under -# --live; pcr-schedule is a shape check, so it reports without gating unless --strict. +# --live. pcr-schedule gates on the generated clip, whose keyframes outgrow a PCR slot +# and whose rate is known. The exporter is VBR for the ~2 s the importer takes to +# measure that rate, so the gate allows 4 s off schedule: 80 % over the default 20 s, +# where the scheduled output holds ~92 % and VBR held 34 %. A --source capture +# reports without gating unless --strict. +GATE=() +[[ -z "$SOURCE" ]] && GATE=(--schedule-pct-min $((DURATION > 4 ? 100 * (DURATION - 4) / DURATION : 0))) echo -if ! python3 "$DIR/pcr-timing.py" "$SUB_TS" ${SCHEDULE[@]+"${SCHEDULE[@]}"} $STRICT; then +if ! python3 "$DIR/pcr-timing.py" "$SUB_TS" ${SCHEDULE[@]+"${SCHEDULE[@]}"} \ + ${GATE[@]+"${GATE[@]}"} $STRICT; then echo >&2 echo "error: PCR timing analysis failed (see round-trip logs below)" >&2 dump_logs