diff --git a/README.md b/README.md index 6225b1d..36bc9a6 100644 --- a/README.md +++ b/README.md @@ -153,6 +153,17 @@ failover replaces the first relay's copy once that copy ends. It plays the best video and audio its catalog lists, switches when the catalog changes, and comes back on its own when the publisher restarts. +A broadcast source, a guest's included, plays on its publisher's timestamps, +so its sound and picture stay in sync through network jitter and stalls. Its +audio plays as far behind what arrives as the network's jitter needs, measured +as it goes, the same delay `moq play --delay auto` holds, up to 2 s. Each +picture waits for the sound stamped with it, and the canvas shows the last one +meanwhile. A sound stall longer than that delay pauses the picture too and +starts both again together; a picture stall leaves the sound playing. A burst +after a stall skips ahead to the delay. Without audio, the +picture plays 100 ms behind. Sources shows how far behind each broadcast +plays. + ### Audio mixes Audio from every source is mixed whether or not its scene is live. A @@ -174,7 +185,9 @@ studio remembers for you rather than in the scene file. Use headphones: a microphone that hears the speakers feeds the mix back into itself. A source's **delay** holds it back in every mix, up to 2 s, so a microphone -lines up with a camera that lags it. +lines up with a camera that lags it. A broadcast's picture already follows its +sound, so its delay holds the sound back from its own picture: use it only for +a publisher whose sound leads their picture. Each mix is advanced on the program clock, and each output reads its mix on its own, so an output that falls behind skips ahead in its own audio rather diff --git a/quest/m0/README.md b/quest/m0/README.md index b51b3e9..8f4343b 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -91,7 +91,6 @@ repository. - [Studio on macOS with capture granted](/quest/m0/app-macos.md) - camera, screen, and audio capture run on a Mac, not only Linux - [Studio on Windows](/quest/m0/app-windows.md) - the studio runs on Windows, not only Linux - [Audio source tracks](/quest/m0/audio-source-tracks.md) - microphones and system audio get their own tracks, for isolated recording and replay -- [Broadcast sources play in sync](/quest/m0/broadcast-sync.md) - subscribed broadcasts and guests play audio and video on the publisher's timestamps, with `moq play`'s delay - [Wayland display and window sources](/quest/m0/wayland-sources.md) - portal-picked sources that reopen without the picker - [Keyframes](/quest/m0/animation.md) - keyframed item properties with easing, starting when their scene goes on program - [Media file source](/quest/m0/media-file.md) - a local video or audio file plays as a source, looping or from the start on demand diff --git a/quest/m0/broadcast-sync.md b/quest/m0/broadcast-sync.md deleted file mode 100644 index 8c1646b..0000000 --- a/quest/m0/broadcast-sync.md +++ /dev/null @@ -1,34 +0,0 @@ -# [M] Broadcast sources play in sync - -## Goal - -A subscribed MoQ broadcast in `moq-studio`, guests included, plays its audio -and video on the publisher's timestamps, so lip sync holds through network -jitter and stalls. Today each track plays at its own live edge, so a stall -can leave audio and video apart. - -## Plan - -Decided 2026-10-26 (maintainer): - -- **Scope:** broadcast sources only. Local devices keep the manual - per-source `delay`, which [Clap-to-measure](/quest/m0/sync-clap.md) sets - from a clap. -- **Delay:** reuse `moq play`'s timing, the `--delay auto` target from - moq-audio's jitter estimate (`rs/moq-audio/src/jitter.rs`, - `doc/concept/audio-jitter.md`), rather than a second algorithm or a manual - setting. As in `moq play`, the estimate is the broadcast's audio track's, - and video follows it rather than keeping its own, so one delay governs the - whole broadcast. A broadcast with no audio keeps video's own short delay. - Show each source's delay in the UI. - -Use the timing `moq play` shares rather than copying it: moq-mux's -`timeline` module, pinned here (decided 2026-11-02, maintainer). The video -side holds frames until their audio-aligned time, and a source still repeats -its last frame rather than stalling the canvas. This replaces the live-edge behavior the studio app PR chose, where -replaying a publisher's timestamp gaps pushed audio behind its own video. Here -video waits for audio instead. Update `README.md`. - -## Related - -- [Audio jitter target](https://github.com/moq-dev/moq/blob/main/quest/m1/audio-jitter-target/README.md) - the delay estimate this reuses diff --git a/quest/m0/sync-clap.md b/quest/m0/sync-clap.md index 915f631..cdb20d9 100644 --- a/quest/m0/sync-clap.md +++ b/quest/m0/sync-clap.md @@ -22,7 +22,3 @@ The studio sees a source's audio only through its strip's level meter, read once a frame, which is too coarse to time a clap. Timing it needs the source's samples, or an onset, with their position on the bus clock, which is moq-audio work: land it there first and pin it. - -## Related - -- [Broadcast sources play in sync](/quest/m0/broadcast-sync.md) - broadcast sources align on the publisher's timestamps instead diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index f5d0d77..dfed418 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -37,6 +37,7 @@ mod edit; mod effect; mod lut; mod overlay; +mod playout; pub mod publish; pub mod replay; mod scale; diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs new file mode 100644 index 0000000..7ec5903 --- /dev/null +++ b/rs/moq-scene/src/playout.rs @@ -0,0 +1,484 @@ +//! When a broadcast source's audio sounds and its pictures show: on its +//! publisher's timestamps, so they stay in sync through jitter and stalls. +//! +//! A broadcast's audio and video share one [`timeline`] clock, as they do in +//! `moq play`: audio owns the clock's anchor while it plays, and video shows +//! each picture once the clock says it is due and its sound has played. +//! Without audio, video paces itself [`VIDEO_DELAY`] behind its arrivals. +//! +//! `moq play` holds its jitter buffer in the speaker. The bus holds only +//! [`BUS_LATENCY`] of a source, so a [`Speaker`] holds the rest: it places each +//! frame the playout target moq-audio's decoder estimates after it arrives, +//! and a [`Writer`] writes it the bus's latency ahead of then. + +use std::time::Duration; + +use kio::clock::Instant; +use moq_audio::bus; +use moq_mux::timeline; +use tokio::sync::Notify; + +/// How far a broadcast without audio plays behind its arrivals, as under +/// `moq play --delay auto`. With audio, video follows the audio's target. +pub(crate) const VIDEO_DELAY: Duration = Duration::from_millis(100); + +/// The longest playout target, the most lateness the jitter estimate can +/// represent, and so how long a stalled group is waited on before it is +/// skipped, as under `moq play --delay auto`. A shorter wait would cap the +/// estimate at it, since a skipped group is never measured. +pub(crate) const MAX_DELAY: Duration = Duration::from_secs(2); + +/// What a bus channel holds between a write and the mix that plays it, while +/// it is kept fed. +/// +/// moq-audio's `bus::LATENCY`, which is private; `a_write_sounds_when_placed` +/// pins the two together. +pub(crate) const BUS_LATENCY: Duration = Duration::from_millis(50); + +/// How far over its target the queue may run before an arrival is skipped to +/// land back on it, as `moq play`'s speaker allows: one frame of cadence. +const SLACK: Duration = Duration::from_millis(20); + +/// How far apart two stamps may be and still meet: a millisecond-stamped +/// ingest rounds each one. +const CONTIGUOUS: Duration = Duration::from_millis(1); + +/// Places one broadcast audio track's frames on the bus clock, and reports +/// where they sound to the broadcast's clock, which video follows. +/// +/// Dropping it hands the clock's anchor back to video. +pub(crate) struct Speaker<'a> { + clock: timeline::Producer, + /// Woken on every frame placed and when the speaker stops, so pictures + /// held for their sound show once it plays, or once there is none. + changed: &'a Notify, + /// The last frame placed: when it finishes sounding, and the media time it + /// ends at. + tail: Option<(Instant, Duration)>, + /// When the media after a skip starts sounding. The clock holds one + /// mapping, and the audio queued ahead of the skip plays on the old one + /// until then, so frames placed before it are reported only once it comes: + /// see [`commit`](Self::commit). + skipped: Option, + /// The newest frame placed but not reported yet: when it finishes + /// sounding, the media time it ends at, and the target it was placed on. + unreported: Option<(Instant, Duration, Duration)>, +} + +/// What to do with a frame [`Speaker::place`] was given. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Place { + /// Play it from this moment. + Play(Instant), + /// Drop it: more is queued than the target needs, after a burst or a + /// target that fell, so the media skips ahead to land back on the target. + Skip, +} + +impl<'a> Speaker<'a> { + /// Place audio on `clock`, taking its anchor from video on the first frame, + /// and wake `changed` whenever that may let a picture show. + pub(crate) fn new(clock: timeline::Producer, changed: &'a Notify) -> Self { + Self { + clock, + changed, + tail: None, + skipped: None, + unreported: None, + } + } + + /// Place a decoded frame covering `duration` of media from `start`, + /// arriving at `now`, given the playout `target` moq-audio estimated. + /// + /// A frame continues the one before, after silence for any hole in the + /// media between them. A frame that arrives too late for that, because the + /// network stalled longer than the target, re-buffers instead: it sounds the + /// whole target after it arrived, and the clock re-pins on it, so video + /// waits along with the sound rather than running ahead of it. A frame + /// that would queue past the target, or that overlaps media already queued, + /// is skipped, so the queue never holds more than the target. + pub(crate) fn place(&mut self, start: Duration, duration: Duration, target: Duration, now: Instant) -> Place { + let target = target.min(MAX_DELAY); + let end = start.saturating_add(duration); + let begins = match self.tail { + None => now + target + BUS_LATENCY, + Some((sounds, media)) => { + let follows = sounds + start.saturating_sub(media); + // A write half the bus's latency late still finds sound queued + // ahead of it, however coarsely the bus advances. + if follows < now + BUS_LATENCY / 2 { + self.clock.audio_restarted(); + self.skipped = None; + self.unreported = None; + now + target + BUS_LATENCY + } else if start.saturating_add(CONTIGUOUS) < media + || follows.saturating_duration_since(now) > target + BUS_LATENCY + SLACK + { + // Media overlapping what is queued, or past the target, would + // queue without bound however fast it arrives. + self.tail = Some((sounds, media.max(end))); + self.skipped = Some(sounds); + return Place::Skip; + } else { + follows + } + } + }; + let sounds = begins + duration; + self.tail = Some((sounds, end)); + self.unreported = Some((sounds, end, target)); + if self.skipped.is_none_or(|skipped| skipped <= now) { + self.report(now); + } + Place::Play(begins) + } + + /// When [`commit`](Self::commit) has a report to make, after a skip. + pub(crate) fn deferred(&self) -> Option { + self.unreported.and(self.skipped) + } + + /// Report the frames placed since a skip, once the media after it sounds at + /// `now`, whether or not another frame has arrived since. + pub(crate) fn commit(&mut self, now: Instant) { + if self.deferred().is_some_and(|skipped| skipped <= now) { + self.report(now); + } + } + + /// Tell the clock where the newest frame placed sounds, as of `now`. + fn report(&mut self, now: Instant) { + self.skipped = None; + let Some((sounds, end, target)) = self.unreported.take() else { + return; + }; + let audio = timeline::Audio { + end, + buffered: sounds.saturating_duration_since(now), + target, + }; + self.clock.audio(audio, now); + self.changed.notify_one(); + } +} + +impl Drop for Speaker<'_> { + fn drop(&mut self) { + self.clock.audio_stopped(); + self.changed.notify_one(); + } +} + +/// Whether a picture stamped `timestamp` may show at `now` on `schedule`: +/// while audio plays, only once the sound stamped with it has played, so a +/// stall in the audio alone holds the picture too. +pub(crate) fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> bool { + schedule.target().is_none() + || schedule + .playhead(now) + .is_some_and(|head| Duration::from(timestamp) <= head + CONTIGUOUS) +} + +/// Writes placed frames into a bus channel, so each sounds when it was placed. +/// +/// A channel plays a write [`BUS_LATENCY`] later only while it holds that much: +/// one that ran dry plays the next write at once. So the writer keeps track of +/// when what it wrote runs out, and puts silence ahead of a frame for however +/// much of the latency has drained. +pub(crate) struct Writer { + channel: bus::Channel, + /// Bytes per sample frame of the channel's `f32` input. + stride: usize, + rate: u32, + /// When what was written finishes sounding, or `None` before the first + /// write, which the channel's own priming holds back the full latency. + runs_out: Option, + /// Silence and the frame after it, written as one, so the channel pads + /// behind the frame rather than between them. + pending: Vec, +} + +impl Writer { + /// Write into a new `channel`, taking interleaved `f32` at `rate` in `channels`. + pub(crate) fn new(channel: bus::Channel, rate: u32, channels: u32) -> Self { + Self { + channel, + stride: channels as usize * size_of::(), + rate, + runs_out: None, + pending: Vec::new(), + } + } + + /// Write `data` at `now` to sound at `sounds`. Call it no sooner than + /// [`BUS_LATENCY`] ahead. Written late, the part whose moment has passed is + /// dropped, so the rest still sounds on schedule, and so does what follows. + pub(crate) fn write(&mut self, sounds: Instant, data: &[u8], now: Instant) -> Result<(), moq_audio::Error> { + let free = self.runs_out.unwrap_or(now + BUS_LATENCY).max(now); + let late = self.frames(free.saturating_duration_since(sounds)) * self.stride; + let Some(data) = data.get(late..).filter(|data| !data.is_empty()) else { + return Ok(()); + }; + let gap = self.frames(sounds.saturating_duration_since(free).min(BUS_LATENCY)); + self.pending.clear(); + self.pending.resize(gap * self.stride, 0); + self.pending.extend_from_slice(data); + self.channel.write(&self.pending)?; + let runs_out = free + self.duration(self.pending.len() / self.stride); + // A channel left holding a quarter of its latency or less pads itself + // back up to it behind the write, as moq-audio's ring does. + self.runs_out = Some(match runs_out.saturating_duration_since(now) <= BUS_LATENCY / 4 { + true => now + BUS_LATENCY, + false => runs_out, + }); + Ok(()) + } + + fn frames(&self, duration: Duration) -> usize { + (duration.as_secs_f64() * f64::from(self.rate)).round() as usize + } + + fn duration(&self, frames: usize) -> Duration { + Duration::from_secs_f64(frames as f64 / f64::from(self.rate)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const FRAME: Duration = Duration::from_millis(20); + const TARGET: Duration = Duration::from_millis(100); + + fn at(millis: u64) -> Instant { + Instant::ZERO + Duration::from_secs(10) + Duration::from_millis(millis) + } + + fn ms(millis: u64) -> Duration { + Duration::from_millis(millis) + } + + fn stamp(millis: u64) -> moq_net::Timestamp { + moq_net::Timestamp::from_millis(millis).unwrap() + } + + /// The first frame waits out the target and the bus's latency, later ones + /// follow it on their stamps however unevenly they arrive within it, and a + /// picture is due when the sound stamped with it plays. + #[test] + fn video_is_due_when_its_audio_sounds() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + assert_eq!(speaker.place(ms(1_000), FRAME, TARGET, at(0)), Place::Play(at(150))); + assert_eq!(clock.due(stamp(1_000)), Some(at(150))); + // Late within the target, and early: both keep their place. + assert_eq!(speaker.place(ms(1_020), FRAME, TARGET, at(90)), Place::Play(at(170))); + assert_eq!(speaker.place(ms(1_040), FRAME, TARGET, at(91)), Place::Play(at(190))); + assert_eq!(clock.due(stamp(1_040)), Some(at(190))); + assert_eq!(clock.consume().target(), Some(TARGET)); + + // A stopped track hands the clock back to video. + drop(speaker); + assert_eq!(clock.consume().target(), None); + } + + /// A picture whose sound has not arrived waits for it, even once due: the + /// audio stalled, and the picture stalls with it. + #[test] + fn a_picture_waits_for_its_sound_to_play() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let schedule = clock.consume(); + assert!( + heard(&schedule, stamp(5_000), at(0)), + "without audio, only the deadline counts" + ); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + speaker.place(ms(1_000), FRAME, TARGET, at(0)); + assert!(!heard(&schedule, stamp(1_000), at(148))); + assert!(heard(&schedule, stamp(1_000), at(150))); + assert!(clock.due(stamp(1_100)).is_some_and(|due| due < at(500))); + assert!(heard(&schedule, stamp(1_020), at(500))); + assert!(!heard(&schedule, stamp(1_100), at(500)), "shown past its sound"); + } + + /// A stall longer than the target re-buffers: the next frame sounds the + /// target after it arrived, and the pictures stamped with it wait as long. + #[test] + fn a_stall_holds_video_back_with_the_sound() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + speaker.place(ms(1_000), FRAME, TARGET, at(0)); + assert_eq!(speaker.place(ms(1_020), FRAME, TARGET, at(500)), Place::Play(at(650))); + assert_eq!(clock.due(stamp(1_020)), Some(at(650))); + } + + /// Lost media plays as the silence it was, so the pictures after it keep + /// their moments; a hole past the target skips instead, and the clock + /// re-pins on the frame after it. + #[test] + fn a_hole_plays_as_silence_up_to_the_target() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + speaker.place(ms(1_000), FRAME, TARGET, at(0)); + assert_eq!(speaker.place(ms(1_060), FRAME, TARGET, at(40)), Place::Play(at(210))); + assert_eq!(clock.due(stamp(1_060)), Some(at(210))); + + // 1 s lost, with only the target queued: skipped back onto the target. + // The audio queued ahead of the skip keeps its pictures' moments until + // the media after it starts to sound. + assert_eq!(speaker.place(ms(2_080), FRAME, TARGET, at(60)), Place::Skip); + let schedule = clock.consume(); + assert_eq!(speaker.place(ms(2_100), FRAME, TARGET, at(61)), Place::Play(at(230))); + assert_eq!(clock.due(stamp(1_060)), Some(at(210))); + assert!(!heard(&schedule, stamp(2_100), at(61))); + // It sounds whether or not anything arrives after it. + assert_eq!(speaker.deferred(), Some(at(230))); + speaker.commit(at(229)); + assert_eq!(clock.due(stamp(1_060)), Some(at(210))); + speaker.commit(at(230)); + assert_eq!(speaker.deferred(), None); + assert_eq!(clock.due(stamp(2_100)), Some(at(230))); + assert!(heard(&schedule, stamp(2_100), at(230))); + } + + /// A burst queues no more than the target and a frame of slack: the rest + /// is skipped, and play goes on from what arrives next. + #[test] + fn a_burst_skips_past_the_target() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + let placed = (0..10) + .map(|index| speaker.place(ms(1_000 + index * 20), FRAME, TARGET, at(0))) + .filter(|place| *place != Place::Skip) + .collect::>(); + assert_eq!(placed, [Place::Play(at(150)), Place::Play(at(170))]); + // What arrives next follows the queue, and the clock takes its media + // once the first of it sounds, on arrival or not. + for index in 0..9 { + let place = speaker.place(ms(1_200 + index * 20), FRAME, TARGET, at(20 + index * 20)); + assert_eq!(place, Place::Play(at(190 + index * 20))); + assert_eq!(clock.due(stamp(1_000)), Some(at(150)), "{index}"); + } + assert_eq!(speaker.place(ms(1_380), FRAME, TARGET, at(200)), Place::Play(at(370))); + assert_eq!(clock.due(stamp(1_200)), Some(at(190))); + assert_eq!(speaker.deferred(), None); + } + + /// Media overlapping what is queued, stamped the same or a millisecond + /// apart however long each frame is, is skipped rather than queued behind + /// it; once the queue runs out, a rewound timeline plays from the target. + #[test] + fn overlapping_media_queues_no_more_than_the_target() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + for index in 0..1_000 { + for start in [ms(1_000), ms(1_000 + index)] { + if let Place::Play(begins) = speaker.place(start, FRAME, TARGET, at(0)) { + assert!(begins <= at(0) + TARGET + BUS_LATENCY + SLACK, "{index}: {begins:?}"); + } + } + } + assert_eq!(speaker.place(ms(500), FRAME, TARGET, at(1_000)), Place::Play(at(1_150))); + assert_eq!(clock.due(stamp(500)), Some(at(1_150))); + } + + /// A frame sounds when placed: the first, one that follows it, and ones + /// written after the channel ran dry, on time or late, which the bus would + /// otherwise play at once or pad behind, apart from their pictures. One + /// written after its moment loses what has passed, or all of it. + #[test] + fn a_write_sounds_when_placed() { + let (bus, mut driver) = moq_audio::bus::Bus::new(Default::default()).unwrap(); + let config = bus.config().clone(); + let rate = u64::from(config.sample_rate); + let strip = moq_audio::bus::Strip::default(); + let mut input = moq_audio::encode::Input::new(config.sample_rate, config.layout); + input.format = moq_audio::Format::F32; + let channel = bus.channel(input, strip.clone()).unwrap(); + let mut writer = Writer::new(channel, config.sample_rate, config.layout.channels()); + let mix = bus.mix(); + mix.route(&strip); + let mut reader = mix.reader(); + + // The bus clock, as an instant, and a frame of `value` on it. + let block = rate / 100; + let clock = |frames: u64| Instant::ZERO + Duration::from_micros(frames * 1_000_000 / rate); + let frame = |value: f32| bytemuck::cast_slice::(&vec![value; 2 * 960]).to_vec(); + let mut heard = Vec::new(); + let mut position = 0; + let mut advance = |until: u64, heard: &mut Vec<(u64, f32)>| { + while position < until { + driver.advance(block as usize); + position += block; + let std::task::Poll::Ready(Some(read)) = reader.poll_read(&kio::Waiter::noop()) else { + panic!("an advance publishes a block"); + }; + let mut last = heard.last().map_or(0.0, |(_, value)| *value); + for (index, sample) in read.data.chunks(2).enumerate() { + if sample[0] != last { + last = sample[0]; + heard.push((read.position + index as u64, last)); + } + } + } + }; + + advance(block, &mut heard); + let sounds = clock(block) + BUS_LATENCY; + writer.write(sounds, &frame(1.0), clock(block)).unwrap(); + // The next one, on time. + advance(block + 960, &mut heard); + writer.write(sounds + FRAME, &frame(2.0), clock(block + 960)).unwrap(); + // Long after both ran out. + advance(rate, &mut heard); + writer + .write(clock(rate) + BUS_LATENCY, &frame(3.0), clock(rate)) + .unwrap(); + advance(2 * rate, &mut heard); + // Run dry and written late, with less than the latency to go. + advance(2 * rate + 1_920, &mut heard); + writer + .write(clock(2 * rate) + BUS_LATENCY, &frame(4.0), clock(2 * rate + 1_920)) + .unwrap(); + advance(3 * rate, &mut heard); + // Run dry and written after its moment began, then after it ended. + writer + .write( + clock(3 * rate) - Duration::from_millis(10), + &frame(5.0), + clock(3 * rate), + ) + .unwrap(); + advance(4 * rate, &mut heard); + writer + .write(clock(4 * rate) - FRAME, &frame(6.0), clock(4 * rate)) + .unwrap(); + advance(5 * rate, &mut heard); + + let latency = rate * BUS_LATENCY.as_millis() as u64 / 1_000; + let first = block + latency; + let after = rate + latency; + let late = 2 * rate + latency; + assert_eq!( + heard, + [ + (first, 1.0), + (first + 960, 2.0), + (first + 1_920, 0.0), + (after, 3.0), + (after + 960, 0.0), + (late, 4.0), + (late + 960, 0.0), + (3 * rate, 5.0), + (3 * rate + 480, 0.0) + ] + ); + } +} diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 96ad01c..5c7ca86 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -9,7 +9,8 @@ //! publishes it as its own track in the studio's local broadcast, web pages //! aside; see [`crate::publish`]. -use std::collections::BTreeMap; +use std::cell::{Cell, RefCell}; +use std::collections::{BTreeMap, VecDeque}; use std::future::Future; use std::path::{Path, PathBuf}; use std::pin::Pin; @@ -20,12 +21,14 @@ use hang::catalog::{AudioConfig, Reference, VideoConfig}; use kio::clock::Instant; use moq_audio::bus::{Bus, Mix, Strip}; use moq_mux::catalog::Stream; -use tokio::sync::watch; +use moq_mux::timeline; +use tokio::sync::{Notify, watch}; use crate::Error; use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; +use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -115,6 +118,19 @@ impl Stall { /// One instance of a broadcast being decoded. type Playing<'a> = Pin> + 'a>>; +/// The playout clock one instance of a broadcast's audio and video share. +struct Clock { + timeline: timeline::Producer, + /// Woken when audio places a frame or stops, which may move the anchor or + /// let a picture waiting for its sound show. + changed: Notify, +} + +/// Decoded pictures a broadcast source holds until they are due. The rest wait +/// undecoded in the subscription, as in `moq play`, so a deep playout target +/// holds no more decoder output than this. +const QUEUED: usize = 3; + /// What a running source is doing. #[derive(Clone, Debug, PartialEq, Eq)] #[non_exhaustive] @@ -288,6 +304,8 @@ pub struct Feed { played: Playback, /// A replay source's next replay. plays: Option>, + /// A broadcast source's playout clock. + clock: Option, } impl Feed { @@ -370,6 +388,14 @@ impl Feed { (since >= STALL).then(|| stall(since)) } + /// How far a live broadcast source plays behind what arrives, to ride out + /// network jitter: its audio's playout target, which video follows, or + /// without audio, video's own short delay. `None` for any other source. + pub fn delay(&self) -> Option { + let clock = self.clock.as_ref()?; + (self.status() == Status::Live).then(|| clock.target().unwrap_or(VIDEO_DELAY)) + } + /// The source's gain, mute, delay, and meter in every mix, if it has sound. pub fn strip(&self) -> Option<&Strip> { self.strip.as_ref() @@ -405,6 +431,7 @@ pub struct Driver { track: Option, played: Playback, plays: Option>, + clock: Option, } /// Where a source's own track goes, its claim on the local catalog's listing @@ -720,6 +747,7 @@ impl Driver { let mut instances = moq_mux::follow::Instances::new(&self.context.origin, path, None)?; set(&self.status, Status::Offline); + let mut clock = self.clock.clone().expect("a broadcast source has a clock"); let mut playing: Option> = None; loop { tokio::select! { @@ -730,7 +758,9 @@ impl Driver { broadcast: Some(broadcast.clone()), ..Played::default() }; - playing = Some(Box::pin(self.play(broadcast, path))); + // Each instance's timestamps are its own. + clock = clock.reset(VIDEO_DELAY); + playing = Some(Box::pin(self.play(broadcast, path, clock.clone()))); } Ok(moq_mux::follow::Event::End) => { playing = None; @@ -752,17 +782,27 @@ impl Driver { /// Decode one instance of a broadcast: the first video and audio renditions /// each catalog offers, switching when a catalog update changes them, until - /// the catalog ends. - async fn play(&self, broadcast: moq_net::broadcast::Consumer, path: &str) -> Result<(), Error> { + /// the catalog ends. Both play on `clock`. + async fn play( + &self, + broadcast: moq_net::broadcast::Consumer, + path: &str, + clock: timeline::Producer, + ) -> Result<(), Error> { let format = moq_mux::catalog::CatalogFormat::detect(path).unwrap_or_default(); let mut catalogs = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await?; + let clock = Clock { + timeline: clock, + changed: Notify::new(), + }; // A track name always means the same content, so only a new track // restarts a decoder. That is the track an entry resolves to, not its // catalog ID: a publisher may keep the ID and point it at a replacement // track, as the JS publisher does on a codec change. A metadata-only - // update, such as a jitter estimate, leaves playback alone. - let mut video: Option = None; + // update, such as a jitter estimate, leaves playback alone. Video also + // restarts when audio comes or goes, which changes how long it waits. + let mut video: Option<(String, bool)> = None; let mut audio: Option = None; let mut video_playing: Playing<'_> = Box::pin(std::future::pending()); let mut audio_playing: Playing<'_> = Box::pin(std::future::pending()); @@ -773,28 +813,33 @@ impl Driver { let Some(catalog) = catalog? else { return Ok(()); }; - let next = pick(catalog.video.enabled())?; - let track = next.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); + let next_video = pick(catalog.video.enabled())?; + let next_audio = pick(catalog.audio.enabled())?; + let track = next_video + .as_ref() + .map(|(id, config)| (Reference::track(config, id).to_string(), next_audio.is_some())); if track != video { video = track; let mut played = self.played.lock().unwrap(); - played.video = next.clone(); + played.video = next_video.clone(); // The newest picture was the old track's, a moment the // new one may not hold. played.newest = None; drop(played); - video_playing = match next { - Some((id, config)) => Box::pin(self.play_video(&broadcast, id, config)), + // Video waits for its audio as long as audio's target can + // grow; alone, it waits only its own delay. + let max = if next_audio.is_some() { MAX_DELAY } else { VIDEO_DELAY }; + video_playing = match next_video { + Some((id, config)) => Box::pin(self.play_video(&broadcast, id, config, &clock, max)), None => Box::pin(std::future::pending()), }; } - let next = pick(catalog.audio.enabled())?; - let track = next.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); + let track = next_audio.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); if track != audio { audio = track; - self.played.lock().unwrap().audio = next.clone(); - audio_playing = match next { - Some((id, config)) => Box::pin(self.play_audio(&broadcast, id, config)), + self.played.lock().unwrap().audio = next_audio.clone(); + audio_playing = match next_audio { + Some((id, config)) => Box::pin(self.play_audio(&broadcast, id, config, &clock)), None => Box::pin(std::future::pending()), }; } @@ -812,30 +857,76 @@ impl Driver { } } + /// Show each picture once `clock` says it is due, the newest due one when + /// several are, holding a few decoded ones for their moments. The canvas + /// keeps showing the last picture meanwhile. async fn play_video( &self, broadcast: &moq_net::broadcast::Consumer, name: String, config: hang::catalog::VideoConfig, + clock: &Clock, + max: Duration, ) -> Result<(), Error> { + // A replacement track's stamps need not continue the old one's. + clock.timeline.video_restarted(); let mut options = moq_video::decode::Options::new(); options.start = moq_video::decode::Start::Latest; options.decoder.output = self.context.output; + options.delay = moq_video::decode::Delay { + min: Some(VIDEO_DELAY), + max: Some(max), + }; let mut consumer = moq_video::decode::Consumer::new(broadcast, &config, name, options).await?; - while let Some(read) = consumer.read().await? { - if let moq_video::decode::Read::Frame(frame) = read { + let schedule = clock.timeline.consume(); + let mut queue = VecDeque::::new(); + let mut ended = false; + loop { + let now = kio::clock::now(); + // Counted first: `present` holds the clock's lock while it iterates. + let heard = queue + .iter() + .take_while(|frame| crate::playout::heard(&schedule, frame.timestamp, now)) + .count(); + let ready = queue.iter().take(heard).map(|frame| frame.timestamp); + if let Some(index) = schedule.present(ready, now) { + let frame = queue.drain(..=index).next_back().expect("an index into the queue"); self.played.lock().unwrap().newest = Some(frame.timestamp); self.show(Picture::Frame(frame)); } + if ended && queue.is_empty() { + return Ok(()); + } + // A picture due but not yet heard waits for audio to place more. + let due = queue + .front() + .and_then(|frame| schedule.due(frame.timestamp)) + .filter(|due| *due > now); + tokio::select! { + read = consumer.read(), if !ended && queue.len() < QUEUED => match read? { + Some(moq_video::decode::Read::Frame(frame)) => { + clock.timeline.video(frame.timestamp, kio::clock::now()); + queue.push_back(frame); + } + Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => { + clock.timeline.video_restarted(); + } + None => ended = true, + }, + () = kio::clock::sleep_until(due.unwrap_or(now)), if due.is_some() => {} + () = clock.changed.notified() => {} + } } - Ok(()) } + /// Play the broadcast's sound into the bus on its timestamps, holding the + /// jitter estimate moq-audio's decoder makes, and anchor `clock` on it. async fn play_audio( &self, broadcast: &moq_net::broadcast::Consumer, name: String, config: hang::catalog::AudioConfig, + clock: &Clock, ) -> Result<(), Error> { let Some(strip) = self.strip.clone() else { return Ok(()); @@ -843,6 +934,10 @@ impl Driver { let bus = self.context.bus.config().clone(); let mut options = moq_audio::decode::Options::new(); options.start = moq_audio::decode::Start::Latest; + options.delay = moq_audio::decode::Delay { + min: None, + max: Some(MAX_DELAY), + }; options.output.format = moq_audio::Format::F32; options.output.sample_rate = Some(bus.sample_rate); options.output.layout = Some(bus.layout); @@ -850,18 +945,65 @@ impl Driver { let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); input.format = moq_audio::Format::F32; - let mut channel = self.context.bus.channel(input, strip)?; - // Gaps and encoder boundaries need nothing. Like a picture, which shows - // on arrival, the bus plays a source at its live edge on the program - // clock: a stall plays as silence while it lasts, and a burst after it - // is bounded by the bus's buffer. Replaying the publisher's - // timeline, holes included, would put the audio behind the live edge - // and its own video. - while let Some(read) = consumer.read().await? { - if let moq_audio::decode::Read::Frame(frame) = read { - channel.write(&frame.data)?; + let channel = self.context.bus.channel(input, strip)?; + let mut writer = Writer::new(channel, bus.sample_rate, bus.layout.channels()); + let speaker = RefCell::new(Speaker::new(clock.timeline.clone(), &clock.changed)); + let stride = bus.layout.channels() as usize * size_of::(); + let rate = f64::from(bus.sample_rate); + + // Frames wait here for their moment, bounded by the target the speaker + // holds, while every arrival is read at once: the jitter estimate times + // each one as `read` hands it over. + let queue = RefCell::new(VecDeque::new()); + let queued = Notify::new(); + let received = Cell::new(false); + let receive = async { + while let Some(next) = consumer.read().await? { + let moq_audio::decode::Read::Frame(frame) = next else { + // A hole or an encoder restart shows in the stamps that follow. + continue; + }; + let duration = Duration::from_secs_f64((frame.data.len() / stride) as f64 / rate); + let now = kio::clock::now(); + let place = speaker + .borrow_mut() + .place(frame.timestamp.into(), duration, consumer.delay(), now); + if let Place::Play(sounds) = place { + queue.borrow_mut().push_back((sounds, frame)); + queued.notify_one(); + } } - } + received.set(true); + queued.notify_one(); + Ok::<_, Error>(()) + }; + // Writes each frame its bus latency ahead, and tells the clock about + // the frames after a skip once they sound. + let write = async { + loop { + let now = kio::clock::now(); + speaker.borrow_mut().commit(now); + let write = queue + .borrow() + .front() + .map(|(sounds, _)| *sounds - crate::playout::BUS_LATENCY); + if write.is_some_and(|at| at <= now) { + let (sounds, frame) = queue.borrow_mut().pop_front().expect("peeked"); + writer.write(sounds, &frame.data, now)?; + continue; + } + let wake = write.into_iter().chain(speaker.borrow().deferred()).min(); + match wake { + Some(at) => tokio::select! { + () = kio::clock::sleep_until(at) => {} + () = queued.notified() => {} + }, + None if received.get() => return Ok::<_, Error>(()), + None => queued.notified().await, + } + } + }; + tokio::try_join!(receive, write)?; Ok(()) } @@ -1076,6 +1218,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { }); let played = Playback::default(); + let clock = matches!(source, Source::Broadcast { .. }).then(|| timeline::Producer::new(VIDEO_DELAY)); let (plays, replays) = match source { Source::Replay => { let (plays, replays) = watch::channel(Next::default()); @@ -1094,6 +1237,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { track: track.as_ref().map(|track| track.status.consume()), played: played.clone(), plays, + clock: clock.as_ref().map(timeline::Producer::consume), }; let driver = Driver { name: name.to_string(), @@ -1107,6 +1251,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { track, played, plays: replays, + clock, }; (feed, driver) } @@ -1617,6 +1762,181 @@ mod tests { .await; } + /// A live broadcast at `room/guest`, writing nothing until the test does. + struct Live { + _publication: moq_net::origin::Publication, + broadcast: moq_net::broadcast::Producer, + catalog: moq_mux::catalog::Producer, + video: moq_video::encode::Producer, + encoder: moq_video::encode::Encoder, + audio: Option, + } + + /// A video track on its own timestamp base. + async fn rendition( + broadcast: &moq_net::broadcast::Producer, + timebase: moq_mux::catalog::Timebase, + ) -> (moq_video::encode::Producer, moq_video::encode::Encoder) { + let mut config = moq_video::encode::Config::new(64, 64, moq_video::FrameRate::new(30, 1).unwrap()); + config.kind = moq_video::encode::Kind::Software; + let rendition = config.probe().await.unwrap(); + let reserved = timebase.reserve().with_interval(config.gop.duration(config.framerate)); + let video = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition).unwrap(); + (video, moq_video::encode::Encoder::new(&config).unwrap()) + } + + /// A broadcast with a video track, and an audio one with `audio`. + async fn live(origin: &moq_net::origin::Producer, audio: bool) -> Live { + let publication = origin.create_broadcast("room/guest").unwrap(); + let mut broadcast = publication.broadcast().clone(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let (video, encoder) = rendition(&broadcast, catalog.timebase()).await; + let audio = audio.then(|| { + let input = moq_audio::encode::Input::default(); + let mut options = moq_audio::encode::Options::default(); + options.config = moq_audio::encode::Config::from_input(moq_audio::encode::Codec::Pcm, &input); + moq_audio::encode::Producer::new(&mut broadcast, catalog.clone(), input, &options).unwrap() + }); + publication.announce(moq_net::origin::Route::default()).unwrap(); + Live { + _publication: publication, + broadcast, + catalog, + video, + encoder, + audio, + } + } + + /// Encode and publish one gray picture stamped `at`. + fn picture(producer: &mut moq_video::encode::Producer, encoder: &mut moq_video::encode::Encoder, at: u64) { + let gray = vec![0x80; 64 * 64 * 4]; + let surface = moq_video::Surface::rgba(&gray, moq_video::Size::new(64, 64)).unwrap(); + let frame = moq_video::Frame::new(surface, moq_net::Timestamp::from_millis(at).unwrap()); + producer.publish(&encoder.encode(&frame).unwrap()).unwrap(); + } + + /// The picture a feed shows next, once it shows one. + async fn shown(feed: &Feed) -> Duration { + until(|| match feed.take() { + Some(Picture::Frame(frame)) => Some(frame.timestamp.into()), + Some(_) => panic!("a broadcast shows only frames"), + None => None, + }) + .await + } + + /// Without audio to follow, a broadcast's picture shows [`VIDEO_DELAY`] + /// after it arrives, not on arrival. + #[tokio::test(start_paused = true)] + async fn a_picture_waits_for_its_moment() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let mut live = live(&origin, false).await; + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection()); + start(&sources, drivers).await; + let feed = sources.get("guest").unwrap(); + let clock = feed.clock.clone().unwrap(); + + until_time(kio::clock::now()).await; + let arrived = kio::clock::now(); + picture(&mut live.video, &mut live.encoder, 1_000); + let stamp = moq_net::Timestamp::from_millis(1_000).unwrap(); + let due = until(|| clock.due(stamp)).await; + assert_eq!(due, arrived + VIDEO_DELAY); + assert!(feed.take().is_none(), "shown on arrival"); + until_time(due).await; + assert_eq!(shown(feed).await, Duration::from(stamp)); + assert_eq!(feed.delay(), Some(VIDEO_DELAY)); + }) + .await; + } + + /// A video track replaced by one stamped earlier starts its own delay over, + /// rather than showing every picture on arrival against the old track's. + #[tokio::test(start_paused = true)] + async fn a_replacement_track_keeps_its_delay() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let mut live = live(&origin, false).await; + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection()); + start(&sources, drivers).await; + let feed = sources.get("guest").unwrap(); + let clock = feed.clock.clone().unwrap(); + picture(&mut live.video, &mut live.encoder, 1_000); + let first = moq_net::Timestamp::from_millis(1_000).unwrap(); + let due = until(|| clock.due(first)).await; + until_time(due).await; + assert_eq!(shown(feed).await, Duration::from(first)); + + // Its stamps sit half a second before the first track's clock reading. + let timebase = live.catalog.timebase(); + timebase + .place( + moq_net::Timestamp::ZERO, + kio::clock::wall() - Duration::from_millis(500), + ) + .unwrap(); + let (video, mut encoder) = rendition(&live.broadcast, timebase).await; + drop(std::mem::replace(&mut live.video, video)); + until_time(kio::clock::now()).await; + let arrived = kio::clock::now(); + picture(&mut live.video, &mut encoder, 0); + let moved = until(|| match feed.take() { + Some(Picture::Frame(frame)) => panic!("{:?} shown on arrival", frame.timestamp), + _ => clock.due(first).filter(|at| *at != due), + }) + .await; + assert!(moved > due, "the replacement re-anchored behind the first track"); + until_time(arrived + VIDEO_DELAY - Duration::from_millis(1)).await; + assert!(feed.take().is_none(), "shown before its delay"); + until_time(arrived + VIDEO_DELAY).await; + assert!( + shown(feed).await < Duration::from(first), + "stamped before the first track" + ); + }) + .await; + } + + /// A picture waits for the sound stamped with it, which plays the jitter + /// estimate after it arrives, rather than showing on its own delay. + #[tokio::test(start_paused = true)] + async fn a_picture_waits_for_its_sound() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let mut live = live(&origin, true).await; + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection()); + start(&sources, drivers).await; + let feed = sources.get("guest").unwrap(); + let clock = feed.clock.clone().unwrap(); + + until_time(kio::clock::now()).await; + let arrived = kio::clock::now(); + let silence = vec![0u8; 960 * 2 * size_of::()]; + let stamp = moq_net::Timestamp::from_millis(1_000).unwrap(); + let audio = live.audio.as_mut().unwrap(); + audio.write(&moq_audio::Frame::new(silence.into(), stamp)).unwrap(); + picture(&mut live.video, &mut live.encoder, 1_000); + let target = until(|| clock.target()).await; + assert!(target + crate::playout::BUS_LATENCY > VIDEO_DELAY, "{target:?}"); + let due = arrived + target + crate::playout::BUS_LATENCY; + until(|| (clock.due(stamp) == Some(due)).then_some(())).await; + until_time(due - Duration::from_millis(1)).await; + assert!(feed.take().is_none(), "shown before its sound"); + until_time(due).await; + assert_eq!(shown(feed).await, Duration::from(stamp)); + assert_eq!(feed.delay(), Some(target)); + }) + .await; + } + /// Dropping a feed stops its driver, which is what releases a device. #[tokio::test] async fn a_dropped_feed_stops_its_driver() { diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 71615b0..8ab43f2 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1324,6 +1324,13 @@ impl Panels { status(ui, &track); }); } + // A broadcast publishes no track either, and shows how far + // behind its publisher it plays. + None if let Some(delay) = feed.and_then(Feed::delay) => { + ui.weak(format!("{} ms behind", delay.as_millis())).on_hover_text( + "Plays this far behind what arrives, to ride out the network's jitter; the picture waits for its sound", + ); + } None => match feed.and_then(Feed::page).and_then(|page| page.paint()) { Some(Paint::Gpu) => { ui.weak("GPU").on_hover_text("The page is copied on the GPU");