diff --git a/Cargo.lock b/Cargo.lock index bf66dd80..7e2b51b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1109,9 +1109,9 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "combine" -version = "4.6.8" +version = "4.6.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e" +checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92" dependencies = [ "bytes", "memchr", @@ -3715,6 +3715,17 @@ dependencies = [ "hax-lib 0.4.2", ] +[[package]] +name = "libcrux-sha2" +version = "0.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "86aa35d8fc9576a41df34af4d95d1f11186c1f65cd6c829329748915ea556d71" +dependencies = [ + "libcrux-hacl-rs", + "libcrux-macros", + "libcrux-traits", +] + [[package]] name = "libcrux-traits" version = "0.0.9" @@ -3831,9 +3842,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.34" +version = "0.4.35" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +checksum = "5d15f86d12da44cd8d4ec3c73984537eb7f4cadcb9776a575a0a8a05ca11005c" [[package]] name = "lru-slab" @@ -5935,9 +5946,9 @@ checksum = "3d595e54a326bc53c1c197b32d295e14b169e3cfeaa8dc82b529f947fba6bcf5" [[package]] name = "prometheus-client" -version = "0.25.1" +version = "0.25.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1784dc11a05f5a8f57c4a62915686be140f24f70301290b997e317241fa9ffc2" +checksum = "6d150ad7bf80e60f45769d18a0a8c01028ab8f56355f33ed7bfce3a31f82e696" dependencies = [ "dtoa", "itoa", @@ -6351,9 +6362,9 @@ dependencies = [ [[package]] name = "rind" -version = "0.1.12" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "422500e50252ad3bd319f359c54431441711a8fd55cb228973dde123ff8c2039" +checksum = "5b1b559e7ba7aa39bc4aa9b581dd2254919b3434d50669898bb8c8e7b154cfb8" dependencies = [ "cfg-if", "curve25519-dalek", @@ -6364,14 +6375,15 @@ dependencies = [ "libcrux-hacl-rs", "libcrux-macros", "libcrux-poly1305", + "libcrux-sha2", "libcrux-traits", ] [[package]] name = "rind-jsonwebtoken" -version = "0.1.2" +version = "11.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3dbbb3e3d229205e18f9a5b3ee418071a29eacdb61aec66a5f34eeffef8e55f" +checksum = "dd34e9d4695a1a34a4d8fc31d3db5acda9a24cfd2ca90ffbd36f7c62dc4bb7eb" dependencies = [ "jsonwebtoken", "pkcs1", @@ -6380,9 +6392,9 @@ dependencies = [ [[package]] name = "rind-rcgen" -version = "0.1.12" +version = "0.14.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b0f7d386858c030aff879c224a0c3d59b02e7ff953ca96214825a3e77ecba76" +checksum = "8c29833453b6f7def587d905c234d65e9172045504b77300084614d5df23dfdf" dependencies = [ "rcgen", "rind", @@ -6392,9 +6404,9 @@ dependencies = [ [[package]] name = "rind-rustls" -version = "0.1.12" +version = "0.23.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e67845e30d1c9a194485cdcb585abf7dcc011a4413488d57634ec39d7e832f0" +checksum = "8f91c7ed15f653d52782f482dbea8d2ec350b49ee88a911660acf539e7cf6752" dependencies = [ "rind", "rustls", @@ -6416,9 +6428,9 @@ dependencies = [ [[package]] name = "ringbuf" -version = "0.5.3" +version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b58b631e416dd7d84549d7e6a50f767e111f9b03e30d04f40072f6618f386aaf" +checksum = "33d5f59e4145df5f29e84b6389394a8a3fedf7bd8c12570560ae67a917e6eae3" dependencies = [ "crossbeam-utils", "portable-atomic", @@ -7603,9 +7615,9 @@ dependencies = [ [[package]] name = "sysinfo" -version = "0.39.6" +version = "0.40.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2071df9448915b71c4fe6d25deaf1c22f12bd234f01540b77312bb8e41361e6" +checksum = "0e8a3edbe0394ce115f3917a00349dd242be09eb5a091ed509eaf67b55cfb8a8" dependencies = [ "libc", "memchr", @@ -7669,7 +7681,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix 1.1.6", "windows-sys 0.61.2", @@ -8432,21 +8444,21 @@ dependencies = [ [[package]] name = "usage-argv" -version = "7.1.3" +version = "7.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2cc40affbd6c11f829161c500caeb1518f1bd8729a7d2584b963af88b314c592" +checksum = "e4876429514ea5b79a0a41546c5087db5c19d502405321098868874ea940ebab" [[package]] name = "usage-config" -version = "7.1.3" +version = "7.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57f48b8d49761fe8f2f8ec72f73351d3c4c455ca5afac44587f4df54b15c0fb3" +checksum = "28ebe3c1e68c79d1be763d537c61005bad8ac43b6567e6ab279c24c227230a4f" [[package]] name = "usage-derive" -version = "7.1.3" +version = "7.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8f99cca010b31bda87f03718e349a04fcdffd4b374e1c3644695a4f03a6f2fe" +checksum = "ccf026069d7f6a6cc0e691a277906829279e702c1b73c8666f9931928be5a8bf" dependencies = [ "proc-macro2", "quote", @@ -8455,9 +8467,9 @@ dependencies = [ [[package]] name = "usage-rs" -version = "7.1.3" +version = "7.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dcc3ab034ff1f5a62f5e0e19557395ac920734ea3ba8a365e7a4fd375b09c555" +checksum = "1d8a33544b3cf4d61ee879878737cb8baca7ffbcf149631ff611447c9aa83bec" dependencies = [ "usage-argv", "usage-config", @@ -8467,9 +8479,9 @@ dependencies = [ [[package]] name = "usage-test" -version = "7.1.3" +version = "7.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "142d2bec8a20ff24d4fb351d4a45d339926b73472d334d32776d7c676fb0d0ab" +checksum = "300900090803c08c8f04456302ed54aa2f32da5caa4cdbbeb1607a9ec7dd313d" dependencies = [ "usage-argv", ] @@ -9619,9 +9631,9 @@ dependencies = [ [[package]] name = "yuv" -version = "0.8.19" +version = "0.8.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a295923535c00a4e72a50a0a1130097f93160b39892dd4d2d006b1e2b90d6bf9" +checksum = "3d109d1eacc7b034329873d2d848039b365c65ab7d751b19ee84e4921c62c607" dependencies = [ "num-traits", ] diff --git a/README.md b/README.md index a3b91ff6..6a1def8f 100644 --- a/README.md +++ b/README.md @@ -303,15 +303,15 @@ 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. +so its sound and picture stay in sync through network jitter. It plays as far +behind what arrives as the network's jitter needs, measured as it goes as +`moq play --delay auto` does, plus 50 ms to hand its sound to the mix, up to +2 s in all. Sound and picture each +keep that schedule on their own: a sound stall leaves the picture playing, +and the sound comes back on the clock, skipping what it missed; a picture +stall leaves the sound playing, with the canvas showing the last picture +meanwhile. Without audio, the picture plays 100 ms behind. Sources shows how +far behind each broadcast plays. ### Audio scenes diff --git a/moq b/moq index 79960a9d..644d4899 160000 --- a/moq +++ b/moq @@ -1 +1 @@ -Subproject commit 79960a9dd64e48a2fe51a06fc1a3e9b4692dfb22 +Subproject commit 644d48996e908896ad6f2e9c2bd12439a1d37f7f diff --git a/quest/m0/pin-main.md b/quest/m0/pin-main.md index 16a1a21e..ad16fae2 100644 --- a/quest/m0/pin-main.md +++ b/quest/m0/pin-main.md @@ -3,8 +3,9 @@ ## Goal The `moq` submodule is at moq's `main` as of the day this quest starts, and -`just ci` passes on it. Every breaking change between `644d48996` (where -[pin-sync](/quest/m0/pin-sync.md) leaves it) and that commit is fixed in one +`just ci` passes on it. Every breaking change between `644d48996` +([moq-dev/moq#8272](https://github.com/moq-dev/moq/pull/8272), where the +studio's pin sits) and that commit is fixed in one PR, so each later wait-gated pin moves past only its own upstream change. ## Plan @@ -36,7 +37,3 @@ On 2026-11-09, moq's `main` (`103052ed0`) was 117 first-parent commits past Each change's PR has an Impact section with the migration. If an upstream change a pin waits for is already on `main` when this starts, this PR also does that pin's work and deletes the pin quest. - -## Required - -- [moq-mux's decoupled playout clock is pinned](/quest/m0/pin-sync.md) - the playout rewrite onto `moq_mux::sync`, which this builds on diff --git a/quest/m0/pin-sync.md b/quest/m0/pin-sync.md deleted file mode 100644 index a3228882..00000000 --- a/quest/m0/pin-sync.md +++ /dev/null @@ -1,68 +0,0 @@ -# [M] moq-mux's decoupled playout clock is pinned - -## Goal - -The `moq` submodule is at `644d48996`, -[moq-dev/moq#8272](https://github.com/moq-dev/moq/pull/8272), which replaces -`moq_mux::timeline` with `moq_mux::sync`, and a broadcast source's playout -(`rs/moq-scene/src/playout.rs`, `rs/moq-scene/src/source.rs`) runs on the new -clock. `README.md`'s broadcast-source paragraph under "Keyframes" says what a -stall does on it. - -## Plan - -Decided 2026-11-09 (maintainer, /quest-audit): ✅ crossing #8272 is this -quest, after playout's writes moved onto positioned bus channels -([moq-dev/studio#137](https://github.com/moq-dev/studio/pull/137)) -(rejected: crossing it in the positioned bus pin). Both rewrite -`playout.rs`, so the positioned work finishes on the API it was planned -against, then this crosses. - -Advance with `just upgrade-moq 644d48996` in its own PR. #8272's Impact holds -the migration: - -- `timeline::Kind` is `moq_mux::container::Kind`, which `Producer::track` - takes. -- `Producer::video(ts, now)` is `track.received(ts, now)` on the video - `Track`. `Producer::audio(Audio { .. }, now)` is `track.received(ts, - arrival)` on the audio `Track` plus `track.set_target(jitter_target)`, and - `timeline::Audio` is gone. -- `video_restarted` and `audio_restarted` are `track.reset()`; - `audio_stopped` is dropping or retiring the track. -- `Consumer::target()` is `Consumer::delay()`, always resolved. - `Consumer::playhead()` is gone, which `playout::heard` reads: the speaker - tracks what it sounds. -- Audio no longer owns the clock. Each frame sounds at `producer.due(..)`, - which the `Speaker` steers toward itself. - -The other breaking changes in `79960a9dd..644d48996` that reach studio code, -read against the tree on 2026-11-09; `just ci` finds any other: - -- [moq-dev/moq#8274](https://github.com/moq-dev/moq/pull/8274): - `mix.minus(strip)` takes every strip to remove, `mix.minus([&strip])` - (`rs/moq-studio/src/feed.rs`). -- [moq-dev/moq#8269](https://github.com/moq-dev/moq/pull/8269): - `moq_video::encode::Producer::new` takes a track, not a broadcast, and - `with_track` is gone; `render::Exporter::new` takes `&encode::Config` and - returns `(Exporter, encode::Config, VideoConfig)`. - -Not crossed here: [moq-dev/moq#8270](https://github.com/moq-dev/moq/pull/8270) -(`241d2ce92`), [moq-dev/moq#8372](https://github.com/moq-dev/moq/pull/8372) -(`147327e14`), and -[moq-dev/moq#8220](https://github.com/moq-dev/moq/pull/8220) (`16a673036`, -kio's `Instant` loses `+` and `-` with a `Duration`, which `playout.rs` and -`source.rs` use throughout) all merged after #8272. -[pin-main](/quest/m0/pin-main.md) fixes them. - -Decided 2026-11-12 (maintainer, /quest-unblock): ✅ adopt moq's decoupled -rule for a sound stall (rejected: keeping the studio's rule, with a `Speaker` -playhead of its own that holds the picture after a stall). The picture keeps -playing through a sound stall, and the sound comes back on the clock: each -frame is written at the bus position `due` gives it, and the late part of a -burst is trimmed, as the positioned writer already does. The studio plays a -broadcast as `moq play` and the browser do, following the 2026-11-05 ruling on -[moq-dev/moq#7941](https://github.com/moq-dev/moq/pull/7941): lip sync in the -steady state, desync under congestion, and one slow track never stalls or -moves another. `README.md` drops "Each picture waits for the sound stamped -with it" and "A sound stall longer than that delay pauses the picture too and -starts both again together", and says the picture plays on through a stall. diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index eb7b6161..15e4ed24 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -1,24 +1,26 @@ //! 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. +//! publisher's timestamps, against one [`sync`] clock per broadcast, as in +//! `moq play`. //! -//! 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. +//! Each track folds its arrivals into the clock, which plays the largest +//! jitter target any track holds behind the earliest of them. In the steady +//! state, sound and pictures stamped alike are due together. A track that +//! stalls only arrives late, and never holds or moves the other: the picture +//! plays on through a sound stall, and the sound comes back on the clock, the +//! part of it already due trimmed. Without audio, video plays [`VIDEO_DELAY`] +//! behind its arrivals. //! -//! `moq play` holds its jitter buffer in the speaker. Here a [`Speaker`] holds -//! it: it places each frame the playout target moq-audio's decoder estimates -//! after it arrives, and a [`Writer`] hands it to the bus [`WRITE_LEAD`] ahead -//! of then, at the bus position that moment maps to. A media or replay -//! source's sound goes through a [`Writer`] too, so it sounds with the -//! pictures it shows. +//! A [`Writer`] hands each sound to the bus [`WRITE_LEAD`] ahead of its +//! moment, at the bus position that moment maps to. A media or replay source's +//! sound goes through a [`Writer`] too, so it sounds with the pictures it +//! shows. use std::time::Duration; use kio::clock::Instant; use moq_audio::bus; -use moq_mux::timeline; -use tokio::sync::Notify; +use moq_mux::sync; +use tokio::sync::watch; use crate::source::Mixed; @@ -26,7 +28,7 @@ use crate::source::Mixed; /// `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 +/// The longest playout delay, 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. @@ -44,187 +46,87 @@ const _: () = assert!( "a sound handed over the lead ahead is in the bus's reach" ); -/// 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. +/// One track's hold on its broadcast's clock: its arrivals, and for audio its +/// jitter target. Every move of the schedule is sent on `moved`, since a track +/// asleep on a deadline computed before it is due earlier or later now. /// -/// 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)>, +/// Dropping it releases its hold, which may move the schedule too. +pub(crate) struct Hold<'a> { + track: sync::Track, + moved: &'a watch::Sender<()>, } -/// 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 { +impl<'a> Hold<'a> { + /// Hold `clock` for a track of `kind`, the only one of that kind on it. + pub(crate) fn new(clock: &sync::Producer, kind: moq_mux::container::Kind, moved: &'a watch::Sender<()>) -> Self { Self { - clock, - changed, - tail: None, - skipped: None, - unreported: None, + track: clock.track(kind), + moved, } } - /// 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 + WRITE_LEAD, - Some((sounds, media)) => { - let follows = sounds + start.saturating_sub(media); - // A write half its lead late is still ahead of the mix, however - // coarsely the bus advances. - if follows < now + WRITE_LEAD / 2 { - self.clock.audio_restarted(); - self.skipped = None; - self.unreported = None; - now + target + WRITE_LEAD - } else if start.saturating_add(CONTIGUOUS) < media - || follows.saturating_duration_since(now) > target + WRITE_LEAD + 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); + /// Fold a frame stamped `timestamp` that arrived at `now` into the clock. + pub(crate) fn received(&self, timestamp: moq_net::Timestamp, now: Instant) { + if self.track.received(timestamp, now) { + self.moved.send_replace(()); } - 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); + /// Hold the audio's jitter `target`, plus the lead its writes need so a + /// frame on time is handed to the bus whole, up to [`MAX_DELAY`]. + pub(crate) fn set_target(&self, target: Duration) { + if self.track.set_target(target.saturating_add(WRITE_LEAD).min(MAX_DELAY)) { + self.moved.send_replace(()); } } - /// 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(); + /// End this track's timeline at lost media or an encoder restart: its + /// arrivals hold the clock again from the next frame. + pub(crate) fn reset(&self) { + self.track.reset(); + self.moved.send_replace(()); } } -impl Drop for Speaker<'_> { +impl Drop for Hold<'_> { fn drop(&mut self) { - self.clock.audio_stopped(); - self.changed.notify_one(); + self.moved.send_replace(()); } } -/// How soon a picture held by [`heard`] looks at the playhead again: another -/// broadcast's playhead moves, or re-anchors across a gap, without telling -/// anyone reading its clock, so a held picture polls it. -pub const RECHECK: Duration = Duration::from_millis(10); - -/// 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 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) -} +/// How soon a sound waiting on another broadcast's clock looks at it again: +/// that clock moves without telling anyone reading it. +pub(crate) const RECHECK: Duration = Duration::from_millis(10); -/// What to do at `now` with a sound stamped `timestamp` that plays on -/// `schedule`, another track's clock. +/// What to do with a sound at a moment. #[derive(Debug, PartialEq, Eq)] pub(crate) enum Follow { /// Write it to sound at this moment. Write(Instant), - /// Look again at this moment. + /// Look again at this moment, its lead before it sounds. Wait(Instant), - /// Its moment has passed, or the clock has none. + /// It has sounded through, or the clock has no moment for it. Drop, } -/// Decide what to do with a followed sound from the clock as it is now, not -/// as it was when the sound arrived, so a re-anchor moves it. The clock's own -/// sound stamped alike must have been written first: when that stalls, this -/// one waits with the pictures, until its moment has passed. -pub(crate) fn follow(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> Follow { - let Some(sounds) = schedule.due(timestamp) else { +/// What to do at `now` with a sound lasting `duration` that the clock says +/// sounds at `due`. Read `due` from the clock as it is now, not as it was when +/// the sound arrived, so a move of the schedule moves the sound. One already +/// partly due is still written: the [`Writer`] trims what the bus has mixed. +pub(crate) fn follow(due: Option, duration: Duration, now: Instant) -> Follow { + let Some(sounds) = due else { return Follow::Drop; }; - if sounds < now { + if sounds + duration <= now { return Follow::Drop; } - // The clock re-anchors without telling its readers, so a wait for a - // moment it may yet move earlier is rechecked too. let write = sounds - WRITE_LEAD; if write > now { - return Follow::Wait(write.min(now + RECHECK)); - } - if !heard(schedule, timestamp, sounds) { - return Follow::Wait(now + RECHECK); + return Follow::Wait(write); } Follow::Write(sounds) } @@ -329,155 +231,66 @@ mod tests { 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. + /// A sound is written its lead before its moment, and still once partly + /// due, for the writer to trim; once it has sounded through, or before the + /// clock has a moment for it, it is dropped. #[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"); + fn a_sound_is_written_its_lead_ahead() { + assert_eq!(follow(Some(at(150)), FRAME, at(0)), Follow::Wait(at(100))); + assert_eq!(follow(Some(at(150)), FRAME, at(100)), Follow::Write(at(150))); + assert_eq!(follow(Some(at(150)), FRAME, at(160)), Follow::Write(at(150))); + assert_eq!(follow(Some(at(150)), FRAME, at(170)), Follow::Drop); + assert_eq!(follow(None, FRAME, at(0)), Follow::Drop); } - /// A followed sound, rechecked until then, is written the bus's latency - /// before its moment once the clock's own sound stamped alike has been, - /// waits while that has not, and is dropped once its moment has passed. + /// Sound and pictures stamped alike are due together, the audio's target + /// and the write lead behind the first arrival. A sound stall leaves the + /// pictures on their moments, and the backlog that lands late is dropped + /// up to what is still to sound, which plays on the clock. #[test] - fn a_followed_sound_waits_for_the_clocks_own() { - let clock = timeline::Producer::new(VIDEO_DELAY); - let schedule = clock.consume(); - let changed = Notify::new(); - let mut speaker = Speaker::new(clock.clone(), &changed); - speaker.place(ms(1_000), FRAME, TARGET, at(0)); - assert_eq!( - follow(&schedule, stamp(1_000), at(0)), - Follow::Wait(at(10)), - "rechecked" - ); - assert_eq!(follow(&schedule, stamp(1_000), at(95)), Follow::Wait(at(100))); - assert_eq!(follow(&schedule, stamp(1_000), at(100)), Follow::Write(at(150))); - // Past what the clock's own sound has written: it stalled. - assert_eq!(follow(&schedule, stamp(1_100), at(200)), Follow::Wait(at(210))); - assert_eq!(follow(&schedule, stamp(1_100), at(250)), Follow::Wait(at(260))); - assert_eq!(follow(&schedule, stamp(1_100), at(251)), Follow::Drop); - let fresh = clock.clone().reset(VIDEO_DELAY).consume(); - assert_eq!(follow(&fresh, stamp(1_000), at(0)), Follow::Drop, "no moment yet"); - } - - /// 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))); - } + fn a_sound_stall_leaves_the_pictures_on_their_moments() { + let clock = sync::Producer::new(VIDEO_DELAY); + let (moved, mut seen) = watch::channel(()); + let video = Hold::new(&clock, moq_mux::container::Kind::Video, &moved); + let audio = Hold::new(&clock, moq_mux::container::Kind::Audio, &moved); + audio.received(stamp(1_000), at(0)); + audio.set_target(TARGET); + video.received(stamp(1_000), at(0)); + assert!(seen.has_changed().unwrap(), "the first arrival anchors the clock"); + seen.mark_unchanged(); + assert_eq!(clock.due(stamp(1_000)), Some(at(150))); + assert_eq!(clock.consume().delay(), TARGET + WRITE_LEAD); - /// 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}"); + // The audio stalls; the pictures go on arriving, and showing, on time. + for index in 1..25 { + video.received(stamp(1_000 + index * 20), at(index * 20)); + assert_eq!(clock.due(stamp(1_000 + index * 20)), Some(at(150 + index * 20))); } - 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); + // Its backlog lands half a second late, moving nothing: what sounded + // through meanwhile is dropped, and the rest written on the clock. + for index in 0..25 { + audio.received(stamp(1_000 + index * 20), at(500)); + } + assert!(!seen.has_changed().unwrap(), "a late track moved the schedule"); + assert_eq!(follow(clock.due(stamp(1_320)), FRAME, at(500)), Follow::Drop); + assert_eq!(follow(clock.due(stamp(1_340)), FRAME, at(500)), Follow::Write(at(490))); + assert_eq!(follow(clock.due(stamp(1_400)), FRAME, at(500)), Follow::Write(at(550))); + assert_eq!(follow(clock.due(stamp(1_420)), FRAME, at(500)), Follow::Wait(at(520))); + + // A stopped audio track lets the pictures play on video's own delay. + drop(audio); + assert!(seen.has_changed().unwrap()); + assert_eq!(clock.consume().delay(), VIDEO_DELAY); } - /// 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. + /// However deep the jitter target, playback stays within the longest delay. #[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 + WRITE_LEAD + 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))); + fn the_delay_never_passes_the_longest() { + let clock = sync::Producer::new(VIDEO_DELAY); + let (moved, _) = watch::channel(()); + let audio = Hold::new(&clock, moq_mux::container::Kind::Audio, &moved); + audio.set_target(MAX_DELAY); + assert_eq!(clock.consume().delay(), MAX_DELAY); } /// The bus mixed a block at a time along with a paused clock, as the diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 1404acf7..0a29ab15 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -19,6 +19,18 @@ use serde::{Deserialize, Serialize}; use crate::Error; +/// The track-name suffix a video rendition of `codec` is published under, or +/// an error for a codec the studio does not publish. +pub fn suffix(codec: &hang::catalog::VideoCodec) -> Result<&'static str, Error> { + match codec { + hang::catalog::VideoCodec::H264(_) => Ok(".avc3"), + hang::catalog::VideoCodec::H265(_) => Ok(".hev1"), + other => Err(Error::Unsupported(format!( + "{other} is not a codec the studio publishes" + ))), + } +} + /// The local catalog's own root section, namespaced as hang asks of /// application sections. pub const SECTION: &str = "live.kixel.studio"; @@ -723,7 +735,11 @@ impl<'a> Video<'a> { config.kind = listing.kind.clone(); let rendition = config.probe().await?; let reserved = listing.catalog.reserve().with_interval(config.effective_join()); - let producer = moq_video::encode::Producer::new(listing.broadcast.clone(), reserved, rendition)?; + let track = listing.broadcast.unique_track( + suffix(&rendition.codec)?, + reserved.track_info(hang::catalog::PRIORITY.video), + )?; + let producer = moq_video::encode::Producer::new(track, reserved, rendition)?; Ok(Self { track: Encoding::new(producer, listing.sessions.clone()), claim: listing.claim, @@ -1668,8 +1684,12 @@ mod tests { let mut config = moq_video::encode::Config::new(size.width, size.height, framerate); config.kind = moq_video::encode::Kind::Named("openh264".into()); let rendition = config.probe().await.unwrap(); - let producer = - moq_video::encode::Producer::new(listing.broadcast.clone(), listing.catalog.reserve(), rendition).unwrap(); + let reserved = listing.catalog.reserve(); + let track = listing + .broadcast + .unique_track(".avc3", reserved.track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let producer = moq_video::encode::Producer::new(track, reserved, rendition).unwrap(); let (name, rendition) = listing.catalog.snapshot().video.renditions.pop_first().unwrap(); // Every group, however far behind the newest, since this reads only once // the track has ended. diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index b101060a..e94a0e5d 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -23,7 +23,7 @@ use hang::catalog::{AudioConfig, Reference, VideoConfig}; use kio::clock::Instant; use moq_audio::bus::{Bus, Mix, Strip}; use moq_mux::catalog::Stream; -use moq_mux::timeline; +use moq_mux::sync; use tokio::sync::{Notify, watch}; use crate::Error; @@ -31,8 +31,8 @@ use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::media::Clip; -use crate::playout::{Follow, Place, Speaker, VIDEO_DELAY, WRITE_LEAD, Writer, follow}; -pub use crate::playout::{MAX_DELAY, heard}; +pub use crate::playout::MAX_DELAY; +use crate::playout::{Follow, Hold, RECHECK, VIDEO_DELAY, WRITE_LEAD, Writer, follow}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -206,10 +206,9 @@ 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, + schedule: sync::Producer, + /// Sent whenever a track's [`Hold`] may have moved the schedule. + moved: watch::Sender<()>, } /// Decoded pictures a broadcast source holds until they are due. The rest wait @@ -529,7 +528,7 @@ pub struct Feed { /// Set once a media source's file turns out to hold no sound. silent: Arc, /// A broadcast source's playout clock, or the one a layer plays on. - clock: Option, + clock: Option, } impl Feed { @@ -704,17 +703,18 @@ impl Feed { } /// 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. + /// network jitter: its audio's playout target and the bus's write lead, + /// 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)) + (self.status() == Status::Live).then(|| clock.delay()) } /// When a broadcast source plays what its broadcast stamps, or a layer /// what its broadcast source's does, so what else that broadcast carries /// can play on the same clock. `None` for any other source. - pub fn clock(&self) -> Option<&timeline::Consumer> { + pub fn clock(&self) -> Option<&sync::Consumer> { self.clock.as_ref() } @@ -763,10 +763,10 @@ pub struct Driver { cues: Option>, runs: Runs, silent: Arc, - clock: Option, + clock: Option, /// A layer's broadcast path, and the clock of the broadcast source it /// plays on. - follows: Option<(String, timeline::Consumer)>, + follows: Option<(String, sync::Consumer)>, } /// Where a source's own track goes, its claim on the local catalog's listing @@ -1189,7 +1189,7 @@ impl Driver { broadcast: moq_net::broadcast::Consumer, path: &str, source: &str, - schedule: &timeline::Consumer, + schedule: &sync::Consumer, ) -> Result<(), Error> { let format = moq_mux::catalog::CatalogFormat::detect(path).unwrap_or_default(); let mut catalogs = moq_mux::catalog::Consumer::::new(&broadcast, format).await?; @@ -1262,14 +1262,14 @@ impl Driver { } /// Show each picture once `schedule`, another track's clock, says it is - /// due and its sound stamped alike has played, the newest such one when - /// several are, as [`play_video`](Self::play_video) does. + /// due, the newest due one when several are, as + /// [`play_video`](Self::play_video) does. async fn follow_video( &self, broadcast: &moq_net::broadcast::Consumer, name: String, config: hang::catalog::VideoConfig, - schedule: &timeline::Consumer, + schedule: &sync::Consumer, ) -> Result<(), Error> { let mut options = moq_video::decode::Options::new(); options.start = moq_video::decode::Start::Latest; @@ -1283,25 +1283,17 @@ impl Driver { let mut ended = false; loop { let now = kio::clock::now(); - // Counted first: `present` holds the clock's lock while it iterates. - let heard_count = queue - .iter() - .take_while(|frame| heard(schedule, frame.timestamp, now)) - .count(); - let ready = queue.iter().take(heard_count).map(|frame| frame.timestamp); - if let Some(index) = schedule.present(ready, now) { + if let Some(index) = schedule.present(queue.iter().map(|frame| frame.timestamp), now) { let frame = queue.drain(..=index).next_back().expect("an index into the queue"); self.show(Picture::Frame(frame)); } if ended && queue.is_empty() { return Ok(()); } - // Its moment, or once due, soon to see whether its sound has played. - let wake = queue.front().and_then(|frame| match schedule.due(frame.timestamp) { - Some(due) if due > now => Some(due), - _ if !heard(schedule, frame.timestamp, now) => Some(now + crate::playout::RECHECK), - _ => None, - }); + let wake = 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)) => queue.push_back(frame), @@ -1314,15 +1306,14 @@ impl Driver { } /// Play a track's sound into the bus when `schedule`, another track's - /// clock, says each frame sounds and its own sound stamped alike has been - /// written, dropping a frame too late for its moment, or before the clock - /// has one. + /// clock, says each frame sounds, dropping a frame that has sounded + /// through, or that arrives before the clock has a moment for it. async fn follow_audio( &self, broadcast: &moq_net::broadcast::Consumer, name: String, config: hang::catalog::AudioConfig, - schedule: &timeline::Consumer, + schedule: &sync::Consumer, ) -> Result<(), Error> { let Some(strip) = self.strip.clone() else { return Ok(()); @@ -1347,13 +1338,13 @@ impl Driver { let now = kio::clock::now(); let mut write = None; while let Some(frame) = queue.front() { - match follow(schedule, frame.timestamp, now) { + match follow(schedule.due(frame.timestamp), length(frame, bus), now) { Follow::Write(sounds) => { let frame = queue.pop_front().expect("peeked"); writer.write(sounds, &frame.data).await?; } Follow::Wait(at) => { - write = Some(at); + write = Some(at.min(now + RECHECK)); break; } Follow::Drop => { @@ -1367,12 +1358,12 @@ impl Driver { tokio::select! { read = consumer.read(), if !ended => match read? { Some(moq_audio::decode::Read::Frame(frame)) => { - // One due past the longest target would hold whatever a + // One due past the longest delay would hold whatever a // peer stamps. let now = kio::clock::now(); - let placed = schedule.due(frame.timestamp).is_some_and(|sounds| { - sounds >= now && sounds <= now + MAX_DELAY + WRITE_LEAD - }); + let due = schedule.due(frame.timestamp); + let placed = follow(due, length(&frame, bus), now) != Follow::Drop + && due.is_some_and(|sounds| sounds <= now + MAX_DELAY); if placed { queue.push_back(frame); } @@ -1393,13 +1384,13 @@ impl Driver { &self, broadcast: moq_net::broadcast::Consumer, path: &str, - clock: timeline::Producer, + clock: sync::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(), + schedule: clock, + moved: watch::Sender::new(()), }; // A track name always means the same content, so only a new track @@ -1432,8 +1423,9 @@ impl Driver { // new one may not hold. played.newest = None; drop(played); - // Video waits for its audio as long as audio's target can - // grow; alone, it waits only its own delay. + // Video plays as far behind as audio's target can grow, + // so its subscription holds that long; alone, 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)), @@ -1474,8 +1466,6 @@ impl Driver { 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; @@ -1484,18 +1474,16 @@ impl Driver { max: Some(max), }; let mut consumer = moq_video::decode::Consumer::new(broadcast, &config, name, options).await?; - let schedule = clock.timeline.consume(); + // A replacement track's stamps need not continue the old one's: its + // hold starts afresh. + let hold = Hold::new(&clock.schedule, moq_mux::container::Kind::Video, &clock.moved); + let mut moved = clock.moved.subscribe(); + let schedule = clock.schedule.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) { + if let Some(index) = schedule.present(queue.iter().map(|frame| frame.timestamp), 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)); @@ -1503,7 +1491,6 @@ impl Driver { 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)) @@ -1511,22 +1498,22 @@ impl Driver { 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()); + hold.received(frame.timestamp, kio::clock::now()); queue.push_back(frame); } - Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => { - clock.timeline.video_restarted(); - } + Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => hold.reset(), None => ended = true, }, () = kio::clock::sleep_until(due.unwrap_or(now)), if due.is_some() => {} - () = clock.changed.notified() => {} + // The sender lives in `clock`, past this track. + _ = moved.changed() => {} } } } - /// 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. + /// Play the broadcast's sound into the bus on its timestamps, each frame + /// when `clock` says it is due, holding the jitter estimate moq-audio's + /// decoder makes on it. async fn play_audio( &self, broadcast: &moq_net::broadcast::Consumer, @@ -1550,60 +1537,80 @@ impl Driver { let mut consumer = moq_audio::decode::Consumer::new(broadcast, &config, name, options).await?; let mut writer = self.writer(strip)?; - 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); + let hold = Hold::new(&clock.schedule, moq_mux::container::Kind::Audio, &clock.moved); + let schedule = &clock.schedule; // Frames wait here for their moment while every arrival is read at - // once: the jitter estimate times each one as `read` hands it over. The - // speaker places none past its target, and a frame that has finished - // sounding by the time another arrives is dropped as too late, so + // once: the jitter estimate times each one as `read` hands it over. A + // frame is due no later than the delay after it arrives, and one that + // has sounded through by the time another arrives is dropped, so // however long a write waits on the bus, this holds only what is - // still to sound within the longest target. - let queue = RefCell::new(VecDeque::<(Instant, Instant, moq_audio::Frame)>::new()); + // still to sound within the longest delay. + // + // Each frame sounds when the clock says, or at the moment frozen for it + // when the timeline broke after it, which the clock no longer holds. + let queue = RefCell::new(VecDeque::<(moq_audio::Frame, Option)>::new()); + let due = + |(frame, frozen): &(moq_audio::Frame, Option)| frozen.or_else(|| schedule.due(frame.timestamp)); let queued = Notify::new(); let received = Cell::new(false); let receive = async { + // The stamps after lost media or an encoder restart need not + // continue the ones before. The hold resets only once the next frame + // arrives, so the clock keeps the moments of the tail still queued + // for as long as nothing replaces it. + let mut broke = false; 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. + broke = true; 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 { - let mut queue = queue.borrow_mut(); - while queue.front().is_some_and(|(_, ends, _)| *ends <= now) { - queue.pop_front(); + let mut queue = queue.borrow_mut(); + if std::mem::take(&mut broke) { + for queued in queue.iter_mut() { + queued.1 = due(queued); } - queue.push_back((sounds, sounds + duration, frame)); - queued.notify_one(); + hold.reset(); + } + hold.received(frame.timestamp, now); + hold.set_target(consumer.delay()); + while queue + .front() + .is_some_and(|front| follow(due(front), length(&front.0, bus), now) == Follow::Drop) + { + queue.pop_front(); } + queue.push_back((frame, None)); + queued.notify_one(); } received.set(true); queued.notify_one(); Ok::<_, Error>(()) }; - // Writes each frame its lead ahead, and tells the clock about - // the frames after a skip once they sound. + // Writes each frame its lead ahead of when the clock says it sounds now, + // which a move of the schedule changes. let write = async { + let mut moved = clock.moved.subscribe(); loop { let now = kio::clock::now(); - speaker.borrow_mut().commit(now); - let write = queue.borrow().front().map(|(sounds, ..)| *sounds - WRITE_LEAD); - if write.is_some_and(|at| at <= now) { - let (sounds, _, frame) = queue.borrow_mut().pop_front().expect("peeked"); - writer.write(sounds, &frame.data).await?; - continue; - } - let wake = write.into_iter().chain(speaker.borrow().deferred()).min(); - match wake { - Some(at) => tokio::select! { + let next = queue + .borrow() + .front() + .map(|front| follow(due(front), length(&front.0, bus), now)); + match next { + Some(Follow::Write(sounds)) => { + let (frame, _) = queue.borrow_mut().pop_front().expect("peeked"); + writer.write(sounds, &frame.data).await?; + } + Some(Follow::Drop) => { + queue.borrow_mut().pop_front(); + } + Some(Follow::Wait(at)) => tokio::select! { () = kio::clock::sleep_until(at) => {} () = queued.notified() => {} + // The sender lives in `clock`, past this track. + _ = moved.changed() => {} }, None if received.get() => return Ok::<_, Error>(()), None => queued.notified().await, @@ -2313,7 +2320,7 @@ fn open( source: &Source, context: &Context, held: bool, - follows: Option<(String, timeline::Consumer)>, + follows: Option<(String, sync::Consumer)>, ) -> (Feed, Driver) { let picture = Mailbox::default(); let delivered = Delivered::default(); @@ -2333,7 +2340,7 @@ fn open( }); let played = Playback::default(); - let clock = matches!(source, Source::Broadcast { .. }).then(|| timeline::Producer::new(VIDEO_DELAY)); + let clock = matches!(source, Source::Broadcast { .. }).then(|| sync::Producer::new(VIDEO_DELAY)); let (plays, replays) = match source { Source::Replay => { let (plays, replays) = watch::channel(Next::default()); @@ -2369,7 +2376,7 @@ fn open( silent: silent.clone(), clock: match &follows { Some((_, schedule)) => Some(schedule.clone()), - None => clock.as_ref().map(timeline::Producer::consume), + None => clock.as_ref().map(sync::Producer::consume), }, }; let driver = Driver { @@ -2470,6 +2477,12 @@ fn image(path: &Path) -> Result<(Picture, bytes::Bytes), Error> { Ok((picture, bytes.into())) } +/// How long a frame of interleaved `f32` at `bus`'s rate and layout sounds. +fn length(frame: &moq_audio::Frame, bus: &moq_audio::bus::Config) -> Duration { + let stride = bus.layout.channels() as usize * size_of::(); + Duration::from_secs_f64((frame.data.len() / stride) as f64 / f64::from(bus.sample_rate)) +} + /// A PNG or JPEG file's picture, within the image crate's default limits. fn decode(bytes: &[u8]) -> Result { let image = image::load_from_memory(bytes)?.into_rgba8(); @@ -2535,7 +2548,10 @@ pub(crate) mod test { config.kind = moq_video::encode::Kind::Software; let rendition = config.probe().await.unwrap(); let reserved = catalog.reserve().with_interval(config.effective_join()); - let mut producer = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition).unwrap(); + let track = broadcast + .unique_track(".avc3", reserved.track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let mut producer = moq_video::encode::Producer::new(track, reserved, rendition).unwrap(); let mut encoder = moq_video::encode::Encoder::new(&config).unwrap(); let gray = vec![0x80; 64 * 64 * 4]; for index in 0..90u64 { @@ -2995,7 +3011,10 @@ mod tests { config.kind = moq_video::encode::Kind::Software; let rendition = config.probe().await.unwrap(); let reserved = timebase.reserve().with_interval(config.effective_join()); - let video = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition).unwrap(); + let track = broadcast + .unique_track(".avc3", reserved.track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let video = moq_video::encode::Producer::new(track, reserved, rendition).unwrap(); (video, moq_video::encode::Encoder::new(&config).unwrap()) } @@ -3273,10 +3292,12 @@ mod tests { .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. + /// A picture shows with the sound stamped alike, the jitter estimate and + /// the bus's write lead after both arrive, rather than on its own delay. + /// Through a sound stall the pictures play on, each on its moment, and the + /// sound's late backlog holds none back. #[tokio::test(start_paused = true)] - async fn a_picture_waits_for_its_sound() { + async fn a_picture_plays_with_its_sound_and_through_its_stall() { local(async { let dir = tempfile::tempdir().unwrap(); let (context, origin) = context(dir.path()); @@ -3286,23 +3307,81 @@ mod tests { start(&sources, drivers).await; let feed = sources.get("guest").unwrap(); let clock = feed.clock.clone().unwrap(); + let stamp = |millis: u64| moq_net::Timestamp::from_millis(millis).unwrap(); + // 20 ms of silence at the bus's 48 kHz stereo. + let silence = + |millis: u64| moq_audio::Frame::new(vec![0u8; 960 * 2 * size_of::()].into(), stamp(millis)); 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(); + live.audio.as_mut().unwrap().write(&silence(1_000)).unwrap(); picture(&mut live.video, &mut live.encoder, 1_000); - let target = until(|| clock.target()).await; - assert!(target + crate::playout::WRITE_LEAD > VIDEO_DELAY, "{target:?}"); - let due = arrived + target + crate::playout::WRITE_LEAD; - until(|| (clock.due(stamp) == Some(due)).then_some(())).await; + let delay = until(|| Some(clock.delay()).filter(|delay| *delay != VIDEO_DELAY)).await; + assert!(delay > crate::playout::WRITE_LEAD, "{delay:?}"); + let due = arrived + delay; + until(|| (clock.due(stamp(1_000)) == Some(due)).then_some(())).await; until_time(due - Duration::from_millis(1)).await; assert!(feed.next_picture().is_none(), "shown before its sound"); until_time(due).await; - assert_eq!(shown(feed).await, Duration::from(stamp)); - assert_eq!(feed.delay(), Some(target)); + assert_eq!(shown(feed).await, Duration::from(stamp(1_000))); + assert_eq!(feed.delay(), Some(delay)); + + // The sound stalls; each picture still shows on its moment. + for index in 1..=10 { + let at = 1_000 + index * 100; + until_time(arrived + Duration::from_millis(index * 100)).await; + picture(&mut live.video, &mut live.encoder, at); + let due = arrived + Duration::from_millis(index * 100) + delay; + until_time(due - Duration::from_millis(1)).await; + assert!(feed.next_picture().is_none(), "{at} shown before its moment"); + until_time(due).await; + assert_eq!( + shown(feed).await, + Duration::from(stamp(at)), + "held by the stalled sound" + ); + } + + // The backlog lands at once, late, and the pictures play on. + let audio = live.audio.as_mut().unwrap(); + for at in (1_020..=2_000).step_by(20) { + audio.write(&silence(at)).unwrap(); + } + let at = 1_000 + kio::clock::now().duration_since(arrived).as_millis() as u64; + picture(&mut live.video, &mut live.encoder, at); + let due = until(|| clock.due(stamp(at))).await; + until_time(due).await; + assert_eq!(shown(feed).await, Duration::from(stamp(at))); + }) + .await; + } + + /// A sound that breaks off keeps the moments of the tail it queued until + /// the media replacing it arrives, rather than dropping it unplayed. + #[tokio::test(start_paused = true)] + async fn a_broken_off_sound_keeps_its_tail() { + 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(); + let stamp = |millis: u64| moq_net::Timestamp::from_millis(millis).unwrap(); + let silence = + |millis: u64| moq_audio::Frame::new(vec![0u8; 960 * 2 * size_of::()].into(), stamp(millis)); + + until_time(kio::clock::now()).await; + let audio = live.audio.as_mut().unwrap(); + for at in (1_000..1_100).step_by(20) { + audio.write(&silence(at)).unwrap(); + } + let due = until(|| clock.due(stamp(1_080))).await; + audio.discontinuity().unwrap(); + until_time(kio::clock::now() + Duration::from_millis(10)).await; + assert_eq!(clock.due(stamp(1_080)), Some(due), "the tail lost its moment"); }) .await; } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 541a8804..cc88cb6a 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -2062,11 +2062,7 @@ impl Renditions { } = probed; let (broadcast, timebase) = (&self.broadcast, &self.timebase); let reserved = timebase.reserve().with_interval(config.effective_join()); - let suffix = match &video.codec { - hang::catalog::VideoCodec::H264(_) => ".avc3", - hang::catalog::VideoCodec::H265(_) => ".hev1", - other => anyhow::bail!("{other} is not a codec the studio publishes"), - }; + let suffix = moq_scene::publish::suffix(&video.codec)?; let rank = role as usize; let priority = match role { Role::Record => hang::catalog::PRIORITY.video, @@ -2074,7 +2070,7 @@ impl Renditions { }; let info = reserved.track_info(priority).with_refresh(ladder::REFRESH); let track = broadcast.unique_track(suffix, info)?; - let producer = moq_video::encode::Producer::with_track(track.clone(), reserved, video)?; + let producer = moq_video::encode::Producer::new(track.clone(), reserved, video)?; let (gate, grant) = match role { // Reads nothing from the connection, so it always encodes at its // ceiling, and only while recorded. @@ -2184,10 +2180,10 @@ impl Probe { let mut config = moq_video::encode::Config::new(size.width, size.height, framerate); config.bitrate = Some(ladder::ceiling(size, framerate)); config.kind = kind.clone(); - let probed = Exporter::new(&gpu.device, &gpu.queue, &mut config) + let probed = Exporter::new(&gpu.device, &gpu.queue, &config) .await .with_context(|| format!("cannot encode a {}x{} canvas", size.width, size.height)); - let (exporter, video) = match probed { + let (exporter, config, video) = match probed { Ok(probed) => probed, Err(err) if role != Role::Half => return Err(err), // The program goes on without it. diff --git a/rs/moq-studio/src/feed.rs b/rs/moq-studio/src/feed.rs index e1d6cba4..87a1799c 100644 --- a/rs/moq-studio/src/feed.rs +++ b/rs/moq-studio/src/feed.rs @@ -122,7 +122,7 @@ impl Feeds { }); let voice = heard.or(voices.first()); match voice.and_then(|name| sources.get(name)?.strip()) { - Some(strip) => pick.set(program.minus(strip)), + Some(strip) => pick.set(program.minus([strip])), None => pick.set(program.clone()), } } diff --git a/rs/moq-studio/src/replay.rs b/rs/moq-studio/src/replay.rs index 4e4dd54b..c97339b0 100644 --- a/rs/moq-studio/src/replay.rs +++ b/rs/moq-studio/src/replay.rs @@ -202,7 +202,10 @@ mod tests { config.kind = moq_video::encode::Kind::Software; let rendition = config.probe().await.unwrap(); let reserved = catalog.reserve().with_interval(config.effective_join()); - let mut producer = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition).unwrap(); + let track = broadcast + .unique_track(".avc3", reserved.track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let mut producer = moq_video::encode::Producer::new(track, reserved, rendition).unwrap(); let mut encoder = moq_video::encode::Encoder::new(&config).unwrap(); let gray = vec![0x80; 64 * 64 * 4]; let at = |index: u64| moq_net::Timestamp::from_micros(index * 1_000_000 / 30).unwrap(); diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index 1790e3d3..0aca8325 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -139,8 +139,8 @@ async fn probe_encoders(gpu: engine::Gpu) -> Encoders { async fn probe_encoder(gpu: &engine::Gpu, picture: &Picture, canvas: Canvas) -> anyhow::Result { use moq_video::render::export::{Exporter, Route}; - let mut config = moq_video::encode::Config::new(canvas.width, canvas.height, canvas.framerate()); - let (mut exporter, rendition) = Exporter::new(&gpu.device, &gpu.queue, &mut config).await?; + let config = moq_video::encode::Config::new(canvas.width, canvas.height, canvas.framerate()); + let (mut exporter, config, rendition) = Exporter::new(&gpu.device, &gpu.queue, &config).await?; let bitrate = rendition.bitrate.context("the encoder reports no bitrate")?; let encoder = match &config.kind { moq_video::encode::Kind::Named(name) => name.clone(), diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index c4228f87..87b5d696 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1967,7 +1967,7 @@ impl Panels { // 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", + "Plays this far behind what arrives, to ride out the network's jitter, with the picture and sound on one clock", ); } None => match feed.and_then(Feed::page).and_then(|page| page.paint()) { diff --git a/rs/moq-studio/src/watch.rs b/rs/moq-studio/src/watch.rs index 1f7c9161..731a5007 100644 --- a/rs/moq-studio/src/watch.rs +++ b/rs/moq-studio/src/watch.rs @@ -16,7 +16,7 @@ use std::time::Duration; use anyhow::Context; use kio::clock::Instant; -use moq_mux::timeline; +use moq_mux::sync; use moq_scene::layout::{self, Layout}; use moq_scene::publish::Ext; use moq_scene::source::{self, MAX_DELAY, Sources, Status}; @@ -190,16 +190,14 @@ impl Read { } /// Take the newest layout due at `now`: each takes effect once the program - /// plays the frame it is stamped with on `schedule`, and its sound has - /// played. `None` while none is due; `Some(None)` for the program alone. - fn take_due(&mut self, schedule: Option<&timeline::Consumer>, now: Instant) -> Option> { + /// plays the frame it is stamped with on `schedule`. `None` while none is + /// due; `Some(None)` for the program alone. + fn take_due(&mut self, schedule: Option<&sync::Consumer>, now: Instant) -> Option> { let mut taken = None; while let Some((_, at)) = self.pending.front() { let due = match at { None => true, - Some(at) => schedule.is_some_and(|schedule| { - schedule.due(*at).is_some_and(|due| due <= now) && source::heard(schedule, *at, now) - }), + Some(at) => schedule.is_some_and(|schedule| schedule.due(*at).is_some_and(|due| due <= now)), }; if !due { break; @@ -415,15 +413,12 @@ impl Watch { self.synced = true; } - // The keyframes play from `start` as far as the program has played, - // following the clock as it re-anchors or stalls with the sound. + // The keyframes play from when the program plays `start`, following + // the clock as it moves. if let (Some(layout), Some(schedule)) = (&self.layout, &schedule) { - let went_on = match schedule.playhead(now) { - Some(head) => now.checked_sub(head.saturating_sub(layout.start)), - None => moq_net::Timestamp::try_from(layout.start) - .ok() - .and_then(|start| schedule.due(start)), - }; + let went_on = moq_net::Timestamp::try_from(layout.start) + .ok() + .and_then(|start| schedule.due(start)); if let Some(went_on) = went_on { self.compositor.set_program_start(went_on); } @@ -597,17 +592,10 @@ mod tests { } /// Once the layout track is unlisted, the program plays alone at once, - /// even while a stalled program holds back a layout still waiting. + /// even while a program not playing yet holds back a layout still waiting. #[tokio::test(start_paused = true)] async fn unlisting_supersedes_a_waiting_layout() { - let clock = timeline::Producer::new(Duration::from_millis(100)); - let audio = timeline::Audio { - end: Duration::from_secs(1), - buffered: Duration::ZERO, - target: Duration::from_millis(100), - }; - clock.audio(audio, Instant::now()); - let schedule = clock.consume(); + let schedule = sync::Producer::new(Duration::from_millis(100)).consume(); let (broadcast, catalog, read, reader) = broadcast(); let mut publisher = layout::Publisher::new(broadcast.clone(), catalog.clone()); @@ -623,7 +611,7 @@ mod tests { }); taken }; - assert_eq!(take(), None, "the program stalled before it"); + assert_eq!(take(), None, "the program has no moment for it"); publisher.finish().unwrap(); alone(&read).await; diff --git a/sh/upgrade-moq.sh b/sh/upgrade-moq.sh index bf3fc122..07b438a2 100755 --- a/sh/upgrade-moq.sh +++ b/sh/upgrade-moq.sh @@ -11,8 +11,8 @@ git -C "$root/moq" checkout --quiet --detach "$rev" # Resolve without --locked: Cargo adds and prunes only what the new moq needs. cargo metadata --manifest-path "$root/Cargo.toml" --format-version 1 >/dev/null -# Print the first registry crate both locks hold at semver-compatible but -# different versions, as `@ `. +# Print each registry crate both locks hold at semver-compatible but different +# versions, as `@ `. drift() { awk ' function compat(v, p) { @@ -25,7 +25,7 @@ drift() { /^source = "registry/ { key = name " " compat(version) if (file == 1) moq[key] = version - else if (key in moq && moq[key] != version) { print name "@" version, moq[key]; exit } + else if (key in moq && moq[key] != version) print name "@" version, moq[key] } ' "$root/moq/Cargo.lock" "$root/Cargo.lock" } @@ -33,15 +33,22 @@ drift() { # Cargo resolves a crate the new moq adds at its newest release, not the one # moq's lock tested, and that release can pull a second copy of a crate moq # holds once: two copies of rind's assembly fail the fat-LTO release link. -# So move every shared crate to moq's version; one the studio needs newer fails here. -last="" -while next="$(drift)" && [ -n "$next" ]; do - if [ "$next" = "$last" ]; then - echo "upgrade-moq: cargo did not move ${next% *} to moq's ${next#* }" >&2 +# So move every shared crate to moq's version. A crate a newer dependent holds +# back (rind under a newer rind-rcgen) moves on a later pass, once that +# dependent has moved; a pass that moves nothing means the studio needs one +# newer, and fails here. +while pending="$(drift)" && [ -n "$pending" ]; do + moved="" + while read -r crate version; do + if cargo update --manifest-path "$root/Cargo.toml" --quiet -p "$crate" --precise "$version" 2>/dev/null; then + moved=1 + fi + done <<<"$pending" + if [ -z "$moved" ]; then + echo "upgrade-moq: cargo could not move these to moq's versions:" >&2 + echo "$pending" >&2 exit 1 fi - cargo update --manifest-path "$root/Cargo.toml" --quiet -p "${next% *}" --precise "${next#* }" - last="$next" done # The join page's workspace includes moq's JS packages, whose dependencies move too. (cd "$root" && bun install)