diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 91ce37a6d5..b15a385730 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -187,6 +187,20 @@ ffmpeg -i input.mp4 -c copy -f mpegts - | \ moq-cli subscribe --url https://relay.example.com --broadcast my-stream --format ts | ffplay - ``` +By default a retained broadcast is written out as fast as it can be read. Add +`--pace` to emit the TS stream at its real-time rate instead, like ffmpeg's +`-re`, which is what a downstream player or re-publish expects: + +```bash +moq-cli subscribe --url https://relay.example.com --broadcast my-stream --format ts --pace \ + | ffmpeg -i - -c copy -f mpegts udp://239.0.0.1:1234 +``` + +Pacing follows the live edge: it holds at most `--max-latency` of buffer (default +500ms) ahead of the playhead, so a retained broadcast streams at its media rate +while a live source stays near the edge instead of letting the initial burst +become permanent latency. + TS export carries H.264 / H.265 as Annex-B and AAC as ADTS. Both in-band (avc3 / hev1) and out-of-band (avc1 / hvc1, e.g. from an fMP4 import) video sources work: the parameter sets are read from the bitstream or the catalog diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 152330d65a..dbb4e3deb8 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -283,6 +283,8 @@ async fn run_subscribe( broadcast: String, args: SubscribeArgs, ) -> anyhow::Result<()> { + args.validate()?; + let origin = moq_net::Origin::random().produce(); let consumer = origin.consume(); diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 36a6e2dd3e..ae441713f6 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -43,6 +43,15 @@ pub struct SubscribeArgs { #[arg(long, default_value = "500ms", value_parser = humantime::parse_duration)] pub max_latency: Duration, + /// Pace output at the media's real-time rate, like ffmpeg's `-re`. + /// + /// Without this, a retained broadcast is written out as fast as it can be read. + /// With it, frames are emitted on the media clock (anchored to the first frame), + /// which is what a downstream player or re-publish expects. Only `--format ts` + /// supports pacing today. + #[arg(long)] + pub pace: bool, + /// Cap the output fragment duration (e.g. `2s`, `500ms`). /// /// By default a fragment covers one GOP (rolled over on video keyframes). @@ -61,6 +70,15 @@ pub struct SubscribeArgs { } impl SubscribeArgs { + /// Reject flag combinations that don't apply to the chosen format, before we + /// bother connecting to the relay. + pub fn validate(&self) -> anyhow::Result<()> { + if self.pace && !matches!(self.format, SubscribeFormat::Ts) { + anyhow::bail!("--pace is only supported with --format ts"); + } + Ok(()) + } + /// Resolve the catalog format, falling back to detection from the broadcast /// name suffix and then to the default. pub fn catalog_format(&self, broadcast: &str) -> CatalogFormat { @@ -135,6 +153,7 @@ impl Subscribe { async fn run_ts(self) -> anyhow::Result<()> { let mut stdout = tokio::io::stdout(); + let pace = self.args.pace; // TS emits PAT/PMT then a continuous PES stream (re-emitting PAT/PMT at // keyframes for tune-in). Avc3/Hev1 sources pass through as Annex-B; AAC @@ -145,6 +164,13 @@ impl Subscribe { moq_mux::container::ts::Export::with_ts(self.broadcast, self.catalog)?.with_latency(self.args.max_latency); while let Some(frame) = ts.next().await? { + // `--pace`: hold each frame until it's due. The muxer paces on its decode clock + // and follows the live edge with up to `--max-latency` of buffer, so a retained + // broadcast drains in real time while a live source stays near the edge. + if pace { + tokio::time::sleep_until(frame.pace.into()).await; + } + stdout.write_all(&frame.payload).await?; stdout.flush().await?; } @@ -170,3 +196,26 @@ impl Subscribe { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn args(format: SubscribeFormat, pace: bool) -> SubscribeArgs { + SubscribeArgs { + format, + max_latency: Duration::from_millis(500), + pace, + fragment_duration: None, + catalog: None, + } + } + + #[test] + fn pace_requires_ts_format() { + assert!(args(SubscribeFormat::Ts, true).validate().is_ok()); + assert!(args(SubscribeFormat::Fmp4, true).validate().is_err()); + // Without --pace, any format is fine. + assert!(args(SubscribeFormat::Fmp4, false).validate().is_ok()); + } +} diff --git a/rs/moq-mux/src/container/mod.rs b/rs/moq-mux/src/container/mod.rs index ae05aeefde..dd77c5b29a 100644 --- a/rs/moq-mux/src/container/mod.rs +++ b/rs/moq-mux/src/container/mod.rs @@ -16,6 +16,7 @@ use bytes::Bytes; mod consumer; pub(crate) mod jitter; +mod pace; mod producer; mod source; @@ -27,6 +28,7 @@ pub mod mkv; pub mod ts; pub use consumer::Consumer; +pub(crate) use pace::Pacer; pub use producer::Producer; pub(crate) use source::ExportSource; diff --git a/rs/moq-mux/src/container/pace.rs b/rs/moq-mux/src/container/pace.rs new file mode 100644 index 0000000000..c516b955b0 --- /dev/null +++ b/rs/moq-mux/src/container/pace.rs @@ -0,0 +1,162 @@ +//! Pace media output on a media clock, following the live edge. + +use std::time::{Duration, Instant}; + +use super::Timestamp; + +/// Maps media (decode) timestamps onto the wall clock so a caller can emit frames at +/// the source's real-time rate while bounding how far behind the live edge it falls. +/// +/// The first frame anchors the media clock to "now"; every later frame is due at +/// `anchor + (timestamp - base)`. Sleeping until [`Pacer::due`] before emitting a frame +/// drains a retained broadcast at its media rate, like ffmpeg's `-re`, instead of as +/// fast as it can be read. +/// +/// To keep a bursty or faster-than-real source from accruing unbounded latency, the +/// timeline holds at most `lead` of buffer ahead of now: when a frame would be due +/// further out than that (a tune-in burst delivers a whole GOP at once, or the source +/// drifts ahead of wall-clock), the anchor jumps forward to the live edge so the buffer +/// never exceeds `lead`. A frame that merely trails the edge (network jitter, a +/// reordered B-frame) keeps its earlier instant. +/// +/// `lead` is the target buffer, typically the subscription's max latency. With `lead` +/// = 0 the timeline never leads now, which is what an SRT egress wants (it stamps each +/// payload's TSBPD origin time and the receiver owns the jitter buffer); with `lead` > +/// 0 the caller sleeps to pace output and holds that much buffer itself. +#[derive(Default)] +pub(crate) struct Pacer { + anchor: Option, +} + +/// The media-clock anchor: `base`'s media time maps to `at` on the wall clock. +struct Anchor { + at: Instant, + base: Timestamp, +} + +impl Pacer { + /// A pacer that anchors on the first frame it sees. + pub(crate) fn new() -> Self { + Self::default() + } + + /// The wall-clock instant `timestamp` is due, holding at most `lead` of buffer + /// ahead of now. Reads the clock itself; see [`Self::at`] for the testable core. + pub(crate) fn due(&mut self, timestamp: Timestamp, lead: Duration) -> Instant { + self.at(timestamp, lead, Instant::now()) + } + + /// [`Self::due`] with an explicit `now`, so the mapping is deterministic in tests. + fn at(&mut self, timestamp: Timestamp, lead: Duration, now: Instant) -> Instant { + let anchor = self.anchor.get_or_insert(Anchor { + at: now, + base: timestamp, + }); + + let due = match timestamp.checked_sub(anchor.base) { + Ok(ahead) => anchor.at + Duration::from(ahead), + Err(_) => anchor + .at + .checked_sub(Duration::from(anchor.base - timestamp)) + .unwrap_or(anchor.at), + }; + + // Never schedule more than `lead` ahead: when the source outruns that, re-anchor + // the live edge to `now + lead` so the buffer stays bounded. Re-anchoring only ever + // moves forward, so a frame that trails the edge keeps its earlier instant. + let cap = now + lead; + if due > cap { + anchor.at = cap; + anchor.base = timestamp; + cap + } else { + due + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ms(m: u64) -> Timestamp { + Timestamp::from_micros(m * 1_000).unwrap() + } + + #[test] + fn paces_on_the_media_clock() { + // Within the lead budget, frames pace on the media clock: a frame 40ms later in + // media is due 40ms after the anchor, however quickly it was read. This is the + // `-re` case (the caller sleeps in lockstep, so the cap never trips). + let mut pacer = Pacer::new(); + let start = Instant::now(); + let lead = Duration::from_millis(500); + + assert_eq!( + pacer.at(ms(1_000), lead, start), + start, + "the first frame anchors to now" + ); + assert_eq!( + pacer.at(ms(1_040), lead, start + Duration::from_millis(1)), + start + Duration::from_millis(40), + "output is paced on the media clock, not arrival time" + ); + } + + #[test] + fn re_anchors_past_the_lead_budget() { + // A tune-in burst: a whole GOP is read at once, so the newest frame's media time + // runs far past `now`. It re-anchors to `now + lead` rather than scheduling + // seconds out, so the buffer never exceeds the target latency. + let mut pacer = Pacer::new(); + let start = Instant::now(); + let lead = Duration::from_millis(500); + pacer.at(ms(1_000), lead, start); + + let now = start + Duration::from_millis(2); + let edge = pacer.at(ms(4_132), lead, now); + assert_eq!(edge, now + lead, "the live edge is held `lead` ahead of now"); + + // A later frame paces off the re-anchored edge. The caller slept until the edge + // was due, so `now` has advanced to it; the 40ms-newer frame is due 40ms past it. + let now = now + lead; + assert_eq!( + pacer.at(ms(4_172), lead, now), + now + Duration::from_millis(40), + "subsequent frames pace off the re-anchored edge" + ); + } + + #[test] + fn trailing_frame_keeps_its_instant() { + // A reordered B-frame whose timestamp dips below the edge maps into the past, so + // the caller's sleep is a no-op and it's emitted immediately. No re-anchor. + let mut pacer = Pacer::new(); + let start = Instant::now(); + let lead = Duration::from_millis(500); + pacer.at(ms(1_000), lead, start); + + assert_eq!( + pacer.at(ms(967), lead, start + Duration::from_millis(5)), + start - Duration::from_millis(33), + ); + } + + #[test] + fn zero_lead_never_leads_now() { + // `lead` = 0 is the SRT egress policy: the timeline is capped at now, so a burst + // re-anchors to now (the newest frame is the live edge) and nothing is stamped + // into the future. + let mut pacer = Pacer::new(); + let start = Instant::now(); + pacer.at(ms(1_000), Duration::ZERO, start); + + let now = start + Duration::from_millis(2); + assert_eq!( + pacer.at(ms(4_132), Duration::ZERO, now), + now, + "the live edge paces to now" + ); + } +} diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 6953bdef6d..c5921c22d1 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -15,7 +15,7 @@ use std::collections::HashMap; use std::task::Poll; -use std::time::Duration; +use std::time::{Duration, Instant}; use anyhow::Context; use bytes::Bytes; @@ -32,7 +32,7 @@ use mpeg2ts::ts::{ use crate::catalog::hang::{Catalog, CatalogExt}; use crate::catalog::{CatalogFormat, Stream}; use crate::codec::annexb; -use crate::container::{ExportSource, Frame, Timestamp}; +use crate::container::{ExportSource, Frame, Pacer, Timestamp}; use super::adts; use super::catalog; @@ -46,8 +46,9 @@ const PSI_INTERVAL: Duration = Duration::from_millis(500); /// Subscribe to a broadcast and produce an MPEG-TS byte stream. /// -/// Use [`next`](Self::next) to pull one [`Frame`] per media frame: its `payload` -/// is the TS packets, stamped with the source `timestamp` and `keyframe` flag. +/// Use [`next`](Self::next) to pull one [`Output`] per media frame: its `payload` +/// is the TS packets, stamped with the source `timestamp` and `keyframe` flag, +/// plus a [`pace`](Output::pace) instant on the decode clock for real-time output. /// The leading PAT/PMT rides on the first frame (so it inherits a real /// timestamp), and is re-emitted at video keyframes and periodically for /// mid-stream tune-in. Returns `None` when the broadcast ends. @@ -56,6 +57,14 @@ pub struct Export { catalog: Option>, latency: Duration, + /// Maps each frame's decode timestamp to the wall-clock instant it's due, so the + /// caller can pace output at the source's real-time rate (see [`Output::pace`]). + pacer: Pacer, + /// Output pace buffer: how far ahead of the live edge [`Output::pace`] may schedule. + /// `None` follows `latency`; a transport that buffers downstream (e.g. an SRT + /// receiver's TSBPD) sets it to zero so the pace adds no latency of its own. + pace_lead: Option, + tracks: HashMap, /// Continuity counter per PID (PAT, PMT, and each elementary stream). counters: HashMap, @@ -129,6 +138,33 @@ enum Kind { }, } +/// One muxed MPEG-TS frame produced by [`Export::next`]. +/// +/// `payload` is the TS packets (188-byte aligned) for a single media frame. +#[non_exhaustive] +pub struct Output { + /// The TS packets for one media frame. + pub payload: Bytes, + + /// Source presentation timestamp. + pub timestamp: Timestamp, + + /// Whether this frame is a keyframe. + pub keyframe: bool, + + /// Wall-clock instant this frame is due, paced on the decode clock (DTS) so a + /// reordered (B-frame) stream paces evenly rather than on its non-monotonic + /// presentation timestamps. + /// + /// The muxer follows the live edge: it holds at most the configured latency + /// ([`with_latency`](Export::with_latency)) of buffer ahead of now, re-anchoring + /// past that so a tune-in burst or a faster-than-real source never accrues more + /// latency than the budget. Sleep until it (or stamp a transport send time with + /// it) to emit at the source's real-time rate, like ffmpeg's `-re`; ignore it to + /// emit as fast as the broadcast can be read. + pub pace: Instant, +} + /// The program tables plus the resolved PID layout. struct Psi { pat: Pat, @@ -186,6 +222,8 @@ impl Export { broadcast, catalog: Some(catalog), latency: Duration::ZERO, + pacer: Pacer::new(), + pace_lead: None, tracks: HashMap::new(), counters: HashMap::new(), program_descriptors: Vec::new(), @@ -195,25 +233,42 @@ impl Export { }) } - /// Set the maximum buffering latency for each per-track source. + /// Set the maximum buffering latency for each per-track source (the read/skip + /// budget: how far behind the live edge a track may fall before skipping groups). + /// + /// Unless overridden by [`with_pace_lead`](Self::with_pace_lead), this is also the + /// pace buffer: [`Output::pace`] holds at most this far ahead of the live edge, so + /// a caller honoring it never falls more than `latency` behind. pub fn with_latency(mut self, latency: Duration) -> Self { self.latency = latency; self } + /// Override the [`Output::pace`] buffer independently of the read latency. + /// + /// By default the pace holds up to [`with_latency`](Self::with_latency) ahead of + /// the live edge. Set this to zero when a downstream transport supplies its own + /// jitter buffer (e.g. an SRT receiver's TSBPD), so the pace stamps each frame at + /// the live edge and the transport, not the muxer, owns the buffering. + pub fn with_pace_lead(mut self, lead: Duration) -> Self { + self.pace_lead = Some(lead); + self + } + /// Get the next muxed frame. /// - /// Each [`Frame`] carries the TS packets for one media frame in `payload`, - /// stamped with that frame's media `timestamp` and `keyframe` flag so a - /// transport can pace delivery on the media clock. The leading PAT/PMT rides - /// on the first frame (inheriting its timestamp), and is re-emitted at video - /// keyframes and periodically for mid-stream tune-in. Returns `None` when the - /// broadcast ends. `duration` is always `None`: the muxer has no use for it. - pub async fn next(&mut self) -> anyhow::Result> { + /// Each [`Output`] carries the TS packets for one media frame in `payload`, + /// stamped with that frame's media `timestamp` and `keyframe` flag, plus a + /// [`pace`](Output::pace) instant so a transport can emit on the media clock. + /// The leading PAT/PMT rides on the first frame (inheriting its timestamp), and + /// is re-emitted at video keyframes and periodically for mid-stream tune-in. + /// Returns `None` when the broadcast ends. + pub async fn next(&mut self) -> anyhow::Result> { kio::wait(|waiter| self.poll_next(waiter)).await } - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { + /// Poll variant of [`next`](Self::next), drivable from any executor. + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { // 1. Drain catalog updates, discovering the track layout. while let Some(catalog) = self.catalog.as_mut() { match catalog.poll_next(waiter)? { @@ -642,11 +697,11 @@ impl Export { .map(|(n, _)| n) } - /// Packetize one media frame into an output [`Frame`], re-emitting PAT/PMT - /// before video keyframes (and periodically) so receivers can tune in - /// mid-stream. The returned frame keeps the source `timestamp` and `keyframe` - /// flag so the caller can pace it. - fn write_frame(&mut self, name: &str, frame: Frame) -> anyhow::Result { + /// Packetize one media frame into an [`Output`], re-emitting PAT/PMT before video + /// keyframes (and periodically) so receivers can tune in mid-stream. The output + /// keeps the source `timestamp` and `keyframe` flag, plus a decode-clock `pace` + /// instant for real-time delivery. + fn write_frame(&mut self, name: &str, frame: Frame) -> anyhow::Result { let track = self.tracks.get(name).context("missing track")?; let pid = track.pid; let kind = track.kind.clone(); @@ -696,6 +751,17 @@ impl Export { None }; + // Pace on the decode clock: reordered video uses the authored DTS (monotonic), so + // a caller honoring `pace` emits B-frames evenly instead of bursting on their + // non-monotonic PTS. Audio and non-reordered video have DTS == PTS. The pace buffer + // follows the read latency unless overridden (an SRT egress pins it to zero). + let decode = match dts { + Some(ticks) => from_ticks(ticks).unwrap_or(timestamp), + None => timestamp, + }; + let lead = self.pace_lead.unwrap_or(self.latency); + let pace = self.pacer.due(decode, lead); + let mut out = Vec::with_capacity(TsPacket::SIZE); // Refresh PSI at keyframes or after the interval lapses. @@ -738,11 +804,11 @@ impl Export { self.write_pes(&mut out, &unit, &es_payload)?; } } - Ok(Frame { - timestamp, - duration: None, + Ok(Output { payload: Bytes::from(out), + timestamp, keyframe, + pace, }) } @@ -935,6 +1001,14 @@ fn to_ticks(timestamp: Timestamp) -> u64 { (timestamp.as_micros() * 90_000 / 1_000_000) as u64 } +/// Inverse of [`to_ticks`]: continuous 90 kHz decode-clock ticks back to the microsecond +/// timebase, for pacing on the decode clock. Returns [`moq_net::TimeOverflow`] only for +/// absurdly large tick counts (tens of thousands of years). +fn from_ticks(ticks: u64) -> Result { + // 1_000_000 / 90_000 reduces to 100 / 9; multiply first to keep sub-tick precision. + Timestamp::from_micros(ticks.saturating_mul(100) / 9) +} + fn to_ts_timestamp(timestamp: Timestamp) -> anyhow::Result { // Continuous 90 kHz ticks, wrapped into the 33-bit field. TsTimestamp::new(to_ticks(timestamp) & TS_TIMESTAMP_MASK).map_err(anyhow::Error::msg) diff --git a/rs/moq-srt/src/server.rs b/rs/moq-srt/src/server.rs index d6f6094b95..a00b8f80a5 100644 --- a/rs/moq-srt/src/server.rs +++ b/rs/moq-srt/src/server.rs @@ -50,19 +50,24 @@ pub struct Server { /// Held to keep the listener (and its UDP socket) alive for the server's lifetime. _listener: SrtListener, incoming: SrtIncoming, + /// The configured SRT latency, reused as the egress muxer's read/skip budget. + latency: Duration, } impl Server { /// Bind an SRT listener on `addr` (SRT has no well-known port; 9000 is common). /// /// `latency` is the SRT receive latency, negotiated at handshake time; pass - /// `None` for a sensible default (200ms). + /// `None` for a sensible default (200ms). A subscribe (egress) connection also + /// reuses it as the muxer's read/skip budget, so a track tolerates the same jitter + /// the receiver's TSBPD does. pub async fn bind(addr: SocketAddr, latency: impl Into>) -> Result { let latency = latency.into().unwrap_or(DEFAULT_LATENCY); let (listener, incoming) = SrtListener::builder().latency(latency).bind(addr).await?; Ok(Self { _listener: listener, incoming, + latency, }) } @@ -87,6 +92,7 @@ impl Server { resource, stream_id, peer, + latency: self.latency, }; // `m=request` reads a broadcast out; everything else publishes one in. @@ -111,6 +117,8 @@ struct Pending { /// fields out of it (e.g. a token in `u=` or a custom key). stream_id: Option, peer: SocketAddr, + /// The server's configured SRT latency, applied as the egress read/skip budget. + latency: Duration, } /// What an accepted SRT connection wants: to contribute media ([`Publish`]) or to @@ -243,7 +251,7 @@ impl Subscribe { pub async fn accept(self, origin: &OriginConsumer, path: &str) -> Result<()> { let socket = self.0.request.accept(None).await?; tracing::info!(peer = %self.0.peer, %path, "SRT subscribe accepted"); - serve_subscribe(origin, path, socket).await + serve_subscribe(origin, path, socket, self.0.latency).await } /// Reject the subscribe, sending the client a `Forbidden` rejection. @@ -282,18 +290,21 @@ async fn serve_publish(origin: &OriginProducer, path: &str, mut socket: SrtSocke /// Waits for the broadcast to be announced (so a caller may connect before the /// publisher), then packs the muxer's output into [`SRT_PAYLOAD`]-sized SRT /// messages. Returns once the broadcast ends or the caller disconnects. -async fn serve_subscribe(origin: &OriginConsumer, path: &str, mut socket: SrtSocket) -> Result<()> { +async fn serve_subscribe(origin: &OriginConsumer, path: &str, mut socket: SrtSocket, latency: Duration) -> Result<()> { // Resolve the broadcast, but watch the socket while we wait: `announced_broadcast` // parks forever for a stream that is never published, and nothing else polls the // socket during that wait, so without this a caller who requests a non-existent // stream (or hangs up before it starts) would leak this task and its socket. + // The muxer's read/skip budget matches the server's configured SRT latency, so a track + // tolerates the same jitter the receiver's TSBPD does. The pace lead stays zero: the + // TSBPD owns the output buffer. let subscriber = tokio::select! { biased; _ = wait_closed(&mut socket) => { tracing::debug!(%path, "SRT subscribe closed before its broadcast was available"); return Ok(()); } - subscriber = crate::ts::Subscriber::new(origin, path) => subscriber?, + subscriber = crate::ts::Subscriber::new(origin, path, latency) => subscriber?, }; let Some(mut subscriber) = subscriber else { @@ -304,26 +315,16 @@ async fn serve_subscribe(origin: &OriginConsumer, path: &str, mut socket: SrtSoc // MPEG-TS is a continuous byte stream, so we coalesce the muxer's per-frame // output and slice it on a fixed boundary rather than preserving frames. // - // Pace each payload on the media clock: the Instant handed to `send` is the - // payload's origin time feeding the receiver's TSBPD, which reconstructs the - // inter-frame spacing from it. We don't know the live playhead when a subscriber - // attaches, so `pace` anchors it for us -- the newest frame is "now" and earlier - // frames map to proportionally earlier instants, re-anchoring whenever the media - // outruns wall-clock (a tune-in burst, a catch-up, or producer drift). `anchor` - // and `base` are that media-clock anchor (`base`'s media time maps to `anchor`), - // carried across frames and moved forward by `pace`. - let mut anchor = Instant::now(); - let mut base = None; - let mut send_at = anchor; + // The Instant handed to `send` is the payload's TSBPD origin time, from which the + // receiver reconstructs inter-frame spacing. The muxer paces each frame on its + // decode clock; with the pace lead at zero (the SRT receiver owns the jitter buffer) + // it re-anchors any tune-in burst to the live edge, so nothing is stamped seconds + // into the future where SRT would hold it past the TSBPD window and stall after + // ~one packet. + let mut send_at = Instant::now(); let mut buffer = bytes::BytesMut::new(); while let Some(frame) = subscriber.next().await? { - // The media zero-point the rest pace against; `pace` re-anchors it forward to - // the live edge whenever the media outruns wall-clock. - let zero = *base.get_or_insert(frame.timestamp); - let paced = pace(anchor, zero, frame.timestamp, Instant::now()); - send_at = paced.send_at; - anchor = paced.anchor; - base = Some(paced.base); + send_at = frame.pace; buffer.extend_from_slice(&frame.payload); while buffer.len() >= SRT_PAYLOAD { @@ -339,55 +340,6 @@ async fn serve_subscribe(origin: &OriginConsumer, path: &str, mut socket: SrtSoc Ok(()) } -/// One frame's SRT send time plus the media-clock anchor to carry into the next -/// frame. `base`'s media time maps to `anchor` on the wall clock. -struct Paced { - /// The Instant to stamp this payload with: its TSBPD origin time. - send_at: Instant, - anchor: Instant, - base: moq_mux::container::Timestamp, -} - -/// Pace one SRT payload on the media clock, re-anchored to the live edge. -/// -/// `send_at = anchor + (ts - base)` stamps the payload at its media time, so the -/// receiver's TSBPD reconstructs inter-frame spacing from it. But when that runs -/// ahead of `now` -- the media clock has outrun wall-clock: a subscriber tuning in -/// bursts the current GOP plus any catch-up backlog (frames whose media timestamps -/// span seconds, produced within milliseconds), or the producer drifts faster than -/// real time -- re-anchor the live edge to `now` (this frame becomes the playhead). -/// Otherwise the payload is stamped seconds into the future, where SRT holds it in -/// the sender past the receiver's TSBPD latency window and playback stalls after -/// ~one packet. -/// -/// Re-anchoring only ever moves the anchor *forward*, so a frame that merely arrives -/// late -- network/CPU jitter, or a reordered B-frame whose PTS trails the edge -- -/// keeps its earlier media instant instead of collapsing to its arrival instant, and -/// TSBPD still smooths the jitter into even playout. The cap is "never lead `now`": -/// the receiver owns the jitter buffer (the SRT latency parameter), so the sender -/// adds no lookahead of its own (a deliberate lead would just be `now + lead` here). -fn pace( - anchor: Instant, - base: moq_mux::container::Timestamp, - ts: moq_mux::container::Timestamp, - now: Instant, -) -> Paced { - let send_at = match ts.checked_sub(base) { - Ok(offset) => anchor + Duration::from(offset), - Err(_) => anchor.checked_sub(Duration::from(base - ts)).unwrap_or(anchor), - }; - if send_at > now { - // Media outran wall-clock: re-anchor so this newest frame is the live edge. - Paced { - send_at: now, - anchor: now, - base: ts, - } - } else { - Paced { send_at, anchor, base } - } -} - /// Resolve once the SRT caller hangs up (a clean close or an error), draining and /// ignoring any unexpected inbound packets. A subscribe caller normally sends /// nothing, so this is purely a disconnect signal to race against the announce wait. @@ -433,62 +385,6 @@ fn parse_stream_id(stream_id: Option<&StreamId>) -> Option<(String, ConnectionMo mod tests { use super::*; - #[test] - fn pace_re_anchors_to_live_edge() { - use moq_mux::container::Timestamp; - use std::time::{Duration, Instant}; - let ms = |m: u64| Timestamp::from_micros(m * 1_000).unwrap(); - - // Tune-in burst: the live edge (4132ms of media) is produced ~8ms after the - // first frame (1400ms). It must re-anchor to `now` rather than stamp ~2.7s - // into the future, which would stall the receiver's TSBPD after ~one payload. - let start = Instant::now(); - let now = start + Duration::from_millis(8); - let edge = pace(start, ms(1_400), ms(4_132), now); - assert_eq!(edge.send_at, now, "live edge should pace to now"); - assert_eq!(edge.anchor, now); - assert_eq!( - edge.base.as_micros(), - ms(4_132).as_micros(), - "anchor moves up to the live edge" - ); - - // Part 1 moved the anchor to the live edge (now / 4132ms). A frame 33ms newer in - // MEDIA that arrives 80ms later in WALL-clock (jitter) paces from that carried- - // forward anchor -- its media instant (+33ms off the edge), not its 80ms arrival - // instant -- so TSBPD still reconstructs smooth spacing. - let jittered = pace( - edge.anchor, - edge.base, - ms(4_165), - edge.anchor + Duration::from_millis(80), - ); - assert_eq!( - jittered.send_at, - edge.anchor + Duration::from_millis(33), - "a late frame keeps its media instant, not its arrival instant" - ); - assert_eq!( - jittered.anchor, edge.anchor, - "no re-anchor when media is behind wall-clock" - ); - - // A reordered B-frame can carry a PTS before the re-anchored live edge. Keep - // that earlier media instant instead of flattening it onto the anchor. - let reordered = pace( - edge.anchor, - edge.base, - ms(4_099), - edge.anchor + Duration::from_millis(20), - ); - assert_eq!( - reordered.send_at, - edge.anchor - Duration::from_millis(33), - "a reordered frame can pace before the anchor" - ); - assert_eq!(reordered.anchor, edge.anchor, "no re-anchor when media trails the edge"); - } - fn sid(s: &str) -> StreamId { StreamId::try_from(s.as_bytes().to_vec()).unwrap() } diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index 5df51b58e1..44f7d6e61f 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -7,8 +7,10 @@ //! mirror image for egress: it consumes a broadcast from the origin and re-muxes //! it back to MPEG-TS for an SRT caller (VLC, ffmpeg) to play. +use std::time::Duration; + use bytes::Bytes; -use moq_mux::container::{Frame, ts}; +use moq_mux::container::ts; use moq_net::{Broadcast, OriginConsumer, OriginProducer}; use crate::Result; @@ -73,21 +75,28 @@ pub struct Subscriber { impl Subscriber { /// Resolve the broadcast at `path` in the origin and prepare to mux it to TS. /// + /// `latency` is the server's configured SRT latency: the muxer matches it as the + /// read/skip budget so a track tolerates the same jitter the SRT receiver does. The + /// pace lead stays zero, since the receiver's TSBPD owns the output buffer (adding a + /// muxer lead would just double the latency). + /// /// Returns `Ok(None)` if the broadcast can never be served (path outside the /// consumer's scope, or the origin closed). Otherwise waits for the broadcast /// to be announced, so a caller may connect before the publisher does. - pub async fn new(origin: &OriginConsumer, path: &str) -> Result> { + pub async fn new(origin: &OriginConsumer, path: &str, latency: Duration) -> Result> { let Some(broadcast) = origin.announced_broadcast(path).await else { return Ok(None); }; - let export = ts::Export::new(broadcast)?; + let export = ts::Export::new(broadcast)? + .with_latency(latency) + .with_pace_lead(Duration::ZERO); Ok(Some(Self { export })) } - /// Pull the next muxed frame (TS bytes + media timestamp), or `None` once the - /// broadcast ends. - pub async fn next(&mut self) -> Result> { + /// Pull the next muxed frame (TS bytes, media timestamp, and pacing instant), or + /// `None` once the broadcast ends. + pub async fn next(&mut self) -> Result> { Ok(self.export.next().await?) } }