From 6cc79d82485562e2aa299af8eed11b51e6007a09 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:01:44 -0800 Subject: [PATCH 1/7] chore(quest): claim quest/m0/playout-positioned.md Quest-Claim: b36ab86eec56c182 From ccb2a0dd21fd114ebb13932a37f37baec636493c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:24:32 -0800 Subject: [PATCH 2/7] feat(scene)!: playout writes land on positioned bus channels Co-Authored-By: Claude Opus 5.5 --- quest/m0/playout-lag.md | 25 +++ quest/m0/playout-positioned.md | 72 -------- rs/moq-scene/src/compositor.rs | 49 +++-- rs/moq-scene/src/playout.rs | 316 +++++++++++++++++++-------------- rs/moq-scene/src/source.rs | 125 ++++++------- rs/moq-studio/src/engine.rs | 22 ++- rs/moq-studio/src/feed.rs | 1 + rs/moq-studio/tests/browser.rs | 1 + 8 files changed, 321 insertions(+), 290 deletions(-) create mode 100644 quest/m0/playout-lag.md delete mode 100644 quest/m0/playout-positioned.md diff --git a/quest/m0/playout-lag.md b/quest/m0/playout-lag.md new file mode 100644 index 00000000..2346fed4 --- /dev/null +++ b/quest/m0/playout-lag.md @@ -0,0 +1,25 @@ +# [S] A lagging audio clock holds playout writes instead of failing the source + +## Goal + +When the program's audio clock thread falls behind the program clock, a +broadcast, media, or replay source's sound waits for the bus to catch up, or +loses only what the bus can no longer place, rather than failing the source. + +## Plan + +`playout::Writer` (`rs/moq-scene/src/playout.rs`) maps a moment to a bus +position from the program clock's start, and bounds each write by where the +clock says the bus is, since moq-audio's `Positioned` does not say where the +bus has mixed to. A write ends at most `WRITE_LEAD` plus one frame (about +70 ms) past the clock, inside `bus::REACH` (200 ms). If the audio thread lags +the clock by more than the rest (about 130 ms), `Positioned::write` refuses the +write as past its reach, with the same `Error::Unsupported` it gives a +non-finite sample, and the source fails: a media source stays failed until +restarted. + +Recommended: moq-audio's `Positioned` reports the bus's next position (or +refuses a write past its reach with an error of its own), landed in moq-dev/moq +with a moq-side test and pinned here with `just upgrade-moq`. The writer then +cuts against the clock as now, but waits on the bus's actual position for +what lies past its reach. diff --git a/quest/m0/playout-positioned.md b/quest/m0/playout-positioned.md deleted file mode 100644 index 87f302e9..00000000 --- a/quest/m0/playout-positioned.md +++ /dev/null @@ -1,72 +0,0 @@ -# [M] Playout writes land on positioned bus channels - -## Goal - -A broadcast, media, or replay source's sound plays at exactly the bus -position its pictures are timed for. `rs/moq-scene`'s playout writes it -through moq-audio's `Bus::positioned`, and its copy of the live ring's pad -rule is deleted. How far ahead of its moment a sound is handed to the bus -stays a constant the studio owns: today's `BUS_LATENCY`, renamed for what it -is (such as `WRITE_LEAD`) and no longer a copy of moq-audio's private -`bus::LATENCY`. - -Non-goals: the `Speaker`'s jitter targeting stays as it is, and live inputs -(microphones, capture, guests) stay on `Bus::live`. - -## Plan - -Planned 2026-11-08 (maintainer, moq's /quest-complete), a follow-up of -[moq-dev/moq#8273](https://github.com/moq-dev/moq/pull/8273), which added -`Bus::positioned` for this consumer. - -Today `playout::Writer` writes into a live `bus::Channel` and keeps track of -when what it wrote runs out, padding silence ahead of a frame so it sounds a -copied 50 ms `BUS_LATENCY` later, the way moq-audio's ring would. Each source -that plays timed audio (`source.rs`'s broadcast, media, and replay paths) -goes through it. - -Decided with the maintainer: - -- ✅ **Studio maps an instant to a bus position** (rejected: an upstream - moq-audio bus clock). `Program::run_audio` (`rs/moq-studio/src/engine.rs`) already advances the bus - sample-exact on the program clock from `start`, so the position a frame - sounds at is `(sounds - start) * rate`. The writer gets the program clock's - `start` and the bus rate through the sources' `Context`. moq-audio's bus - counts frames and has no wall clock, so a clock there would guess timing - only the app knows. -- ✅ **A late frame is trimmed, not dropped** (rejected: drop the whole - frame). `Positioned::write` fails a write any part of which is already - mixed, overlaps earlier audio, or lands past `bus::REACH`. The writer cuts - the frames whose position has passed and writes the rest at its position, - as today's writer does with a late part. - -Decided 2026-11-09 (maintainer, /quest-audit): - -- ✅ **Keep a studio-owned write lead** (rejected: deleting `BUS_LATENCY` - outright). The Goal deleted it while the `Speaker`'s targeting, a non-goal, - is built on it: `Speaker::place` sounds a frame `target + BUS_LATENCY` - after it arrives and re-buffers one that would follow less than half of it - away, and `source.rs` hands each sound over that far ahead of its moment - (`ahead`, the broadcast write loop, and where a media or replay play - begins). A positioned write needs that lead too: the bus mixes a block at a - time, and a write with any part already mixed fails. So the constant stays, - within `bus::REACH`. What goes is the `Writer`'s pad rule and the claim - that the constant is moq-audio's `bus::LATENCY`, which - `a_write_sounds_when_placed` pins today. - -Recommendations, not settled: - -- Trim against the later of the already-mixed position and the previous - write's end, so a frame overlapping queued audio writes its playable tail - instead of failing whole. -- Keep consecutive writes contiguous: a frame that starts within the - existing one-millisecond `CONTIGUOUS` stamp rounding of where the last one - ended is written at that end, since a one-frame overlap from rounding fails - the write. The snap only absorbs stamp rounding: confirm the source's stamp - resolution and shrink the tolerance to it (e.g. a sample or two) if that is - tighter, so a genuine sub-millisecond gap stays silent. -- A frame the writer holds earlier than `REACH` ahead waits until it is in - reach rather than failing. -- Test on the scene's driver-stepped bus, as `a_write_sounds_when_placed` - does: frames written ahead, late, and across a stall sound at the expected - bus positions, sample-exact. diff --git a/rs/moq-scene/src/compositor.rs b/rs/moq-scene/src/compositor.rs index 755c5ad3..01a921e1 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -2195,7 +2195,7 @@ mod tests { use super::*; use crate::collection::{Blend, MaskMode, Rgba, Scene, Size, Video}; - use crate::source::test::{context, listen, local, start, until, until_time}; + use crate::source::test::{context, local, start, until, until_time}; /// A fade over a second. const FADE: Transition = Transition::Fade { @@ -2256,15 +2256,29 @@ mod tests { collection } - /// A compositor over running sources, read back to check pixels. + /// Mix the bus sample-exact on the clock from `start`, as the program's + /// audio clock does. + async fn mix(mut mixer: moq_audio::bus::Driver, start: Instant) { + let rate = u128::from(moq_audio::bus::Config::default().sample_rate); + let mut clock = kio::clock::interval_at(start, Duration::from_millis(10)); + clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); + let mut mixed = 0; + loop { + let tick = clock.tick().await; + let due = tick.saturating_duration_since(start).as_nanos() * rate / 1_000_000_000; + mixer.advance((due - mixed) as usize); + mixed = due; + } + } + + /// A compositor over running sources, read back to check pixels, whose + /// sound mixes along with the clock. pub(super) struct Studio { pub(super) collection: Collection, pub(super) sources: Sources, pub(super) compositor: Compositor, pub(super) device: wgpu::Device, queue: wgpu::Queue, - /// Mixes the sources' sound, only as a test advances it. - mixer: moq_audio::bus::Driver, _origin: moq_net::origin::Producer, _dir: tempfile::TempDir, } @@ -2281,6 +2295,7 @@ mod tests { context.output = moq_video::render::output(&device); let (bus, mixer) = moq_audio::bus::Bus::new(Default::default()).unwrap(); context.bus = bus; + tokio::task::spawn_local(mix(mixer, context.bus_start)); let mut sources = Sources::new(context); let drivers = sources.sync(&collection); start(&sources, drivers).await; @@ -2290,12 +2305,22 @@ mod tests { queue, collection, sources, - mixer, _origin: origin, _dir: dir, } } + /// The loudest `mix` gets from now until `until`. + async fn hear(mix: &moq_audio::bus::Mix, until: Instant) -> f32 { + mix.level(); + let mut loudest = 0.0f32; + while Instant::now() < until { + kio::clock::sleep(Duration::from_millis(10)).await; + loudest = loudest.max(mix.level().peak()); + } + loudest + } + pub(super) fn render(&mut self, now: Instant) { self.compositor.render(&self.collection, &self.sources, now).unwrap(); } @@ -3219,8 +3244,6 @@ mod tests { local(async { let mut studio = stinging(crate::media::tests::CLIP).await; let program = studio.sources.mix(crate::AudioScene::DEFAULT).unwrap().clone(); - let mut at = Instant::now(); - studio.mixer.advance(480); let transition = studio.collection.transition.clone(); studio .compositor @@ -3228,8 +3251,7 @@ mod tests { .unwrap(); loop { studio.render(Instant::now()); - let until = at + Duration::from_millis(10); - if listen(&mut studio.mixer, &program, &mut at, until).await > 0.0 { + if Studio::hear(&program, Instant::now() + Duration::from_millis(10)).await > 0.0 { break; } } @@ -3239,8 +3261,7 @@ mod tests { while feed.status() != Status::Ended { kio::clock::sleep(Duration::from_millis(5)).await; } - let until = at + Duration::from_millis(500); - let heard = listen(&mut studio.mixer, &program, &mut at, until).await; + let heard = Studio::hear(&program, Instant::now() + Duration::from_millis(500)).await; assert_eq!(heard, 0.0, "the clip's sound stops with the cut"); // Stopped from the Sources panel, the stinger ends with it. @@ -3274,16 +3295,14 @@ mod tests { .unwrap(); while studio.compositor.transitioning() { studio.render(Instant::now()); - let until = at + Duration::from_millis(10); - listen(&mut studio.mixer, &program, &mut at, until).await; + Studio::hear(&program, Instant::now() + Duration::from_millis(10)).await; } let feed = studio.sources.get("sting").unwrap(); assert!(feed.run().ended.is_some(), "after its first pass"); while feed.status() != Status::Ended { kio::clock::sleep(Duration::from_millis(5)).await; } - let until = at + Duration::from_millis(500); - let heard = listen(&mut studio.mixer, &program, &mut at, until).await; + let heard = Studio::hear(&program, Instant::now() + Duration::from_millis(500)).await; assert_eq!(heard, 0.0, "the loop stops with the stinger"); }) .await; diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 86fcc0be..7d6d7aa0 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -6,11 +6,11 @@ //! each picture once the clock says it is due and its sound has played. //! Without audio, video paces itself [`VIDEO_DELAY`] behind its arrivals. //! -//! `moq play` holds its jitter buffer in the speaker. The bus holds only -//! [`BUS_LATENCY`] of a source, so a [`Speaker`] holds the rest: it places each -//! frame the playout target moq-audio's decoder estimates after it arrives, -//! and a [`Writer`] writes it the bus's latency ahead of then. A media or -//! replay source's sound goes through a [`Writer`] too, so it sounds with the +//! `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. use std::time::Duration; @@ -30,12 +30,22 @@ pub(crate) const VIDEO_DELAY: Duration = Duration::from_millis(100); /// estimate at it, since a skipped group is never measured. pub(crate) const MAX_DELAY: Duration = Duration::from_secs(2); -/// What a bus channel holds between a write and the mix that plays it, while -/// it is kept fed. +/// How far ahead of its moment a sound is handed to the bus. /// -/// moq-audio's `bus::LATENCY`, which is private; `a_write_sounds_when_placed` -/// pins the two together. -pub(crate) const BUS_LATENCY: Duration = Duration::from_millis(50); +/// The bus mixes a block at a time, on a clock thread that may wake late, and +/// fails a write any part of which it has already mixed, so a sound is handed +/// over this long before its moment to be mixed whole. +pub(crate) const WRITE_LEAD: Duration = Duration::from_millis(50); + +/// How far ahead of the clock a [`Writer`] lets a write end: the bus's reach, +/// less the lead, so the bus may lag the clock by up to the lead before a +/// write ends past what it holds. +const AHEAD: Duration = bus::REACH.saturating_sub(WRITE_LEAD); + +const _: () = assert!( + WRITE_LEAD.as_nanos() < AHEAD.as_nanos(), + "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. @@ -104,18 +114,18 @@ impl<'a> Speaker<'a> { let target = target.min(MAX_DELAY); let end = start.saturating_add(duration); let begins = match self.tail { - None => now + target + BUS_LATENCY, + None => now + target + WRITE_LEAD, Some((sounds, media)) => { let follows = sounds + start.saturating_sub(media); - // A write half the bus's latency late still finds sound queued - // ahead of it, however coarsely the bus advances. - if follows < now + BUS_LATENCY / 2 { + // 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 + BUS_LATENCY + now + target + WRITE_LEAD } else if start.saturating_add(CONTIGUOUS) < media - || follows.saturating_duration_since(now) > target + BUS_LATENCY + SLACK + || 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. @@ -182,67 +192,89 @@ pub(crate) fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp .is_some_and(|head| Duration::from(timestamp) <= head + CONTIGUOUS) } -/// Writes placed frames into a bus channel, so each sounds when it was placed. +/// Writes placed frames into a positioned bus channel, each at the bus +/// position its moment maps to, so it sounds exactly then. /// -/// A channel plays a write [`BUS_LATENCY`] later only while it holds that much: -/// one that ran dry plays the next write at once. So the writer keeps track of -/// when what it wrote runs out, and puts silence ahead of a frame for however -/// much of the latency has drained. +/// The bus's driver advances sample-exact on the program clock from `start`, +/// so a moment `t` is bus position `(t - start) * rate`, and the bus has never +/// mixed past the position of the clock's reading. pub(crate) struct Writer { - channel: bus::Live, + channel: bus::Positioned, /// Bytes per sample frame of the channel's `f32` input. stride: usize, rate: u32, - /// When what was written finishes sounding, or `None` before the first - /// write, which the channel's own priming holds back the full latency. - runs_out: Option, - /// Silence and the frame after it, written as one, so the channel pads - /// behind the frame rather than between them. - pending: Vec, + /// When the bus mixes its position 0. + start: Instant, + /// The bus position the last write ended at, which the next follows. + end: i128, } impl Writer { - /// Write into a new `channel`, taking interleaved `f32` at `rate` in `channels`. - pub(crate) fn new(channel: bus::Live, rate: u32, channels: u32) -> Self { + /// Write into `channel` on a bus mixing position 0 at `start`, taking + /// interleaved `f32` at the bus `rate` in `channels`. + pub(crate) fn new(channel: bus::Positioned, start: Instant, rate: u32, channels: u32) -> Self { Self { channel, stride: channels as usize * size_of::(), rate, - runs_out: None, - pending: Vec::new(), + start, + end: 0, } } - /// Write `data` at `now` to sound at `sounds`. Call it no sooner than - /// [`BUS_LATENCY`] ahead. Written late, the part whose moment has passed is - /// dropped, so the rest still sounds on schedule, and so does what follows. - pub(crate) fn write(&mut self, sounds: Instant, data: &[u8], now: Instant) -> Result<(), moq_audio::Error> { - let free = self.runs_out.unwrap_or(now + BUS_LATENCY).max(now); - let late = self.frames(free.saturating_duration_since(sounds)) * self.stride; - let Some(data) = data.get(late..).filter(|data| !data.is_empty()) else { - return Ok(()); + /// Write `data` to sound at `sounds`, handed over [`WRITE_LEAD`] ahead. + /// + /// A frame that starts within [`CONTIGUOUS`] of where the last one ended + /// continues it. The part already mixed, or overlapping the last frame, is + /// cut, so the rest still sounds on schedule. The part further ahead than + /// the bus holds waits until it is in reach. + pub(crate) async fn write(&mut self, sounds: Instant, data: &[u8]) -> Result<(), moq_audio::Error> { + let mut position = self.position(sounds); + if position.abs_diff(self.end) <= self.frames(CONTIGUOUS) { + position = self.end; + } + let end = position + (data.len() / self.stride) as i128; + loop { + let mixed = self.mixed(kio::clock::now()); + let from = position.max(self.end).max(mixed); + let until = end.min(mixed + self.frames(AHEAD) as i128); + if from < until { + let bytes = |at: i128| (at - position) as usize * self.stride; + self.channel.write(from as u64, &data[bytes(from)..bytes(until)])?; + self.end = until; + } + if until >= end { + return Ok(()); + } + kio::clock::sleep_until(self.instant(until) - WRITE_LEAD).await; + } + } + + /// The bus position sounding at `at`, to the nearest frame. + fn position(&self, at: Instant) -> i128 { + let nanos = match at.checked_duration_since(self.start) { + Some(after) => after.as_nanos() as i128, + None => -(self.start.duration_since(at).as_nanos() as i128), }; - let gap = self.frames(sounds.saturating_duration_since(free).min(BUS_LATENCY)); - self.pending.clear(); - self.pending.resize(gap * self.stride, 0); - self.pending.extend_from_slice(data); - self.channel.write(&self.pending)?; - let runs_out = free + self.duration(self.pending.len() / self.stride); - // A channel left holding a quarter of its latency or less pads itself - // back up to it behind the write, as moq-audio's ring does. - self.runs_out = Some(match runs_out.saturating_duration_since(now) <= BUS_LATENCY / 4 { - true => now + BUS_LATENCY, - false => runs_out, - }); - Ok(()) + (nanos * i128::from(self.rate) + 500_000_000).div_euclid(1_000_000_000) + } + + /// The first bus position not mixed by `now` at the latest: rounded up, + /// since the bus mixes a frame only once the clock has passed it. + fn mixed(&self, now: Instant) -> i128 { + let nanos = now.saturating_duration_since(self.start).as_nanos(); + (nanos * u128::from(self.rate)).div_ceil(1_000_000_000) as i128 } - fn frames(&self, duration: Duration) -> usize { - (duration.as_secs_f64() * f64::from(self.rate)).round() as usize + /// When bus position `position` sounds. + fn instant(&self, position: i128) -> Instant { + let nanos = position.max(0) as u128 * 1_000_000_000 / u128::from(self.rate); + self.start + Duration::from_nanos(nanos as u64) } - fn duration(&self, frames: usize) -> Duration { - Duration::from_secs_f64(frames as f64 / f64::from(self.rate)) + /// The whole frames `duration` spans. + fn frames(&self, duration: Duration) -> u128 { + duration.as_nanos() * u128::from(self.rate) / 1_000_000_000 } } @@ -383,7 +415,7 @@ mod tests { for index in 0..1_000 { for start in [ms(1_000), ms(1_000 + index)] { if let Place::Play(begins) = speaker.place(start, FRAME, TARGET, at(0)) { - assert!(begins <= at(0) + TARGET + BUS_LATENCY + SLACK, "{index}: {begins:?}"); + assert!(begins <= at(0) + TARGET + WRITE_LEAD + SLACK, "{index}: {begins:?}"); } } } @@ -391,95 +423,109 @@ mod tests { assert_eq!(clock.due(stamp(500)), Some(at(1_150))); } - /// A frame sounds when placed: the first, one that follows it, and ones - /// written after the channel ran dry, on time or late, which the bus would - /// otherwise play at once or pad behind, apart from their pictures. One - /// written after its moment loses what has passed, or all of it. - #[test] - fn a_write_sounds_when_placed() { - let (bus, mut driver) = moq_audio::bus::Bus::new(Default::default()).unwrap(); - let config = bus.config().clone(); - let rate = u64::from(config.sample_rate); - let strip = moq_audio::bus::Strip::default(); - let mut input = moq_audio::encode::Input::new(config.sample_rate, config.layout); - input.format = moq_audio::Format::F32; - let channel = bus.live(input, strip.clone()).unwrap(); - let mut writer = Writer::new(channel, config.sample_rate, config.layout.channels()); - let mix = bus.mix(); - mix.route(&strip); - let mut reader = mix.reader(); - - // The bus clock, as an instant, and a frame of `value` on it. - let block = rate / 100; - let clock = |frames: u64| Instant::ZERO + Duration::from_micros(frames * 1_000_000 / rate); - let frame = |value: f32| bytemuck::cast_slice::(&vec![value; 2 * 960]).to_vec(); - let mut heard = Vec::new(); - let mut position = 0; - let mut advance = |until: u64, heard: &mut Vec<(u64, f32)>| { - while position < until { - driver.advance(block as usize); - position += block; - let std::task::Poll::Ready(Some(read)) = reader.poll_read(&kio::Waiter::noop()) else { + /// The bus mixed a block at a time along with a paused clock, as the + /// program's audio clock mixes it, noting each change in what one mix heard. + struct Ear { + driver: bus::Driver, + reader: bus::Reader, + start: Instant, + rate: u64, + /// Each bus position the first channel changed at, and what to. + heard: Vec<(u64, f32)>, + } + + impl Ear { + const BLOCK: u64 = 480; + + /// Mix every block the clock passes up to `until`, as it passes each. + async fn until(&mut self, until: Instant) { + loop { + let next = self.driver.position() + Self::BLOCK; + let due = self.start + Duration::from_nanos(next * 1_000_000_000 / self.rate); + if due > until { + return; + } + kio::clock::sleep_until(due).await; + self.driver.advance(Self::BLOCK as usize); + let std::task::Poll::Ready(Some(read)) = self.reader.poll_read(&kio::Waiter::noop()) else { panic!("an advance publishes a block"); }; - let mut last = heard.last().map_or(0.0, |(_, value)| *value); + let mut last = self.heard.last().map_or(0.0, |(_, value)| *value); for (index, sample) in read.data.chunks(2).enumerate() { if sample[0] != last { last = sample[0]; - heard.push((read.position + index as u64, last)); + self.heard.push((read.position + index as u64, last)); } } } + } + } + + /// A frame sounds at the bus position its moment maps to, sample-exact: + /// handed over ahead, continuing the last across a stamp's rounding, + /// overlapping it, late, wholly late, while the bus lags the clock, and + /// longer than the bus holds. Whatever cannot sound when it says is cut. + #[tokio::test(start_paused = true)] + async fn a_write_sounds_at_its_position() { + let (bus, driver) = bus::Bus::new(Default::default()).unwrap(); + let config = bus.config().clone(); + let strip = bus::Strip::default(); + let mut input = moq_audio::encode::Input::new(config.sample_rate, config.layout); + input.format = moq_audio::Format::F32; + let channel = bus.positioned(input, strip.clone()).unwrap(); + let mix = bus.mix(); + mix.route(&strip); + let start = kio::clock::now(); + let mut writer = Writer::new(channel, start, config.sample_rate, config.layout.channels()); + let mut ear = Ear { + driver, + reader: mix.reader(), + start, + rate: u64::from(config.sample_rate), + heard: Vec::new(), }; + let at = |millis: u64| start + ms(millis); + // `millis` of a frame of `value`, at 48 frames a millisecond. + let frame = |value: f32, millis: usize| bytemuck::cast_slice::(&vec![value; 2 * 48 * millis]).to_vec(); + + ear.until(at(10)).await; + writer.write(at(60), &frame(1.0, 20)).await.unwrap(); + ear.until(at(30)).await; + // Stamped half a millisecond after the last ended. + writer.write(at(80) + Duration::from_micros(500), &frame(2.0, 20)).await.unwrap(); + // Overlapping the last by 5 ms. + writer.write(at(95), &frame(3.0, 20)).await.unwrap(); + + ear.until(at(1_000)).await; + writer.write(at(990), &frame(4.0, 20)).await.unwrap(); + ear.until(at(2_000)).await; + writer.write(at(1_980), &frame(5.0, 20)).await.unwrap(); + + // The bus stalls 30 ms behind the clock, then catches up. + ear.until(at(2_970)).await; + kio::clock::sleep_until(at(3_000)).await; + writer.write(at(3_050), &frame(6.0, 20)).await.unwrap(); + ear.until(at(3_200)).await; + + // Longer than the bus holds, handed over as the bus mixes. + ear.until(at(4_000)).await; + let long = frame(7.0, 300); + let (written, ()) = tokio::join!(writer.write(at(4_050), &long), ear.until(at(4_500))); + written.unwrap(); - advance(block, &mut heard); - let sounds = clock(block) + BUS_LATENCY; - writer.write(sounds, &frame(1.0), clock(block)).unwrap(); - // The next one, on time. - advance(block + 960, &mut heard); - writer.write(sounds + FRAME, &frame(2.0), clock(block + 960)).unwrap(); - // Long after both ran out. - advance(rate, &mut heard); - writer - .write(clock(rate) + BUS_LATENCY, &frame(3.0), clock(rate)) - .unwrap(); - advance(2 * rate, &mut heard); - // Run dry and written late, with less than the latency to go. - advance(2 * rate + 1_920, &mut heard); - writer - .write(clock(2 * rate) + BUS_LATENCY, &frame(4.0), clock(2 * rate + 1_920)) - .unwrap(); - advance(3 * rate, &mut heard); - // Run dry and written after its moment began, then after it ended. - writer - .write( - clock(3 * rate) - Duration::from_millis(10), - &frame(5.0), - clock(3 * rate), - ) - .unwrap(); - advance(4 * rate, &mut heard); - writer - .write(clock(4 * rate) - FRAME, &frame(6.0), clock(4 * rate)) - .unwrap(); - advance(5 * rate, &mut heard); - - let latency = rate * BUS_LATENCY.as_millis() as u64 / 1_000; - let first = block + latency; - let after = rate + latency; - let late = 2 * rate + latency; assert_eq!( - heard, + ear.heard, [ - (first, 1.0), - (first + 960, 2.0), - (first + 1_920, 0.0), - (after, 3.0), - (after + 960, 0.0), - (late, 4.0), - (late + 960, 0.0), - (3 * rate, 5.0), - (3 * rate + 480, 0.0) + (48 * 60, 1.0), + (48 * 80, 2.0), + (48 * 100, 3.0), + (48 * 115, 0.0), + (48 * 1_000, 4.0), + (48 * 1_010, 0.0), + (48 * 3_050, 6.0), + (48 * 3_070, 0.0), + (48 * 4_050, 7.0), + (48 * 4_350, 0.0), ] ); } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 574b725a..f2ac6ff4 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::{BUS_LATENCY, MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; +use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY, WRITE_LEAD, Writer}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -241,6 +241,10 @@ pub struct Context { /// Where microphones, system audio, and broadcasts feed their sound, and /// where each of the collection's audio scenes is mixed. pub bus: Bus, + /// When `bus` mixes its position 0: its driver advances sample-exact on + /// the clock from here, so a sound due at `t` plays at bus position + /// `(t - bus_start) * rate`, and never mixes past where the clock is. + pub bus_start: Instant, /// Where broadcast sources are announced: the studio's relay connection. pub origin: moq_net::origin::Consumer, /// What a relative image, LUT, or overlay asset path is relative to: the @@ -1222,8 +1226,7 @@ impl Driver { options.output.layout = Some(bus.layout); let mut consumer = moq_audio::decode::Consumer::new(broadcast, &config, name, options).await?; - let channel = self.channel(strip)?; - let mut writer = Writer::new(channel, bus.sample_rate, bus.layout.channels()); + 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); @@ -1254,16 +1257,16 @@ impl Driver { queued.notify_one(); Ok::<_, Error>(()) }; - // Writes each frame its bus latency ahead, and tells the clock about + // Writes each frame its lead ahead, and tells the clock about // the frames after a skip once they sound. let write = async { loop { let now = kio::clock::now(); speaker.borrow_mut().commit(now); - let write = queue.borrow().front().map(|(sounds, _)| *sounds - BUS_LATENCY); + 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, now)?; + writer.write(sounds, &frame.data).await?; continue; } let wake = write.into_iter().chain(speaker.borrow().deferred()).min(); @@ -1315,8 +1318,7 @@ impl Driver { /// first of each is decoded and the bus can sound it. Ends once the whole /// span has played. /// - /// The bus sounds a write [`BUS_LATENCY`] later, so each sound is written - /// that far ahead of its moment. + /// Each sound is handed to the bus [`WRITE_LEAD`] ahead of its moment. /// /// With `listing`, the pictures are also encoded onto the source's own /// track as they show, while it is used. @@ -1336,9 +1338,7 @@ impl Driver { (Some((name, config)), Some(strip)) if play.speed == 1.0 => { let consumer = moq_audio::decode::Consumer::new(&broadcast, config, name.clone(), self.sounds()).await?; - let bus = self.context.bus.config(); - let writer = Writer::new(self.channel(strip.clone())?, bus.sample_rate, bus.layout.channels()); - Some((consumer, writer)) + Some((consumer, self.writer(strip.clone())?)) } _ => None, }; @@ -1358,8 +1358,7 @@ impl Driver { } } } - // A sound across the start is written whole: the writer drops what - // is due before the replay begins. + // A sound across the start plays from where the replay begins. let mut samples = None; if let Some((consumer, _)) = &mut audio { let bus = self.context.bus.config(); @@ -1379,7 +1378,7 @@ impl Driver { if let (Some(moq_video::decode::Read::Frame(frame)), Some((_, config))) = (&picture, &video) { self.open(&mut own, frame, config).await; } - let begin = kio::clock::now() + BUS_LATENCY; + let begin = kio::clock::now() + WRITE_LEAD; self.ran(play.play, |run| { run.begun.get_or_insert(begin); }); @@ -1427,12 +1426,18 @@ impl Driver { let Some((consumer, writer)) = &mut audio else { return Ok(()); }; + let bus = self.context.bus.config(); + let stride = bus.layout.channels() as usize * size_of::(); while let Some(frame) = samples.take() && frame.timestamp < take.end { let due = at(frame.timestamp); ahead(due).await; - writer.write(due, &frame.data, kio::clock::now())?; + let before = begin.saturating_duration_since(due).as_nanos() * u128::from(bus.sample_rate); + let cut = (before / 1_000_000_000) as usize * stride; + writer + .write(due.max(begin), frame.data.get(cut..).unwrap_or_default()) + .await?; samples = next_samples(consumer).await?; } // The bus drops what a closed channel has not played, so it stays @@ -1525,16 +1530,14 @@ impl Driver { 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())); + sound.writer = Some(self.writer(strip.clone())?); } 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. + // A loop's next play hands its first over now, its + // lead ahead, as this one did. if !looping { kio::clock::sleep_until(end).await; } @@ -1648,12 +1651,11 @@ impl Driver { /// `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. + /// Each sound is handed to the bus [`WRITE_LEAD`] ahead of its moment, and + /// each picture 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 + /// Returns where it ends, once all is handed over and the lead 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( @@ -1709,7 +1711,7 @@ impl Driver { // A cut's play begins after the last picture already published, even // one cut off before it showed, so the track's time never repeats. let begin = begin.unwrap_or_else(|| { - let soonest = kio::clock::now() + BUS_LATENCY; + let soonest = kio::clock::now() + WRITE_LEAD; own.stamped .map_or(soonest, |stamped| soonest.max(stamped + Duration::from_millis(1))) }); @@ -1759,7 +1761,8 @@ impl Driver { .writer .as_mut() .expect("a channel while playing") - .write(due, &frame.data, kio::clock::now())?; + .write(due, &frame.data) + .await?; 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) { @@ -1839,13 +1842,14 @@ impl Driver { options } - /// A channel into the bus for decoded sound, as [`sounds`](Self::sounds) + /// A writer into the bus for decoded sound, as [`sounds`](Self::sounds) /// and broadcasts decode it, through `strip`. - fn channel(&self, strip: Strip) -> Result { + fn writer(&self, strip: Strip) -> Result { let bus = self.context.bus.config(); let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); input.format = moq_audio::Format::F32; - Ok(self.context.bus.live(input, strip)?) + let channel = self.context.bus.positioned(input, strip)?; + Ok(Writer::new(channel, self.context.bus_start, bus.sample_rate, bus.layout.channels())) } } @@ -1859,7 +1863,7 @@ struct Sound<'a> { ended: Option, } -/// A media source's pictures, handed over a bus latency ahead of their moments +/// A media source's pictures, handed over [`WRITE_LEAD`] ahead of their moments /// with their plays, and woken whenever one is. #[derive(Default)] struct Shows { @@ -1891,12 +1895,12 @@ impl<'a, 'b> Own<'a, 'b> { } } -/// Wait to hand over what is due at `due`: a bus latency ahead, so a sound -/// written then is heard when due. One due at once is handed over without a +/// Wait to hand over what is due at `due`: [`WRITE_LEAD`] ahead, so a sound +/// written then is mixed whole when due. 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. async fn ahead(due: Instant) { - let at = due - BUS_LATENCY; + let at = due - WRITE_LEAD; if at > kio::clock::now() { kio::clock::sleep_until(at).await; } @@ -2130,13 +2134,15 @@ pub(crate) mod test { use super::*; /// A context with no relay or device behind it, so broadcasts stay offline - /// while the returned origin lives. Call from a [`tokio::task::LocalSet`]. + /// while the returned origin lives, whose bus starts now. Call from a + /// [`tokio::task::LocalSet`]. pub(crate) fn context(dir: &Path) -> (Context, moq_net::origin::Producer) { let (origin, driver) = moq_net::origin::Producer::new(Default::default()); tokio::task::spawn_local(moq_net::time::run(driver)); let (bus, _) = Bus::new(Default::default()).unwrap(); let context = Context { bus, + bus_start: kio::clock::now(), origin: origin.consume(), dir: dir.to_path_buf(), output: moq_video::Output::Cpu, @@ -2232,7 +2238,8 @@ pub(crate) mod test { } /// Mix `mixer` along with the clock from `at` to `until`, 10 ms at a time, - /// returning the loudest `mix` got. + /// each block once the clock has passed it, returning the loudest `mix` + /// got. A bus started at `at` mixes on the clock as the program's does. pub(crate) async fn listen(mixer: &mut moq_audio::bus::Driver, mix: &Mix, at: &mut Instant, until: Instant) -> f32 { const STEP: Duration = Duration::from_millis(10); let frames = @@ -2787,8 +2794,8 @@ mod tests { audio.write(&moq_audio::Frame::new(silence.into(), stamp)).unwrap(); picture(&mut live.video, &mut live.encoder, 1_000); let target = until(|| clock.target()).await; - assert!(target + crate::playout::BUS_LATENCY > VIDEO_DELAY, "{target:?}"); - let due = arrived + target + crate::playout::BUS_LATENCY; + 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; until_time(due - Duration::from_millis(1)).await; assert!(feed.next_picture().is_none(), "shown before its sound"); @@ -3134,7 +3141,7 @@ mod media_tests { /// The next picture `feed` shows, passing over the clear a restart puts up, /// and when it showed. The clock moves only to a play's beginning, which - /// comes a bus latency after its first picture decodes. + /// comes the write lead after its first picture decodes. async fn shown(feed: &Feed) -> (moq_net::Timestamp, Instant) { loop { let next = until(|| match feed.next_picture() { @@ -3290,8 +3297,7 @@ mod media_tests { let (mut context, _origin) = context(dir.path()); let (bus, mut mixer) = Bus::new(Default::default()).unwrap(); context.bus = bus; - let mut at = kio::clock::now(); - mixer.advance(480); + let mut at = context.bus_start; let mut sources = Sources::new(context); let mut collection = media(dir.path(), &[("clip.mkv", CLIP, false, false)]); let red = Source::Color { @@ -3413,32 +3419,29 @@ mod media_tests { /// 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. + /// The bus mixed along with a paused clock a block at a time, each once + /// the clock has passed it, as the program's audio clock mixes it, 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. + /// When the bus starts, 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 { + /// Mix along with the clock from `sources`' bus start. + fn new(mixer: moq_audio::bus::Driver, sources: &Sources, mix: &Mix) -> Self { + let origin = sources.context().bus_start; + Self { mixer, reader: mix.reader(), - origin: now - BLOCK, - at: now, + origin, + at: origin, heard: Vec::new(), - }; - ear.mix(); - ear + } } /// Move the clock a block on, and mix that block. @@ -3479,7 +3482,7 @@ mod media_tests { } /// A file's first picture shows as its first sound is heard, within a bus - /// block, rather than a bus latency ahead of it. + /// block, rather than ahead of it. #[tokio::test(start_paused = true)] async fn a_media_files_sound_plays_with_its_pictures() { local(async { @@ -3491,7 +3494,7 @@ mod media_tests { 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); + let mut ear = Ear::new(mixer, &sources, &program); start(&sources, drivers).await; in_step(&mut ear, sources.get("clip.mkv").unwrap()).await; }) @@ -3499,7 +3502,7 @@ mod media_tests { } /// A replay's first picture shows as its first sound is heard, within a - /// bus block, rather than a bus latency ahead of it. + /// bus block, rather than ahead of it. #[tokio::test(start_paused = true)] async fn a_replays_sound_plays_with_its_pictures() { local(async { @@ -3513,7 +3516,7 @@ mod media_tests { collection.audio_scenes[0].sounds.insert("replay".into()); let drivers = sources.sync(&collection); let program = sources.mix(crate::AudioScene::DEFAULT).unwrap().clone(); - let mut ear = Ear::new(mixer, &program); + let mut ear = Ear::new(mixer, &sources, &program); start(&sources, drivers).await; // The clip's whole second, pictures and sound from its start. @@ -3548,7 +3551,7 @@ mod media_tests { collection.audio_scenes[0].sounds.insert("replay".into()); let drivers = sources.sync(&collection); let program = sources.mix(crate::AudioScene::DEFAULT).unwrap().clone(); - let mut ear = Ear::new(mixer, &program); + let mut ear = Ear::new(mixer, &sources, &program); start(&sources, drivers).await; // The clip's sound alone, from 5 ms in: inside its first 20 ms frame. @@ -3622,7 +3625,7 @@ mod media_tests { 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); + let mut ear = Ear::new(mixer, &sources, &program); start(&sources, drivers).await; let feed = sources.get("tone.webm").unwrap(); diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 15dba650..2f311289 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -783,7 +783,10 @@ impl Program { let published = match collection.published.is_empty() { true => publisher.finish(), false => { - let start = at.saturating_sub(state.compositor.program_since()); + // To the whole microsecond as the frame's stamp is, rather than + // rounded as the wire would, so a cut's layout starts on its frame. + let start = at.saturating_sub(state.compositor.program_since()).as_micros(); + let start = Duration::from_micros(start as u64); let layout = moq_scene::layout::Layout::new(collection, state.compositor.program_scene(), start); publisher.update(layout, stamp) } @@ -938,8 +941,12 @@ impl Engine { // Absolute, so a copied image names the same file in another collection. let dir = autosave.path().parent().filter(|dir| !dir.as_os_str().is_empty()); let dir = std::path::absolute(dir.unwrap_or(Path::new(".")))?; + // The program clock's start, from which the audio clock advances the + // bus too, so the sources place their sound on it. + let start = Instant::now(); let sources = Sources::new(source::Context { bus: bus.clone(), + bus_start: start, origin: sources, dir, output: moq_video::render::output(&gpu.device), @@ -999,7 +1006,6 @@ impl Engine { sources: _, } = publish; - let start = Instant::now(); // Where the program clock starts on the catalog's, so the wall time the // catalog advertises for a frame is when the program drew it. let epoch = Duration::from(clock.capture(start)?); @@ -1038,6 +1044,13 @@ impl Engine { // already running, which would otherwise keep this engine and its // publication alive beside a retry's. let starting = Starting(Some(engine.clone())); + // The audio clock first, so the bus keeps up with the program clock + // by the time the program starts the sources writing to it. + let mix = engine.clone(); + let config = bus.config().clone(); + let rate = config.sample_rate; + let thread = clocked("audio".into(), async move || mix.run_audio(start, config, mixer).await)?; + engine.threads.lock().unwrap().push(thread); let program = engine.clone(); let encoders = Encoders { main, @@ -1049,11 +1062,6 @@ impl Engine { program.run_program(start, epoch, encoders).await })?; engine.threads.lock().unwrap().push(thread); - let mix = engine.clone(); - let config = bus.config().clone(); - let rate = config.sample_rate; - let thread = clocked("audio".into(), async move || mix.run_audio(start, config, mixer).await)?; - engine.threads.lock().unwrap().push(thread); // Each on its own thread, so an encoder that falls behind skips its own // audio rather than holding up the bus or the other. let audio = [ diff --git a/rs/moq-studio/src/feed.rs b/rs/moq-studio/src/feed.rs index b3b058c4..6813a089 100644 --- a/rs/moq-studio/src/feed.rs +++ b/rs/moq-studio/src/feed.rs @@ -322,6 +322,7 @@ mod tests { let origin = moq_tokio::origin::spawn(); let mut sources = Sources::new(moq_scene::source::Context { bus: bus.clone(), + bus_start: kio::clock::now(), origin: origin.consume(), dir: ".".into(), output: moq_video::Output::Cpu, diff --git a/rs/moq-studio/tests/browser.rs b/rs/moq-studio/tests/browser.rs index 2b7de377..8b7a0120 100644 --- a/rs/moq-studio/tests/browser.rs +++ b/rs/moq-studio/tests/browser.rs @@ -83,6 +83,7 @@ impl Studio { let helper = Helper::new(program, profile, &adapter, &device, &queue); let mut sources = Sources::new(Context { bus, + bus_start: kio::clock::now(), origin: origin.consume(), dir: dir.path().to_path_buf(), output: moq_video::render::output(&device), From fa4c4cad3ea265fa34b442b1ea77ef38b6e55106 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:24:47 -0800 Subject: [PATCH 3/7] chore(quest): pin-sync no longer waits on playout-positioned Co-Authored-By: Claude Opus 5.5 --- quest/m0/pin-sync.md | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/quest/m0/pin-sync.md b/quest/m0/pin-sync.md index 9456f3b1..eef73802 100644 --- a/quest/m0/pin-sync.md +++ b/quest/m0/pin-sync.md @@ -12,10 +12,9 @@ stall does on it. ## Plan Decided 2026-11-09 (maintainer, /quest-audit): ✅ crossing #8272 is this -quest, after -[Playout writes land on positioned bus channels](/quest/m0/playout-positioned.md) -(rejected: crossing it in the -positioned bus pin). Both rewrite +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. @@ -81,7 +80,3 @@ kio's `Instant` loses `+` and `-` with a `Duration`, which `playout.rs` and Check: the maintainer answers in `/quest-unblock`, and the choice is recorded here. - -## Required - -- [Playout writes land on positioned bus channels](/quest/m0/playout-positioned.md) - the `playout.rs` rewrite this crosses on top of From 42779fb293c7f40637bf61b70610c4c3ae199e49 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:25:08 -0800 Subject: [PATCH 4/7] style: rustfmt Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 5 ++++- rs/moq-scene/src/source.rs | 7 ++++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 7d6d7aa0..e30f7318 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -492,7 +492,10 @@ mod tests { writer.write(at(60), &frame(1.0, 20)).await.unwrap(); ear.until(at(30)).await; // Stamped half a millisecond after the last ended. - writer.write(at(80) + Duration::from_micros(500), &frame(2.0, 20)).await.unwrap(); + writer + .write(at(80) + Duration::from_micros(500), &frame(2.0, 20)) + .await + .unwrap(); // Overlapping the last by 5 ms. writer.write(at(95), &frame(3.0, 20)).await.unwrap(); diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index f2ac6ff4..428493df 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -1849,7 +1849,12 @@ impl Driver { let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); input.format = moq_audio::Format::F32; let channel = self.context.bus.positioned(input, strip)?; - Ok(Writer::new(channel, self.context.bus_start, bus.sample_rate, bus.layout.channels())) + Ok(Writer::new( + channel, + self.context.bus_start, + bus.sample_rate, + bus.layout.channels(), + )) } } From c66d9d0675a7a9b18ec2ff5f9150999c1dfecbd9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 15:33:34 -0800 Subject: [PATCH 5/7] fix(scene): bound playout writes by how far the bus has mixed A source's playout waits for a bus lagging the clock instead of failing on a write past its reach: the audio clock's Mixer publishes each advance, and the Writer bounds each write by it. Co-Authored-By: Claude Opus 5.5 --- quest/m0/playout-lag.md | 25 ---------- rs/moq-scene/src/compositor.rs | 9 ++-- rs/moq-scene/src/playout.rs | 91 +++++++++++++++++++--------------- rs/moq-scene/src/source.rs | 86 +++++++++++++++++++++++++------- rs/moq-studio/src/engine.rs | 12 +++-- rs/moq-studio/src/feed.rs | 17 ++++--- rs/moq-studio/tests/browser.rs | 7 +-- 7 files changed, 148 insertions(+), 99 deletions(-) delete mode 100644 quest/m0/playout-lag.md diff --git a/quest/m0/playout-lag.md b/quest/m0/playout-lag.md deleted file mode 100644 index 2346fed4..00000000 --- a/quest/m0/playout-lag.md +++ /dev/null @@ -1,25 +0,0 @@ -# [S] A lagging audio clock holds playout writes instead of failing the source - -## Goal - -When the program's audio clock thread falls behind the program clock, a -broadcast, media, or replay source's sound waits for the bus to catch up, or -loses only what the bus can no longer place, rather than failing the source. - -## Plan - -`playout::Writer` (`rs/moq-scene/src/playout.rs`) maps a moment to a bus -position from the program clock's start, and bounds each write by where the -clock says the bus is, since moq-audio's `Positioned` does not say where the -bus has mixed to. A write ends at most `WRITE_LEAD` plus one frame (about -70 ms) past the clock, inside `bus::REACH` (200 ms). If the audio thread lags -the clock by more than the rest (about 130 ms), `Positioned::write` refuses the -write as past its reach, with the same `Error::Unsupported` it gives a -non-finite sample, and the source fails: a media source stays failed until -restarted. - -Recommended: moq-audio's `Positioned` reports the bus's next position (or -refuses a write past its reach with an error of its own), landed in moq-dev/moq -with a moq-side test and pinned here with `just upgrade-moq`. The writer then -cuts against the clock as now, but waits on the bus's actual position for -what lies past its reach. diff --git a/rs/moq-scene/src/compositor.rs b/rs/moq-scene/src/compositor.rs index 01a921e1..963f9351 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -2258,7 +2258,7 @@ mod tests { /// Mix the bus sample-exact on the clock from `start`, as the program's /// audio clock does. - async fn mix(mut mixer: moq_audio::bus::Driver, start: Instant) { + async fn mix(mut mixer: crate::source::Mixer, start: Instant) { let rate = u128::from(moq_audio::bus::Config::default().sample_rate); let mut clock = kio::clock::interval_at(start, Duration::from_millis(10)); clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); @@ -2293,9 +2293,12 @@ mod tests { let (device, queue) = gpu().await; let (mut context, origin) = context(dir.path()); context.output = moq_video::render::output(&device); - let (bus, mixer) = moq_audio::bus::Bus::new(Default::default()).unwrap(); + let (bus, driver) = moq_audio::bus::Bus::new(Default::default()).unwrap(); + let begin = Instant::now(); + let (mixer, mixed) = crate::source::Mixer::new(driver, begin); context.bus = bus; - tokio::task::spawn_local(mix(mixer, context.bus_start)); + context.mixed = mixed; + tokio::task::spawn_local(mix(mixer, begin)); let mut sources = Sources::new(context); let drivers = sources.sync(&collection); start(&sources, drivers).await; diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index e30f7318..b644bcad 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -20,6 +20,8 @@ use moq_audio::bus; use moq_mux::timeline; use tokio::sync::Notify; +use crate::source::Mixed; + /// How far a broadcast without audio plays behind its arrivals, as under /// `moq play --delay auto`. With audio, video follows the audio's target. pub(crate) const VIDEO_DELAY: Duration = Duration::from_millis(100); @@ -37,13 +39,8 @@ pub(crate) const MAX_DELAY: Duration = Duration::from_secs(2); /// over this long before its moment to be mixed whole. pub(crate) const WRITE_LEAD: Duration = Duration::from_millis(50); -/// How far ahead of the clock a [`Writer`] lets a write end: the bus's reach, -/// less the lead, so the bus may lag the clock by up to the lead before a -/// write ends past what it holds. -const AHEAD: Duration = bus::REACH.saturating_sub(WRITE_LEAD); - const _: () = assert!( - WRITE_LEAD.as_nanos() < AHEAD.as_nanos(), + WRITE_LEAD.as_nanos() < bus::REACH.as_nanos(), "a sound handed over the lead ahead is in the bus's reach" ); @@ -195,7 +192,7 @@ pub(crate) fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp /// Writes placed frames into a positioned bus channel, each at the bus /// position its moment maps to, so it sounds exactly then. /// -/// The bus's driver advances sample-exact on the program clock from `start`, +/// The bus's [`Mixer`](crate::source::Mixer) advances it sample-exact on the clock from its start, /// so a moment `t` is bus position `(t - start) * rate`, and the bus has never /// mixed past the position of the clock's reading. pub(crate) struct Writer { @@ -203,21 +200,20 @@ pub(crate) struct Writer { /// Bytes per sample frame of the channel's `f32` input. stride: usize, rate: u32, - /// When the bus mixes its position 0. - start: Instant, + mixed: Mixed, /// The bus position the last write ended at, which the next follows. end: i128, } impl Writer { - /// Write into `channel` on a bus mixing position 0 at `start`, taking - /// interleaved `f32` at the bus `rate` in `channels`. - pub(crate) fn new(channel: bus::Positioned, start: Instant, rate: u32, channels: u32) -> Self { + /// Write into `channel` on the bus `mixed` follows, taking interleaved + /// `f32` at the bus `rate` in `channels`. + pub(crate) fn new(channel: bus::Positioned, mixed: Mixed, rate: u32, channels: u32) -> Self { Self { channel, stride: channels as usize * size_of::(), rate, - start, + mixed, end: 0, } } @@ -226,52 +222,53 @@ impl Writer { /// /// A frame that starts within [`CONTIGUOUS`] of where the last one ended /// continues it. The part already mixed, or overlapping the last frame, is - /// cut, so the rest still sounds on schedule. The part further ahead than - /// the bus holds waits until it is in reach. + /// cut, so the rest still sounds on schedule. The part past what the bus + /// holds waits until the bus has mixed far enough to hold it, however far + /// behind the clock it runs. Once the bus's mixer is gone, nothing mixes, + /// and what is left is dropped. pub(crate) async fn write(&mut self, sounds: Instant, data: &[u8]) -> Result<(), moq_audio::Error> { let mut position = self.position(sounds); if position.abs_diff(self.end) <= self.frames(CONTIGUOUS) { position = self.end; } let end = position + (data.len() / self.stride) as i128; + let reach = self.frames(bus::REACH) as i128; loop { - let mixed = self.mixed(kio::clock::now()); - let from = position.max(self.end).max(mixed); - let until = end.min(mixed + self.frames(AHEAD) as i128); + let from = position.max(self.end).max(self.ceiling(kio::clock::now())); + if from >= end { + return Ok(()); + } + // The bus may have mixed further since, which only adds room. + let next = *self.mixed.position.borrow_and_update() as i128; + let until = end.min(next + reach); if from < until { let bytes = |at: i128| (at - position) as usize * self.stride; self.channel.write(from as u64, &data[bytes(from)..bytes(until)])?; self.end = until; } - if until >= end { + if until >= end || self.mixed.position.changed().await.is_err() { return Ok(()); } - kio::clock::sleep_until(self.instant(until) - WRITE_LEAD).await; } } /// The bus position sounding at `at`, to the nearest frame. fn position(&self, at: Instant) -> i128 { - let nanos = match at.checked_duration_since(self.start) { + let start = self.mixed.start; + let nanos = match at.checked_duration_since(start) { Some(after) => after.as_nanos() as i128, - None => -(self.start.duration_since(at).as_nanos() as i128), + None => -(start.duration_since(at).as_nanos() as i128), }; (nanos * i128::from(self.rate) + 500_000_000).div_euclid(1_000_000_000) } - /// The first bus position not mixed by `now` at the latest: rounded up, - /// since the bus mixes a frame only once the clock has passed it. - fn mixed(&self, now: Instant) -> i128 { - let nanos = now.saturating_duration_since(self.start).as_nanos(); + /// The first bus position the bus cannot have mixed by `now`: rounded up, + /// since it mixes a frame only once the clock has passed it. + fn ceiling(&self, now: Instant) -> i128 { + let nanos = now.saturating_duration_since(self.mixed.start).as_nanos(); (nanos * u128::from(self.rate)).div_ceil(1_000_000_000) as i128 } - /// When bus position `position` sounds. - fn instant(&self, position: i128) -> Instant { - let nanos = position.max(0) as u128 * 1_000_000_000 / u128::from(self.rate); - self.start + Duration::from_nanos(nanos as u64) - } - /// The whole frames `duration` spans. fn frames(&self, duration: Duration) -> u128 { duration.as_nanos() * u128::from(self.rate) / 1_000_000_000 @@ -281,6 +278,7 @@ impl Writer { #[cfg(test)] mod tests { use super::*; + use crate::source::Mixer; const FRAME: Duration = Duration::from_millis(20); const TARGET: Duration = Duration::from_millis(100); @@ -426,8 +424,10 @@ mod tests { /// The bus mixed a block at a time along with a paused clock, as the /// program's audio clock mixes it, noting each change in what one mix heard. struct Ear { - driver: bus::Driver, + mixer: Mixer, reader: bus::Reader, + /// The bus position of the next frame mixed. + next: u64, start: Instant, rate: u64, /// Each bus position the first channel changed at, and what to. @@ -440,13 +440,14 @@ mod tests { /// Mix every block the clock passes up to `until`, as it passes each. async fn until(&mut self, until: Instant) { loop { - let next = self.driver.position() + Self::BLOCK; + let next = self.next + Self::BLOCK; let due = self.start + Duration::from_nanos(next * 1_000_000_000 / self.rate); if due > until { return; } kio::clock::sleep_until(due).await; - self.driver.advance(Self::BLOCK as usize); + self.mixer.advance(Self::BLOCK as usize); + self.next = next; let std::task::Poll::Ready(Some(read)) = self.reader.poll_read(&kio::Waiter::noop()) else { panic!("an advance publishes a block"); }; @@ -463,8 +464,9 @@ mod tests { /// A frame sounds at the bus position its moment maps to, sample-exact: /// handed over ahead, continuing the last across a stamp's rounding, - /// overlapping it, late, wholly late, while the bus lags the clock, and - /// longer than the bus holds. Whatever cannot sound when it says is cut. + /// overlapping it, late, wholly late, while the bus lags the clock a little + /// and far past what it holds, and longer than the bus holds. Whatever + /// cannot sound when it says is cut. #[tokio::test(start_paused = true)] async fn a_write_sounds_at_its_position() { let (bus, driver) = bus::Bus::new(Default::default()).unwrap(); @@ -476,9 +478,11 @@ mod tests { let mix = bus.mix(); mix.route(&strip); let start = kio::clock::now(); - let mut writer = Writer::new(channel, start, config.sample_rate, config.layout.channels()); + let (mixer, mixed) = Mixer::new(driver, start); + let mut writer = Writer::new(channel, mixed, config.sample_rate, config.layout.channels()); let mut ear = Ear { - driver, + mixer, + next: 0, reader: mix.reader(), start, rate: u64::from(config.sample_rate), @@ -516,6 +520,13 @@ mod tests { let (written, ()) = tokio::join!(writer.write(at(4_050), &long), ear.until(at(4_500))); written.unwrap(); + // The bus stalls half a second behind the clock: the write waits for + // it to catch up, rather than landing past what it holds. + kio::clock::sleep_until(at(5_000)).await; + let short = frame(8.0, 20); + let (written, ()) = tokio::join!(writer.write(at(5_050), &short), ear.until(at(5_200))); + written.unwrap(); + assert_eq!( ear.heard, [ @@ -529,6 +540,8 @@ mod tests { (48 * 3_070, 0.0), (48 * 4_050, 7.0), (48 * 4_350, 0.0), + (48 * 5_050, 8.0), + (48 * 5_070, 0.0), ] ); } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 428493df..f1e4a93a 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -241,10 +241,9 @@ pub struct Context { /// Where microphones, system audio, and broadcasts feed their sound, and /// where each of the collection's audio scenes is mixed. pub bus: Bus, - /// When `bus` mixes its position 0: its driver advances sample-exact on - /// the clock from here, so a sound due at `t` plays at bus position - /// `(t - bus_start) * rate`, and never mixes past where the clock is. - pub bus_start: Instant, + /// How far `bus`'s [`Mixer`] has mixed it, on which broadcast, media, and + /// replay sources place their sound. + pub mixed: Mixed, /// Where broadcast sources are announced: the studio's relay connection. pub origin: moq_net::origin::Consumer, /// What a relative image, LUT, or overlay asset path is relative to: the @@ -259,6 +258,48 @@ pub struct Context { pub publish: Option, } +/// Advances a bus sample-exact on the clock from its start, as the clock's +/// owner moves it, telling the sources how far it has mixed. +pub struct Mixer { + driver: moq_audio::bus::Driver, + position: watch::Sender, +} + +impl Mixer { + /// Advance `driver`'s bus, which mixes its position 0 at `start`, and + /// follow it with the returned [`Mixed`]. + pub fn new(driver: moq_audio::bus::Driver, start: Instant) -> (Self, Mixed) { + let (position, follow) = watch::channel(driver.position()); + let mixed = Mixed { + start, + position: follow, + }; + (Self { driver, position }, mixed) + } + + /// Mix the next `frames` of every mix, as the clock moves by that much. + pub fn advance(&mut self, frames: usize) { + self.driver.advance(frames); + self.position.send_replace(self.driver.position()); + } +} + +/// How far a [`Mixer`] has mixed its bus, and when it mixes position 0. +#[derive(Clone)] +pub struct Mixed { + pub(crate) start: Instant, + /// The bus position of the next frame mixed. + pub(crate) position: watch::Receiver, +} + +impl Mixed { + /// When the bus mixes its position 0, so a sound due at `t` plays at bus + /// position `(t - start) * rate`. + pub fn start(&self) -> Instant { + self.start + } +} + /// The running sources of a collection, and the mix of each audio scene. pub struct Sources { context: Context, @@ -1851,7 +1892,7 @@ impl Driver { let channel = self.context.bus.positioned(input, strip)?; Ok(Writer::new( channel, - self.context.bus_start, + self.context.mixed.clone(), bus.sample_rate, bus.layout.channels(), )) @@ -2144,10 +2185,11 @@ pub(crate) mod test { pub(crate) fn context(dir: &Path) -> (Context, moq_net::origin::Producer) { let (origin, driver) = moq_net::origin::Producer::new(Default::default()); tokio::task::spawn_local(moq_net::time::run(driver)); - let (bus, _) = Bus::new(Default::default()).unwrap(); + let (bus, driver) = Bus::new(Default::default()).unwrap(); + let (_, mixed) = Mixer::new(driver, kio::clock::now()); let context = Context { bus, - bus_start: kio::clock::now(), + mixed, origin: origin.consume(), dir: dir.to_path_buf(), output: moq_video::Output::Cpu, @@ -2245,7 +2287,7 @@ pub(crate) mod test { /// Mix `mixer` along with the clock from `at` to `until`, 10 ms at a time, /// each block once the clock has passed it, returning the loudest `mix` /// got. A bus started at `at` mixes on the clock as the program's does. - pub(crate) async fn listen(mixer: &mut moq_audio::bus::Driver, mix: &Mix, at: &mut Instant, until: Instant) -> f32 { + pub(crate) async fn listen(mixer: &mut Mixer, mix: &Mix, at: &mut Instant, until: Instant) -> f32 { const STEP: Duration = Duration::from_millis(10); let frames = (STEP.as_nanos() * u128::from(moq_audio::bus::Config::default().sample_rate) / 1_000_000_000) as usize; @@ -3300,9 +3342,11 @@ mod media_tests { local(async { let dir = tempfile::tempdir().unwrap(); let (mut context, _origin) = context(dir.path()); - let (bus, mut mixer) = Bus::new(Default::default()).unwrap(); + let (bus, driver) = Bus::new(Default::default()).unwrap(); + let mut at = kio::clock::now(); + let (mut mixer, mixed) = Mixer::new(driver, at); context.bus = bus; - let mut at = context.bus_start; + context.mixed = mixed; let mut sources = Sources::new(context); let mut collection = media(dir.path(), &[("clip.mkv", CLIP, false, false)]); let red = Source::Color { @@ -3428,7 +3472,7 @@ mod media_tests { /// the clock has passed it, as the program's audio clock mixes it, keeping /// the first channel of what one mix heard. struct Ear { - mixer: moq_audio::bus::Driver, + mixer: Mixer, reader: moq_audio::bus::Reader, /// When the bus starts, so a frame `n` in sounds `n` samples after it. origin: Instant, @@ -3438,8 +3482,8 @@ mod media_tests { impl Ear { /// Mix along with the clock from `sources`' bus start. - fn new(mixer: moq_audio::bus::Driver, sources: &Sources, mix: &Mix) -> Self { - let origin = sources.context().bus_start; + fn new(mixer: Mixer, sources: &Sources, mix: &Mix) -> Self { + let origin = sources.context().mixed.start(); Self { mixer, reader: mix.reader(), @@ -3493,8 +3537,10 @@ mod media_tests { local(async { let dir = tempfile::tempdir().unwrap(); let (mut context, _origin) = context(dir.path()); - let (bus, mixer) = Bus::new(Default::default()).unwrap(); + let (bus, driver) = Bus::new(Default::default()).unwrap(); + let (mixer, mixed) = Mixer::new(driver, kio::clock::now()); context.bus = bus; + context.mixed = mixed; let mut sources = Sources::new(context); let collection = media(dir.path(), &[("clip.mkv", CLIP, false, false)]); let drivers = sources.sync(&collection); @@ -3513,8 +3559,10 @@ mod media_tests { local(async { let dir = tempfile::tempdir().unwrap(); let (mut context, _origin) = context(dir.path()); - let (bus, mixer) = Bus::new(Default::default()).unwrap(); + let (bus, driver) = Bus::new(Default::default()).unwrap(); + let (mixer, mixed) = Mixer::new(driver, kio::clock::now()); context.bus = bus; + context.mixed = mixed; let mut sources = Sources::new(context); let mut collection = Collection::new(); collection.sources.insert("replay".into(), Source::Replay); @@ -3548,8 +3596,10 @@ mod media_tests { local(async { let dir = tempfile::tempdir().unwrap(); let (mut context, _origin) = context(dir.path()); - let (bus, mixer) = Bus::new(Default::default()).unwrap(); + let (bus, driver) = Bus::new(Default::default()).unwrap(); + let (mixer, mixed) = Mixer::new(driver, kio::clock::now()); context.bus = bus; + context.mixed = mixed; let mut sources = Sources::new(context); let mut collection = Collection::new(); collection.sources.insert("replay".into(), Source::Replay); @@ -3624,8 +3674,10 @@ mod media_tests { local(async { let dir = tempfile::tempdir().unwrap(); let (mut context, _origin) = context(dir.path()); - let (bus, mixer) = Bus::new(Default::default()).unwrap(); + let (bus, driver) = Bus::new(Default::default()).unwrap(); + let (mixer, mixed) = Mixer::new(driver, kio::clock::now()); context.bus = bus; + context.mixed = mixed; let mut sources = Sources::new(context); let collection = media(dir.path(), &[("tone.webm", TONE, true, false)]); let drivers = sources.sync(&collection); diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 2f311289..208ed9fa 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -882,7 +882,7 @@ impl Engine { recordings, room, } = launch; - let (bus, mixer) = moq_audio::bus::Bus::new(Default::default())?; + let (bus, driver) = moq_audio::bus::Bus::new(Default::default())?; let subscribe = moq_tokio::origin::spawn(); // Every relay feeds `subscribe`, so a guest either one routes is seen. let guests = match &room { @@ -944,9 +944,10 @@ impl Engine { // The program clock's start, from which the audio clock advances the // bus too, so the sources place their sound on it. let start = Instant::now(); + let (mixer, mixed) = source::Mixer::new(driver, start); let sources = Sources::new(source::Context { bus: bus.clone(), - bus_start: start, + mixed, origin: sources, dir, output: moq_video::render::output(&gpu.device), @@ -1044,8 +1045,9 @@ impl Engine { // already running, which would otherwise keep this engine and its // publication alive beside a retry's. let starting = Starting(Some(engine.clone())); - // The audio clock first, so the bus keeps up with the program clock - // by the time the program starts the sources writing to it. + // The audio clock first, so the bus keeps up with the program clock by + // the time the program starts the sources, whose sound otherwise waits + // for it. let mix = engine.clone(); let config = bus.config().clone(); let rate = config.sample_rate; @@ -1602,7 +1604,7 @@ impl Engine { /// Runs without a relay too, so the meters move and sources never back up, /// until the studio [stops](Self::stop), which ends every mix, so each /// output reads its mix to the end. - async fn run_audio(&self, start: Instant, config: moq_audio::bus::Config, mut mixer: moq_audio::bus::Driver) { + async fn run_audio(&self, start: Instant, config: moq_audio::bus::Config, mut mixer: source::Mixer) { let rate = u128::from(config.sample_rate); let mut clock = kio::clock::interval_at(start, AUDIO_PERIOD); clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); diff --git a/rs/moq-studio/src/feed.rs b/rs/moq-studio/src/feed.rs index 6813a089..e1d6cba4 100644 --- a/rs/moq-studio/src/feed.rs +++ b/rs/moq-studio/src/feed.rs @@ -266,6 +266,7 @@ async fn list( #[cfg(test)] mod tests { use moq_audio::bus::{Bus, Live}; + use moq_scene::source::{Mixed, Mixer}; use moq_scene::{History, Source}; use url::Url; @@ -309,8 +310,8 @@ mod tests { } /// Alice and bob in `room` with their cameras on, and a collection playing - /// each camera, whose sources run on `bus`. - fn guests(room: &Room, bus: &Bus) -> (Collection, Sources) { + /// each camera, whose sources run on `bus`, which `mixed` follows. + fn guests(room: &Room, bus: &Bus, mixed: Mixed) -> (Collection, Sources) { let mut collection = Collection::new(); for guest in ["alice", "bob"] { let camera = format!("{guest}/camera.hang"); @@ -322,7 +323,7 @@ mod tests { let origin = moq_tokio::origin::spawn(); let mut sources = Sources::new(moq_scene::source::Context { bus: bus.clone(), - bus_start: kio::clock::now(), + mixed, origin: origin.consume(), dir: ".".into(), output: moq_video::Output::Cpu, @@ -373,8 +374,9 @@ mod tests { let room = Arc::new(Room::new(&connect, "guests", key, page, EXPIRY).unwrap()); let origin = moq_tokio::origin::spawn(); let (mut feeds, _program) = feeds(&room, &origin); - let (bus, mut driver) = Bus::new(Default::default()).unwrap(); - let (mut collection, mut sources) = guests(&room, &bus); + let (bus, driver) = Bus::new(Default::default()).unwrap(); + let (mut driver, mixed) = Mixer::new(driver, kio::clock::now()); + let (mut collection, mut sources) = guests(&room, &bus, mixed); // Copies of alice's camera listed first, one muted and one with its // fader down, which nobody hears on program, so her mix-minus takes // out the copy they do. @@ -462,8 +464,9 @@ mod tests { let room = relay.room(); let origin = room.origin().unwrap().clone(); let (mut feeds, _program) = feeds(&room, &origin); - let (bus, _driver) = Bus::new(Default::default()).unwrap(); - let (collection, sources) = guests(&room, &bus); + let (bus, driver) = Bus::new(Default::default()).unwrap(); + let (_driver, mixed) = Mixer::new(driver, kio::clock::now()); + let (collection, sources) = guests(&room, &bus, mixed); feeds.sync(&collection, &sources); let read = moq_tokio::origin::spawn(); diff --git a/rs/moq-studio/tests/browser.rs b/rs/moq-studio/tests/browser.rs index 8b7a0120..1855748f 100644 --- a/rs/moq-studio/tests/browser.rs +++ b/rs/moq-studio/tests/browser.rs @@ -27,7 +27,7 @@ struct Studio { device: wgpu::Device, queue: wgpu::Queue, /// Advances the bus, which meters each source's strip. - mixer: moq_audio::bus::Driver, + mixer: moq_scene::source::Mixer, helper: Helper, _origin: moq_net::origin::Producer, _dir: tempfile::TempDir, @@ -79,11 +79,12 @@ impl Studio { let dir = tempfile::tempdir().unwrap(); let (origin, driver) = moq_net::origin::Producer::new(Default::default()); tokio::spawn(moq_net::time::run(driver)); - let (bus, mixer) = moq_audio::bus::Bus::new(Default::default()).unwrap(); + let (bus, driver) = moq_audio::bus::Bus::new(Default::default()).unwrap(); + let (mixer, mixed) = moq_scene::source::Mixer::new(driver, kio::clock::now()); let helper = Helper::new(program, profile, &adapter, &device, &queue); let mut sources = Sources::new(Context { bus, - bus_start: kio::clock::now(), + mixed, origin: origin.consume(), dir: dir.path().to_path_buf(), output: moq_video::render::output(&device), From 897d0f3db6d6199829a5ebbc53ef4aba81ddb3b3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 16:07:45 -0800 Subject: [PATCH 6/7] fix(scene): write playout audio atomically with the bus's progress, and bound the broadcast queue --- rs/moq-scene/src/playout.rs | 69 +++++++++++++++++++++---------------- rs/moq-scene/src/source.rs | 31 +++++++++++------ 2 files changed, 60 insertions(+), 40 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index b644bcad..127f18ff 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -192,9 +192,8 @@ pub(crate) fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp /// Writes placed frames into a positioned bus channel, each at the bus /// position its moment maps to, so it sounds exactly then. /// -/// The bus's [`Mixer`](crate::source::Mixer) advances it sample-exact on the clock from its start, -/// so a moment `t` is bus position `(t - start) * rate`, and the bus has never -/// mixed past the position of the clock's reading. +/// The bus's [`Mixer`](crate::source::Mixer) advances it sample-exact on the +/// clock from its start, so a moment `t` is bus position `(t - start) * rate`. pub(crate) struct Writer { channel: bus::Positioned, /// Bytes per sample frame of the channel's `f32` input. @@ -221,11 +220,11 @@ impl Writer { /// Write `data` to sound at `sounds`, handed over [`WRITE_LEAD`] ahead. /// /// A frame that starts within [`CONTIGUOUS`] of where the last one ended - /// continues it. The part already mixed, or overlapping the last frame, is - /// cut, so the rest still sounds on schedule. The part past what the bus - /// holds waits until the bus has mixed far enough to hold it, however far - /// behind the clock it runs. Once the bus's mixer is gone, nothing mixes, - /// and what is left is dropped. + /// continues it. The part the bus has already mixed, or overlapping the + /// last frame, is cut, so the rest still sounds on schedule. The part past + /// what the bus holds waits until the bus has mixed far enough to hold it, + /// however far behind the clock it runs. Once the bus's mixer is gone, + /// nothing mixes, and what is left is dropped. pub(crate) async fn write(&mut self, sounds: Instant, data: &[u8]) -> Result<(), moq_audio::Error> { let mut position = self.position(sounds); if position.abs_diff(self.end) <= self.frames(CONTIGUOUS) { @@ -234,18 +233,20 @@ impl Writer { let end = position + (data.len() / self.stride) as i128; let reach = self.frames(bus::REACH) as i128; loop { - let from = position.max(self.end).max(self.ceiling(kio::clock::now())); + // The bus cannot advance while its position is borrowed, so what + // is written lands where it has not mixed. + let next = self.mixed.position.borrow_and_update(); + let from = position.max(self.end).max(*next as i128); if from >= end { return Ok(()); } - // The bus may have mixed further since, which only adds room. - let next = *self.mixed.position.borrow_and_update() as i128; - let until = end.min(next + reach); + let until = end.min(*next as i128 + reach); if from < until { let bytes = |at: i128| (at - position) as usize * self.stride; self.channel.write(from as u64, &data[bytes(from)..bytes(until)])?; self.end = until; } + drop(next); if until >= end || self.mixed.position.changed().await.is_err() { return Ok(()); } @@ -262,13 +263,6 @@ impl Writer { (nanos * i128::from(self.rate) + 500_000_000).div_euclid(1_000_000_000) } - /// The first bus position the bus cannot have mixed by `now`: rounded up, - /// since it mixes a frame only once the clock has passed it. - fn ceiling(&self, now: Instant) -> i128 { - let nanos = now.saturating_duration_since(self.mixed.start).as_nanos(); - (nanos * u128::from(self.rate)).div_ceil(1_000_000_000) as i128 - } - /// The whole frames `duration` spans. fn frames(&self, duration: Duration) -> u128 { duration.as_nanos() * u128::from(self.rate) / 1_000_000_000 @@ -446,17 +440,22 @@ mod tests { return; } kio::clock::sleep_until(due).await; - self.mixer.advance(Self::BLOCK as usize); - self.next = next; - let std::task::Poll::Ready(Some(read)) = self.reader.poll_read(&kio::Waiter::noop()) else { - panic!("an advance publishes a block"); - }; - let mut last = self.heard.last().map_or(0.0, |(_, value)| *value); - for (index, sample) in read.data.chunks(2).enumerate() { - if sample[0] != last { - last = sample[0]; - self.heard.push((read.position + index as u64, last)); - } + self.block(); + } + } + + /// Mix the next block now, whatever the clock says. + fn block(&mut self) { + self.mixer.advance(Self::BLOCK as usize); + self.next += Self::BLOCK; + let std::task::Poll::Ready(Some(read)) = self.reader.poll_read(&kio::Waiter::noop()) else { + panic!("an advance publishes a block"); + }; + let mut last = self.heard.last().map_or(0.0, |(_, value)| *value); + for (index, sample) in read.data.chunks(2).enumerate() { + if sample[0] != last { + last = sample[0]; + self.heard.push((read.position + index as u64, last)); } } } @@ -527,6 +526,14 @@ mod tests { let (written, ()) = tokio::join!(writer.write(at(5_050), &short), ear.until(at(5_200))); written.unwrap(); + // The bus mixes the block the clock reaches just as a late frame is + // handed over: its tail lands after that block. + ear.until(at(5_990)).await; + kio::clock::sleep_until(at(5_995)).await; + ear.block(); + writer.write(at(5_990), &frame(9.0, 20)).await.unwrap(); + ear.until(at(6_100)).await; + assert_eq!( ear.heard, [ @@ -542,6 +549,8 @@ mod tests { (48 * 4_350, 0.0), (48 * 5_050, 8.0), (48 * 5_070, 0.0), + (48 * 6_000, 9.0), + (48 * 6_010, 0.0), ] ); } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index f1e4a93a..53545468 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -279,8 +279,12 @@ impl Mixer { /// Mix the next `frames` of every mix, as the clock moves by that much. pub fn advance(&mut self, frames: usize) { - self.driver.advance(frames); - self.position.send_replace(self.driver.position()); + // Under the position's lock, so no source writes in between reading + // it and the bus moving past it. + self.position.send_modify(|position| { + self.driver.advance(frames); + *position = self.driver.position(); + }); } } @@ -288,7 +292,7 @@ impl Mixer { #[derive(Clone)] pub struct Mixed { pub(crate) start: Instant, - /// The bus position of the next frame mixed. + /// The bus position of the next frame mixed, which holds while borrowed. pub(crate) position: watch::Receiver, } @@ -1272,10 +1276,13 @@ impl Driver { let stride = bus.layout.channels() as usize * size_of::(); let rate = f64::from(bus.sample_rate); - // Frames wait here for their moment, bounded by the target the speaker - // holds, while every arrival is read at once: the jitter estimate times - // each one as `read` hands it over. - let queue = RefCell::new(VecDeque::new()); + // 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 + // however long a write waits on the bus, this holds no more than the + // longest target. + let queue = RefCell::new(VecDeque::<(Instant, Instant, moq_audio::Frame)>::new()); let queued = Notify::new(); let received = Cell::new(false); let receive = async { @@ -1290,7 +1297,11 @@ impl Driver { .borrow_mut() .place(frame.timestamp.into(), duration, consumer.delay(), now); if let Place::Play(sounds) = place { - queue.borrow_mut().push_back((sounds, frame)); + let mut queue = queue.borrow_mut(); + while queue.front().is_some_and(|(_, ends, _)| *ends <= now) { + queue.pop_front(); + } + queue.push_back((sounds, sounds + duration, frame)); queued.notify_one(); } } @@ -1304,9 +1315,9 @@ impl Driver { loop { let now = kio::clock::now(); speaker.borrow_mut().commit(now); - let write = queue.borrow().front().map(|(sounds, _)| *sounds - WRITE_LEAD); + 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"); + let (sounds, _, frame) = queue.borrow_mut().pop_front().expect("peeked"); writer.write(sounds, &frame.data).await?; continue; } From 56ca393b09c0743bbec7f887b2f6c30726083ead Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 11 Nov 2026 16:21:38 -0800 Subject: [PATCH 7/7] docs(scene): word the playout queue's bound --- rs/moq-scene/src/source.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 53545468..8dc5e9a1 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -292,7 +292,7 @@ impl Mixer { #[derive(Clone)] pub struct Mixed { pub(crate) start: Instant, - /// The bus position of the next frame mixed, which holds while borrowed. + /// The bus position of the next frame mixed, which cannot move while borrowed. pub(crate) position: watch::Receiver, } @@ -1280,8 +1280,8 @@ impl Driver { // 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 - // however long a write waits on the bus, this holds no more than the - // longest target. + // 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()); let queued = Notify::new(); let received = Cell::new(false);