diff --git a/quest/m0/README.md b/quest/m0/README.md index 357a5c9..df2f874 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -100,7 +100,6 @@ repository. - [Web studio](/quest/m0/web.md) - the full studio in wasm on WebGPU and WebCodecs, served from this repository - [Virtual camera](/quest/m0/vcam/README.md) - the program or a published scene exposed as a system camera on Linux, Windows, and macOS - [Browser sources on the GPU under Mesa](/quest/m0/cef-headless.md) - shared-texture pages paint on Intel and AMD under Wayland, and a page that cannot paint fails loudly -- [A media file's sound plays with its pictures](/quest/m0/media-av-sync.md) - media sources write a bus latency ahead, as broadcast sources do - [The UI keeps its frame rate while the program renders](/quest/m0/ui-frame-rate.md) - the editor draws without waiting on the program's render - [The Apple signing identity is available](/quest/m0/signing.md) - the Apple credentials packaging and the macOS camera need; Windows ships unsigned - [Packaging](/quest/m0/packaging.md) - a signed macOS app, a Windows installer, and a Flatpak for Linux diff --git a/quest/m0/media-av-sync.md b/quest/m0/media-av-sync.md deleted file mode 100644 index 7dda851..0000000 --- a/quest/m0/media-av-sync.md +++ /dev/null @@ -1,20 +0,0 @@ -# [S] A media file's sound plays with its pictures - -## Goal - -A media file source's sound plays in step with its own pictures in every -output (program, published scenes, recordings, and Listen), instead of about -45 ms behind them. - -## Plan - -From [moq-dev/studio#81](https://github.com/moq-dev/studio/pull/81), picked -2026-11-07 (maintainer away under /afk, recommended option), m0 by the rule -that fixes join m0. - -Pictures show when they are due, but a media source writes its sound to the -bus when it is due, and the bus mixes it a bus latency (50 ms) later. -Broadcast sources already write a bus latency ahead (`playout::Writer`); -media sources should do the same. Test that a media file's picture change -and its sound onset leave an output within one bus block of each other, -measured on the media clock rather than with sleeps. diff --git a/rs/moq-scene/src/compositor.rs b/rs/moq-scene/src/compositor.rs index 88f3340..730a86b 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -1064,10 +1064,11 @@ fn stinger(feed: Option<&Feed>, cut: Duration, played: &mut Option, now: In if run.ended.is_some_and(|ended| ended <= now) { return Ok(None); } - Ok(Some( - run.begun - .is_some_and(|begun| now.saturating_duration_since(begun) >= cut), - )) + // A play begins a moment after it is asked, once its first picture and + // sound can play together, so even a cut at its start waits for it. + Ok(Some(run.begun.is_some_and(|begun| { + now.checked_duration_since(begun).is_some_and(|into| into >= cut) + }))) } /// The crop that trims what `crop` leaves of a `size` picture evenly from two @@ -1799,7 +1800,7 @@ mod tests { use super::*; use crate::collection::{Blend, MaskMode, Rgba, Scene, Size, Video}; - use crate::source::test::{context, listen, local, start}; + use crate::source::test::{context, listen, local, start, until, until_time}; /// A fade over a second. const FADE: Transition = Transition::Fade { @@ -2857,6 +2858,42 @@ mod tests { .await; } + /// A stinger cut at its clip's very start switches the scenes no sooner + /// than the clip's first picture shows, which is a moment after it plays. + #[tokio::test(start_paused = true)] + async fn a_stinger_cuts_no_sooner_than_its_clip_begins() { + local(async { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("sting.mkv"), crate::media::tests::CLIP).unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = crate::source::Sources::new(context); + let mut collection = Collection::new(); + let media = Source::Media { + path: "sting.mkv".into(), + looping: false, + restart: false, + }; + collection.sources.insert("sting".into(), media); + collection.transition = Transition::Stinger { + source: "sting".into(), + cut: Duration::ZERO, + }; + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let feed = sources.get("sting"); + + let mut played = None; + let cut = |played: &mut Option| stinger(feed, Duration::ZERO, played, kio::clock::now()).unwrap(); + assert_eq!(cut(&mut played), Some(false), "it plays the clip"); + let begun = until(|| feed.unwrap().run().begun).await; + assert!(begun > kio::clock::now(), "the clip begins ahead"); + assert_eq!(cut(&mut played), Some(false), "not before the clip begins"); + until_time(begun).await; + assert_eq!(cut(&mut played), Some(true), "as the clip begins"); + }) + .await; + } + /// A stinger's clip covers a canvas of another shape unstretched, trimmed /// evenly from the sides that overflow, after its source's own crop. #[test] diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 7ec5903..12afc98 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -9,7 +9,9 @@ //! `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. +//! and a [`Writer`] writes it the bus's latency ahead of then. A media +//! source's sound goes through a [`Writer`] too, so it sounds with the +//! pictures it shows. use std::time::Duration; diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index c45615d..dc86570 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -495,7 +495,7 @@ pub(crate) struct Video<'a> { config: moq_video::encode::Config, clock: moq_mux::Clock, /// Where the capture's private timeline starts on the studio clock, in - /// microseconds, taken at the first frame. + /// microseconds, taken at the first frame unless [anchored](Self::anchor). epoch: Option, } @@ -529,6 +529,14 @@ impl<'a> Video<'a> { }) } + /// Start the private timeline at `origin` on the studio clock, rather than + /// where the first frame encodes, for a source that encodes its pictures + /// ahead of when they show. + pub(crate) fn anchor(&mut self, origin: kio::clock::Instant) -> Result<(), Error> { + self.epoch = Some(self.clock.capture(origin)?.as_micros()); + Ok(()) + } + /// Restamp `frame` onto the program's clock, and encode and publish it /// while the source is used: shown on program, or its track read. The /// encoder opens on the first frame used and closes while none is, marking @@ -675,10 +683,11 @@ impl<'a> Audio<'a> { }) } - /// Encode and publish `samples`, whose first was captured at `captured`, - /// while the source is used: heard on program, or its track read. `gap` - /// says sound was lost just before them. Leaving use marks a gap, and the - /// sound after one is stamped afresh, so the track's timeline holds it. + /// Encode and publish `samples`, whose first was captured or sounds at + /// `captured`, while the source is used: heard on program, or its track + /// read. `gap` says sound was lost just before them. Leaving use marks a + /// gap, and the sound after one is stamped afresh, so the track's timeline + /// holds it. pub(crate) fn write(&mut self, samples: &[f32], captured: kio::clock::Instant, gap: bool) -> Result<(), Error> { let producer = self.producer.as_mut().expect("open until it ends"); let used = self.on_program.borrow().heard.contains(&self.claim.name) || producer.demand().wanted().is_some(); @@ -692,9 +701,13 @@ impl<'a> Audio<'a> { producer.reanchor(); } self.encoding = true; - // A device may stamp a buffer a moment ahead of the clock, which the - // clock refuses as not yet captured. - let timestamp = self.clock.capture(captured.min(kio::clock::now()))?; + // A media source writes its sound ahead of when it sounds, and a device + // may stamp a buffer a moment ahead of the clock, which maps only what + // has passed: the rest is counted on from now. + let now = kio::clock::now(); + let ahead = captured.saturating_duration_since(now).as_micros(); + let timestamp = self.clock.capture(captured.min(now))?.as_micros() + ahead; + let timestamp = moq_net::Timestamp::from_micros(timestamp as u64).map_err(moq_audio::Error::from)?; let data = bytes::Bytes::copy_from_slice(bytemuck::cast_slice(samples)); producer.write(&moq_audio::Frame::new(data, timestamp))?; Ok(()) @@ -797,17 +810,20 @@ mod tests { } } - /// Walk a media source playing a 10 fps clip through `count` pictures, the - /// first due at `first`. The clock moves to each picture's moment and is - /// held while it decodes, encodes, and shows, so no wait counts on work in - /// threads the runtime cannot see, such as a decoder's or an encoder's, - /// finishing within any span of the clock. - async fn pictures(feed: &crate::source::Feed, first: kio::clock::Instant, count: u32) { + /// Walk a media source playing a 10 fps clip through `count` of its play's + /// pictures from the one at index `from`, returning when the play began. + /// The clock is held until the play begins, then moves to each picture's + /// moment and is held while it decodes, encodes, and shows, so no wait + /// counts on work in threads the runtime cannot see, such as a decoder's or + /// an encoder's, finishing within any span of the clock. + async fn pictures(feed: &crate::source::Feed, from: u32, count: u32) -> kio::clock::Instant { + let begun = until(|| feed.run().begun).await; let shown = || matches!(feed.next_picture(), Some(crate::source::Picture::Frame(_))).then_some(()); - for index in 0..count { - until_time(first + std::time::Duration::from_millis(100) * index).await; + for index in from..from + count { + until_time(begun + std::time::Duration::from_millis(100) * index).await; until(shown).await; } + begun } /// Yield until `f` holds of the local catalog's source listings. @@ -984,7 +1000,7 @@ mod tests { tracks.push(consumer.track(name).unwrap().subscribe(subscription).await.unwrap()); } let written = |id: &str| publish.stats().get(id).map_or(0, |rendition| rendition.frames); - pictures(sources.get("bed").unwrap(), kio::clock::now(), 4).await; + pictures(sources.get("bed").unwrap(), 0, 4).await; until(|| names.iter().all(|name| written(name) > 0).then_some(())).await; sources.stop(); @@ -1572,8 +1588,8 @@ mod tests { let drivers = sources.sync(&collection); start(&sources, drivers).await; - // The first picture is due at once, so the clock stays put while - // the encoder opens. + // The pictures' track opens before the play begins, so the clock + // stays put while the encoder opens. let listed = until(|| { let listed = catalog.snapshot().ext.studio.sources.remove("clip"); listed.filter(|listed| !listed.video.renditions.is_empty()) @@ -1594,13 +1610,11 @@ mod tests { let mut track = moq_mux::container::Consumer::new(track, container); // Half way, five pictures in, it starts over, and plays to its end. - let first = kio::clock::now(); - pictures(feed, first, 5).await; + let first = pictures(feed, 0, 5).await; until_time(first + std::time::Duration::from_millis(450)).await; let cut = publish.catalog.clock().now(); feed.play().unwrap(); - let restarted = kio::clock::now(); - pictures(feed, restarted, 10).await; + let restarted = pictures(feed, 0, 10).await; until_time(restarted + std::time::Duration::from_millis(1_100)).await; until(|| (feed.status() == Status::Ended).then_some(())).await; @@ -1620,6 +1634,13 @@ mod tests { .await; assert!(stamps[0] < cut, "the first play's keyframe: {stamps:?}"); assert!(stamps.windows(2).all(|pair| pair[0] < pair[1]), "{stamps:?}"); + // Stamped with when it showed, though encoded ahead of it. + let shown = publish.catalog.clock().capture(first).unwrap(); + assert!( + stamps[0].as_micros().abs_diff(shown.as_micros()) <= 1, + "the first picture as it showed: {:?}, not {shown:?}", + stamps[0] + ); // Ended, it shows nothing new, yet gives its session back once off // program and unread. @@ -1665,8 +1686,7 @@ mod tests { let written = |id: &str| publish.stats().get(id).map_or(0, |rendition| rendition.frames); let feed = sources.get("bed").unwrap(); - let first = kio::clock::now(); - pictures(feed, first, 4).await; + pictures(feed, 0, 4).await; until(|| (written(&audio) > 0).then_some(())).await; assert_eq!(written(&video), 0, "heard, not shown"); assert_eq!(sessions.open(), 0, "no encoder for the pictures"); @@ -1677,7 +1697,7 @@ mod tests { shown: ["bed".to_string()].into(), heard: ["bed".to_string()].into(), }); - pictures(feed, first + std::time::Duration::from_millis(400), 6).await; + pictures(feed, 4, 6).await; until(|| (written(&video) > 0).then_some(())).await; }) .await; @@ -1723,7 +1743,7 @@ mod tests { let container = moq_mux::catalog::hang::Container::try_from(config).unwrap(); let mut track = moq_mux::container::Consumer::new(track, container); // Two and a half loops of a second, each with half a second of sound. - pictures(sources.get("bed").unwrap(), kio::clock::now(), 25).await; + let begun = pictures(sources.get("bed").unwrap(), 0, 25).await; let now = kio::Waiter::noop(); let mut stamps = Vec::new(); let silences = until(|| { @@ -1738,6 +1758,9 @@ mod tests { }) .await; assert_eq!(silences, 2, "each loop's silence is kept: {stamps:?}"); + // Stamped with when it sounded, though written ahead of it. + let sounded = std::time::Duration::from(publish.catalog.clock().capture(begun).unwrap()); + assert_eq!(stamps[0], sounded, "the first sound as it sounded"); }) .await; } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index e4c6466..cb1dcf6 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -31,7 +31,7 @@ use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::media::Clip; -use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; +use crate::playout::{BUS_LATENCY, MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -131,11 +131,13 @@ pub struct Run { /// Which play it is: 0 for the one the source opens with, and one more /// for each [`Feed::play`] since. pub play: u64, - /// When its first picture or sound was due, or `None` until it is decoded. + /// When its first picture shows and sound sounds, or `None` until it is + /// decoded. Set a little ahead: the bus sounds a write 50 ms later, so a + /// play begins that long after it is decoded. pub begun: Option, - /// When its last picture or sound finished, a looping one's first pass - /// did, or [`Feed::stop`] stopped it, or `None` until then. A play cut off - /// by the next never ends. + /// When its last picture or sound finishes, a looping one's first pass + /// does, or [`Feed::stop`] stopped it, or `None` until known. A looping + /// one's is known 50 ms ahead. A play cut off by the next never ends. pub ended: Option, } @@ -1227,10 +1229,7 @@ impl Driver { loop { let now = kio::clock::now(); speaker.borrow_mut().commit(now); - let write = queue - .borrow() - .front() - .map(|(sounds, _)| *sounds - crate::playout::BUS_LATENCY); + let write = queue.borrow().front().map(|(sounds, _)| *sounds - 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)?; @@ -1318,7 +1317,7 @@ impl Driver { moq_video::decode::Read::Frame(frame) if frame.timestamp >= take.end => break, moq_video::decode::Read::Frame(mut frame) => { kio::clock::sleep_until(at(frame.timestamp)).await; - self.encode(&mut own, &mut frame, config).await?; + self.encode(&mut own, &mut frame, config, kio::clock::now()).await?; self.show(Picture::Frame(frame)); } // A stretch the broadcast was offline, paused, or lost @@ -1383,7 +1382,8 @@ impl Driver { /// picture, silent, until the next. /// /// With `listing`, the pictures are also encoded onto the source's own - /// track as they show, while it is used, on one timeline across plays. + /// track, stamped when they show, while it is used, on one timeline across + /// plays. async fn media( &self, path: &Path, @@ -1404,7 +1404,7 @@ impl Driver { self.publish(crate::publish::Audio::new(listing, input)) }); Some(Sound { - channel: None, + writer: None, track, ended: None, }) @@ -1412,64 +1412,105 @@ impl Driver { _ => None, }; let mut own = Own::new(listing, kio::clock::now()); + let shows = Shows::default(); // The last channel on a strip to close drops what it has not played // yet, so a loop keeps its channel, sounding on, while a new cue drops - // it, cutting off what the play before still had queued; the next play - // takes a new one. Its pictures start a group there too, a cut to the - // clip's start. + // it, cutting off what the play before still had queued, pictures + // included; the next play takes a new one. Its pictures start a group + // there too, a cut to the clip's start. let cut = |sound: &mut Option>, own: &mut Own<'_, '_>| { if let Some(sound) = sound { - sound.channel = None; + sound.writer = None; sound.ended = None; } + shows.queue.borrow_mut().clear(); if let Some(video) = &mut own.track && let Err(err) = video.cut() { self.fail_encoding(&mut own.track, err); } }; - let mut cue = *cues.borrow_and_update(); - // Where a loop's next play starts: where the last one ended, so loops - // keep exact time whatever opening the decoders takes. - let mut next = None; - loop { - // Whether to rest until the next cue, rather than take it at once. - let rest = if cue.stopped { - true - } else { - if let (Some(sound), Some(strip)) = (&mut sound, &self.strip) - && sound.channel.is_none() - { - sound.channel = Some(self.channel(strip.clone())?); - } - let played = tokio::select! { - result = self.play_clip(&clip, sound.as_mut(), &mut own, next, cue.play) => Some(result?), - changed = cues.changed() => { - // The feed is gone, which stops this driver anyway. - if changed.is_err() { - return std::future::pending().await; + let plays = async { + let mut cue = *cues.borrow_and_update(); + // Where a loop's next play starts: where the last one ended, so + // loops keep exact time whatever opening the decoders takes. + let mut next = None; + loop { + // Whether to rest until the next cue, rather than take it at once. + let rest = if cue.stopped { + true + } else { + if let (Some(sound), Some(strip)) = (&mut sound, &self.strip) + && sound.writer.is_none() + { + let bus = self.context.bus.config(); + let channel = self.channel(strip.clone())?; + sound.writer = Some(Writer::new(channel, bus.sample_rate, bus.layout.channels())); + } + let play = async { + let end = self + .play_clip(&clip, sound.as_mut(), &mut own, &shows, next, cue.play) + .await?; + // A loop's next play hands its first over now, a bus + // latency ahead, as this one did. + if !looping { + kio::clock::sleep_until(end).await; + } + Ok::<_, Error>(end) + }; + let played = tokio::select! { + result = play => Some(result?), + changed = cues.changed() => { + // The feed is gone, which stops this driver anyway. + if changed.is_err() { + return std::future::pending().await; + } + None + } + }; + if let Some(end) = played { + self.ran(cue.play, |run| { + run.ended.get_or_insert(end); + }); + next = Some(end); + if looping { + continue; } - None } + played.is_some() }; - if let Some(end) = played { - self.ran(cue.play, |run| { - run.ended.get_or_insert(end); - }); - next = Some(end); - if looping { - continue; + if rest { + set(&self.status, Status::Ended); + self.ended(&mut cues, &mut own).await; + } + cut(&mut sound, &mut own); + cue = *cues.borrow_and_update(); + next = None; + } + }; + // Shows each picture handed over at its moment, across plays, so a + // loop's next play can hand its first over before this one's last + // has shown. + let present = async { + loop { + let now = kio::clock::now(); + let due = shows.queue.borrow().front().map(|(due, ..)| *due); + match due { + Some(due) if due <= now => { + let (_, play, frame) = shows.queue.borrow_mut().pop_front().expect("peeked"); + self.show_play(play, Picture::Frame(frame)); } + Some(due) => tokio::select! { + () = kio::clock::sleep_until(due) => {} + () = shows.handed.notified() => {} + }, + None => shows.handed.notified().await, } - played.is_some() - }; - if rest { - set(&self.status, Status::Ended); - self.ended(&mut cues, &mut own).await; } - cut(&mut sound, &mut own); - cue = *cues.borrow_and_update(); - next = None; + }; + tokio::select! { + result = plays => result, + () = present => unreachable!("pictures show for as long as the source plays"), } } @@ -1521,16 +1562,24 @@ impl Driver { } } - /// Play `clip` once, each picture and sound when its moment comes, the - /// first at `begin`, or at once, as part of the feed's `play`. Returns - /// where it ends, once the last has played for as long as it lasts: a - /// sound for its samples, and a picture for as long as the one before it. - /// A clip that decodes to nothing fails. + /// Play `clip` once, each picture and sound at its moment, the first at + /// `begin`, or as soon as the bus can sound it, as part of the feed's + /// `play`. + /// + /// The bus sounds a write [`BUS_LATENCY`] later, so each sound is written + /// that far ahead of its moment, and each picture handed to `shows` then + /// too, to show at the same moment. + /// + /// Returns where it ends, once all is handed over and the bus latency + /// before the end has come, so a loop's next play hands its own first over + /// in time. A sound lasts for its samples, and the last picture as long as + /// the one before it. A clip that decodes to nothing fails. async fn play_clip( &self, clip: &Clip, mut sound: Option<&mut Sound<'_>>, own: &mut Own<'_, '_>, + shows: &Shows, begin: Option, play: u64, ) -> Result { @@ -1575,13 +1624,22 @@ impl Driver { if let (Some(frame), Some((_, config))) = (&picture, &video) { self.open(own, frame, config).await; } - let begin = begin.unwrap_or_else(kio::clock::now); + let begin = begin.unwrap_or_else(|| kio::clock::now() + BUS_LATENCY); self.ran(play, |run| { run.begun.get_or_insert(begin); }); let at = |timestamp: moq_net::Timestamp| begin + Duration::from(timestamp).saturating_sub(start); let end = std::cell::Cell::new(begin); let last = |until: Instant| end.set(end.get().max(until)); + // What is due at `due` is handed over a bus latency ahead. One due at + // once is handed over without a timer, which the timer wheel would + // round up to its next millisecond, a moment too late. + let ahead = async |due: Instant| { + let at = due - BUS_LATENCY; + if at > kio::clock::now() { + kio::clock::sleep_until(at).await; + } + }; let pictures = async { let Some((consumer, config)) = &mut video else { @@ -1590,11 +1648,12 @@ impl Driver { let mut previous = None; while let Some(mut frame) = picture.take() { let due = at(frame.timestamp); - kio::clock::sleep_until(due).await; + ahead(due).await; last(due + previous.map_or(Duration::ZERO, |previous| due - previous)); previous = Some(due); - self.encode(own, &mut frame, config).await?; - self.show_play(play, Picture::Frame(frame)); + self.encode(own, &mut frame, config, due).await?; + shows.queue.borrow_mut().push_back((due, play, frame)); + shows.handed.notify_one(); picture = next_picture(consumer).await?; } Ok::<_, Error>(()) @@ -1612,16 +1671,16 @@ impl Driver { let mut first = true; while let Some(frame) = samples.take() { let due = at(frame.timestamp); - kio::clock::sleep_until(due).await; + ahead(due).await; let length = (frame.data.len() / bytes) as u64 * 1_000_000_000 / u64::from(bus.sample_rate); last(due + Duration::from_nanos(length)); let gap = std::mem::take(&mut first) && sound.ended != Some(due); sound.ended = Some(due + Duration::from_nanos(length)); sound - .channel + .writer .as_mut() .expect("a channel while playing") - .write(&frame.data)?; + .write(due, &frame.data, kio::clock::now())?; if let Some(track) = &mut sound.track { let samples: Vec = bytemuck::pod_collect_to_vec(&frame.data); if let Err(err) = track.write(&samples, due, gap) { @@ -1635,8 +1694,9 @@ impl Driver { Ok::<_, Error>(()) }; tokio::try_join!(pictures, sounds)?; - kio::clock::sleep_until(end.get()).await; - Ok(end.get()) + let end = end.get(); + ahead(end).await; + Ok(end) } /// Open the source's own track for pictures like `frame`, which says their @@ -1644,7 +1704,12 @@ impl Driver { async fn open(&self, own: &mut Own<'_, '_>, frame: &moq_video::Frame, config: &VideoConfig) { if let Some(listing) = own.listing.filter(|_| !own.opened) { let video = crate::publish::Video::new(listing, frame.size(), framerate(config), frame.surface.color()); - own.track = self.publish(video.await); + // Pictures are stamped with when they show on `own`'s timeline, + // which may be ahead of when they encode. + let video = video + .await + .and_then(|mut video| video.anchor(own.origin).map(|()| video)); + own.track = self.publish(video); // Only once it has, so a play cut off while it opens leaves the // next picture to open it. own.opened = true; @@ -1652,18 +1717,20 @@ impl Driver { } /// Encode a played picture onto the source's own track, opened at the - /// first picture if not before. Stamped when it shows, on the timeline - /// `own` keeps. + /// first picture if not before. Stamped with when it `shows`, on the + /// timeline `own` keeps. async fn encode( &self, own: &mut Own<'_, '_>, frame: &mut moq_video::Frame, config: &VideoConfig, + shows: Instant, ) -> Result<(), Error> { self.open(own, frame, config).await; if let Some(video) = &mut own.track { - frame.timestamp = moq_net::Timestamp::from_micros(own.origin.elapsed().as_micros() as u64) - .map_err(moq_video::Error::from)?; + let stamp = shows.saturating_duration_since(own.origin); + frame.timestamp = + moq_net::Timestamp::from_micros(stamp.as_micros() as u64).map_err(moq_video::Error::from)?; if let Err(err) = video.encode(frame) { self.fail_encoding(&mut own.track, err); } @@ -1705,13 +1772,21 @@ impl Driver { /// A media source's sound: the bus channel it plays into, taken anew on a /// restart, and its own track. struct Sound<'a> { - channel: Option, + writer: Option, track: Option>, /// When the last sound written ends, or `None` after a break, such as a /// restart, so the next sound's track starts its timeline afresh. ended: Option, } +/// A media source's pictures, handed over a bus latency ahead of their moments +/// with their plays, and woken whenever one is. +#[derive(Default)] +struct Shows { + queue: RefCell>, + handed: Notify, +} + /// A played source's own track, opened at its first picture, and the timeline /// its pictures are stamped on: from `origin`, when they show. struct Own<'a, 'b> { @@ -2914,16 +2989,22 @@ mod media_tests { } /// The next picture `feed` shows, passing over the clear a restart puts up, - /// and when it showed. + /// and when it showed. The clock moves only to a play's beginning, which + /// comes a bus latency after its first picture decodes. async fn shown(feed: &Feed) -> (moq_net::Timestamp, Instant) { - let timestamp = until(|| match feed.next_picture() { - Some(Picture::Frame(frame)) => Some(frame.timestamp), - Some(Picture::Rgba { pixels, .. }) if pixels == [0; 4] => None, - Some(_) => panic!("not a decoded picture"), - None => None, - }) - .await; - (timestamp, kio::clock::now()) + loop { + let next = until(|| match feed.next_picture() { + Some(Picture::Frame(frame)) => Some(Ok(frame.timestamp)), + Some(Picture::Rgba { pixels, .. }) if pixels == [0; 4] => None, + Some(_) => panic!("not a decoded picture"), + None => feed.run().begun.filter(|begun| *begun > kio::clock::now()).map(Err), + }) + .await; + match next { + Ok(timestamp) => return (timestamp, kio::clock::now()), + Err(begun) => until_time(begun).await, + } + } } /// Wait until `name` reads as `status`, without moving the clock. @@ -3184,6 +3265,161 @@ mod media_tests { .await; } + /// One bus block: how often the bus mixes, and so how finely a mix places + /// a sound in time. + const BLOCK: Duration = Duration::from_millis(10); + + /// The bus mixed along with a paused clock a block at a time, as an audio + /// clock would, keeping the first channel of what one mix heard. + struct Ear { + mixer: moq_audio::bus::Driver, + reader: moq_audio::bus::Reader, + /// When the clock read the bus's start, so a frame `n` in sounds `n` + /// samples after it. + origin: Instant, + at: Instant, + heard: Vec, + } + + impl Ear { + /// Start mixing now, a block of which the bus mixes at once, so a + /// channel keeps what is written to it from here on. + fn new(mixer: moq_audio::bus::Driver, mix: &Mix) -> Self { + let now = kio::clock::now(); + let mut ear = Self { + mixer, + reader: mix.reader(), + origin: now - BLOCK, + at: now, + heard: Vec::new(), + }; + ear.mix(); + ear + } + + /// Move the clock a block on, and mix that block. + async fn step(&mut self) { + self.at += BLOCK; + until_time(self.at).await; + self.mix(); + } + + fn mix(&mut self) { + self.mixer.advance(Self::frames(BLOCK)); + while let std::task::Poll::Ready(Some(block)) = self.reader.poll_read(&kio::Waiter::noop()) { + let channels = block.data.len() / block.frames(); + assert_eq!(block.position as usize, self.heard.len(), "a block skipped"); + self.heard.extend(block.data.iter().step_by(channels)); + } + } + + /// The frames the bus mixes over `duration`. + fn frames(duration: Duration) -> usize { + let rate = moq_audio::bus::Config::default().sample_rate; + (duration.as_nanos() * u128::from(rate) / 1_000_000_000) as usize + } + + /// When the mix first heard anything. + fn onset(&self) -> Option { + let index = self.heard.iter().position(|sample| sample.abs() > 0.01)?; + let rate = moq_audio::bus::Config::default().sample_rate; + Some(self.origin + Duration::from_secs_f64(index as f64 / f64::from(rate))) + } + + /// The loudest the mix heard over `duration` from `from`. + fn peak(&self, from: Instant, duration: Duration) -> f32 { + let start = Self::frames(from - self.origin); + let heard = &self.heard[start..start + Self::frames(duration)]; + heard.iter().fold(0.0, |peak, sample| peak.max(sample.abs())) + } + } + + /// A file's first picture shows as its first sound is heard, within a bus + /// block, rather than a bus latency ahead of it. + #[tokio::test(start_paused = true)] + async fn a_media_files_sound_plays_with_its_pictures() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (mut context, _origin) = context(dir.path()); + let (bus, mixer) = Bus::new(Default::default()).unwrap(); + context.bus = bus; + let mut sources = Sources::new(context); + let collection = media(dir.path(), &[("clip.mkv", CLIP, false, false)]); + let drivers = sources.sync(&collection); + let program = sources.mix(crate::AudioScene::DEFAULT).unwrap().clone(); + let mut ear = Ear::new(mixer, &program); + start(&sources, drivers).await; + let feed = sources.get("clip.mkv").unwrap(); + + while !matches!(feed.next_picture(), Some(Picture::Frame(_))) { + ear.step().await; + } + // Seen at the step after it showed, which may be the block after, + // when the step woke before the picture's timer at the same moment. + let shown = ear.at; + let begun = feed.run().begun.unwrap(); + assert!( + begun <= shown && shown - begun <= BLOCK, + "the first picture shows when it begins: {:?} after", + shown - begun + ); + while ear.at < begun + Duration::from_millis(100) { + ear.step().await; + } + let onset = ear.onset().expect("the clip sounds"); + let apart = onset + .saturating_duration_since(begun) + .max(begun.saturating_duration_since(onset)); + assert!( + apart < BLOCK, + "its first sound is heard with its first picture: {onset:?} and {begun:?}" + ); + }) + .await; + } + + /// A looping file's sound runs on from its end into its start again, as + /// loud as the first time and as its pass begins, each pass heard from + /// when it begins. + #[tokio::test(start_paused = true)] + async fn a_looping_files_sound_runs_on() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (mut context, _origin) = context(dir.path()); + let (bus, mixer) = Bus::new(Default::default()).unwrap(); + context.bus = bus; + let mut sources = Sources::new(context); + let collection = media(dir.path(), &[("tone.webm", TONE, true, false)]); + let drivers = sources.sync(&collection); + let program = sources.mix(crate::AudioScene::DEFAULT).unwrap().clone(); + let mut ear = Ear::new(mixer, &program); + start(&sources, drivers).await; + let feed = sources.get("tone.webm").unwrap(); + + let ended = loop { + if let Some(ended) = feed.run().ended { + break ended; + } + ear.step().await; + }; + let begun = feed.run().begun.unwrap(); + while ear.at < ended + Duration::from_millis(100) { + ear.step().await; + } + // Past a block that either may be heard early by, and the start of + // the tone the codec fades in. + let into = Duration::from_millis(15); + let first = ear.peak(begun + into, Duration::from_millis(30)); + let again = ear.peak(ended + into, Duration::from_millis(30)); + assert!(first > 0.05, "the tone sounds: {first}"); + assert!( + again > first / 2.0, + "the loop starts again as loud: {again} after {first}" + ); + }) + .await; + } + /// A file of sound alone shows nothing, and goes silent at its end. One of /// pictures alone has no sound for a mix to play. #[tokio::test(start_paused = true)] diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 886ffb7..ed13e8d 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -4006,19 +4006,19 @@ mod tests { // Until then, the program's viewer heard nothing while the portrait's // heard the bed. assert!(bed < played, "the bed is heard before the cut"); - let (heard, before) = program.hears(quiet, played, LOUD).await; + // The viewer hears the tone as it plays: a media source writes each + // sound the bus's latency ahead. The bus holds a first write that long + // from the block it last mixed, so the tone may be heard up to a block + // early, or two while the audio clock's tick runs late. + let early = 2 * AUDIO_PERIOD; + let (heard, before) = program.hears(quiet, played - early, LOUD).await; assert!( before < QUIET, "the program's audio scene holds the tone alone: {before}" ); - // The viewer hears the tone once the bus mixes it: a media source - // writes each sound when it is due, and the bus holds every source - // 50 ms (moq-audio's `bus::LATENCY`) before mixing it, give or take a - // block. - let mixed = Duration::from_millis(50) + AUDIO_PERIOD; assert!( - heard >= played && heard - played < mixed, - "the tone is heard as the bus mixes it: played {played:?}, heard {heard:?}" + heard + early > played && heard < played + AUDIO_PERIOD, + "the tone is heard as it plays: played {played:?}, heard {heard:?}" ); let (also, _) = tall.hears(quiet, played, LOUD).await; assert!(