diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 8172d1dd97a..ebe047260b4 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -260,7 +260,9 @@ the one playing. With `--video off`, `--metrics ` writes a JSON line every 250 ms with the playhead, the delay, and every gap and stall the speaker hit, in the schema the [audio quality harness](https://github.com/moq-dev/moq/tree/main/test/audio-quality) -reduces to underruns, skips, and silence. +reduces to underruns, skips, and silence. A silence the publisher declares, such +as a push-to-talk mute, plays as comfort noise at the level of the quietest audio +heard before it, and counts as no gap or stall. The broadcast's announcement is its online signal, followed until the window closes. Playback starts when the name is announced. A diff --git a/doc/concept/audio-jitter.md b/doc/concept/audio-jitter.md index cfd03a5d28c..fbdc52e32e3 100644 --- a/doc/concept/audio-jitter.md +++ b/doc/concept/audio-jitter.md @@ -733,3 +733,54 @@ cooldown, fast-mode bypass, linked stereo, passive noise, rejected active noise, and 8/44.1/48 kHz seams. Both implementations consume these exact vectors; generating expected output from the implementation under test is not conformance. + +## Comfort noise + +A [declared silence](#declared-silence) is not an underrun. A receiver whose +playback runs dry at a declared end counts no underrun or stall, and fills the +silence with comfort noise instead of plain silence, so a muted talker's +background carries on rather than the line going dead. It keeps doing so +through the refill before the next utterance plays. + +The publisher sends nothing while muted, so the level is measured from what the +receiver itself played since playback last resumed: + +``` +played(samples): # audio just played, not comfort noise or startup padding + if filling: filling = false; quietest = infinity; sum = 0; frames = 0 + for each frame: + sum += every channel's sample squared + frames += 1 + if frames == BLOCK: + quietest = min(quietest, sum / (BLOCK * channels)) + sum = 0; frames = 0 + +fill(samples): + if not filling: + filling = true + if quietest is finite: level = min(sqrt(quietest), COMFORT_MAX) + each sample = uniform(-1, 1) * level * sqrt(3) +``` + +`BLOCK` is 20 ms at the output rate, rounded, and blocks count from the first +frame played after a silence. `COMFORT_MAX` is 0.001 RMS, -60 dBFS: audio that +never falls quiet, such as music, would otherwise fill a pause with noise as +loud as itself. `level` starts at 0, so a source that was already silent, such +as a noise-suppressed microphone, gets none, and a silence after too little +audio to fill one block keeps the previous level. Uniform noise on `[-a, a]` +has an RMS of `a / sqrt(3)`, so the noise sits at `level`. Sums are `f64` in +the order written; the noise itself is not part of conformance. + +Decoded audio stops a codec's pre-skip short of the declared end when the +publisher leaves its encoder's lookahead unflushed, so the browser's ring takes +the end that early. Natively, `moq play` declares the end on the playback sink +when it reads the marker, and the stopped encoder's final frames that follow +the marker still play before the silence. + +### Comfort corpus + +`audio-jitter/comfort/corpus.json` holds cases of constant runs played, each +followed by the level the next silence must fill at, within 1e-9. It covers the +quietest block rather than the last, blocks straddling a level change, a stereo +mean, the -60 dBFS cap, an already silent source, and a silence that keeps or +remeasures the level. diff --git a/doc/concept/audio-jitter/comfort/corpus.json b/doc/concept/audio-jitter/comfort/corpus.json new file mode 100644 index 00000000000..0819cb90a0f --- /dev/null +++ b/doc/concept/audio-jitter/comfort/corpus.json @@ -0,0 +1,62 @@ +[ + { + "name": "quietest", + "about": "The quietest whole block sets the level, not the last or the mean", + "rate": 1000, + "channels": 1, + "steps": [ + { "play": [0.5], "frames": 40 }, + { "play": [0.0002], "frames": 20 }, + { "play": [0.3], "frames": 20 }, + { "fill": 0.0002 } + ] + }, + { + "name": "straddle", + "about": "Blocks are counted from the first frame played, so a quiet run that straddles two blocks is diluted", + "rate": 1000, + "channels": 1, + "steps": [ + { "play": [0.0002], "frames": 15 }, + { "play": [0.0004], "frames": 25 }, + { "fill": 0.00026457513110645906 } + ] + }, + { + "name": "stereo", + "about": "A block's mean square spans every channel", + "rate": 1000, + "channels": 2, + "steps": [{ "play": [0.0002, 0.0004], "frames": 20 }, { "fill": 0.00031622776601683794 }] + }, + { + "name": "cap", + "about": "Audio that never falls quiet fills a silence at -60 dBFS, not at its own level", + "rate": 1000, + "channels": 1, + "steps": [{ "play": [0.5], "frames": 40 }, { "fill": 0.001 }] + }, + { + "name": "silent", + "about": "A source that was already silent gets no noise", + "rate": 1000, + "channels": 1, + "steps": [{ "play": [0], "frames": 40 }, { "fill": 0 }] + }, + { + "name": "remeasure", + "about": "Too little audio to fill a block keeps the previous level, and each silence measures afresh from the audio after the last one", + "rate": 1000, + "channels": 1, + "steps": [ + { "play": [0.0002], "frames": 19 }, + { "fill": 0 }, + { "play": [0.0003], "frames": 20 }, + { "fill": 0.0003 }, + { "play": [0.5], "frames": 10 }, + { "fill": 0.0003 }, + { "play": [0.0005], "frames": 20 }, + { "fill": 0.0005 } + ] + } +] diff --git a/js/watch/src/audio/comfort.test.ts b/js/watch/src/audio/comfort.test.ts new file mode 100644 index 00000000000..550da2c89e9 --- /dev/null +++ b/js/watch/src/audio/comfort.test.ts @@ -0,0 +1,33 @@ +import { expect, it } from "bun:test"; +import { readFileSync } from "node:fs"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { ComfortNoise } from "./comfort"; + +type Step = { play: number[]; frames: number } | { fill: number }; +type Case = { name: string; rate: number; channels: number; steps: Step[] }; + +// Shared with moq-audio, so neither implementation is graded against the other. +const corpus = join( + dirname(fileURLToPath(import.meta.url)), + "../../../../doc/concept/audio-jitter/comfort/corpus.json", +); +const cases = JSON.parse(readFileSync(corpus, "utf8")) as Case[]; + +for (const { name, rate, channels, steps } of cases) { + it(`fills a silence at the corpus level: ${name}`, () => { + const comfort = new ComfortNoise(rate); + for (const step of steps) { + if ("play" in step) { + const output = step.play.map((value) => new Float32Array(step.frames).fill(value)); + comfort.played(output, 0, step.frames); + } else { + comfort.fill( + Array.from({ length: channels }, () => new Float32Array(2)), + 0, + ); + expect(Math.abs(comfort.level - step.fill)).toBeLessThan(1e-9); + } + } + }); +} diff --git a/js/watch/src/audio/comfort.ts b/js/watch/src/audio/comfort.ts index 55bf76b9882..1e2fadd174c 100644 --- a/js/watch/src/audio/comfort.ts +++ b/js/watch/src/audio/comfort.ts @@ -24,6 +24,11 @@ export class ComfortNoise { this.#block = Math.max(1, Math.round(rate / 50)); } + /** The RMS of the noise the current or last silence was filled with. */ + get level(): number { + return this.#level; + } + /** Measure `output` from `from` up to `to`, just played. */ played(output: Float32Array[], from: number, to: number): void { if (to <= from) return; diff --git a/quest/m1/push-to-talk/README.md b/quest/m1/push-to-talk/README.md index 848f10e8500..1ff239eec43 100644 --- a/quest/m1/push-to-talk/README.md +++ b/quest/m1/push-to-talk/README.md @@ -93,10 +93,6 @@ in [Container reads drop backfill](/quest/m1/read-backfill.md). The runner fails on a head that is received but never decoded, so the proof catches this. A complete set of runs finishes this epic. -## Required - -- [Native playback fills declared silence with comfort noise](/quest/m1/push-to-talk/native-comfort-noise.md) - `moq play` plays comfort noise over declared silence and counts no underrun, matching the browser - ## Related - [Container reads drop backfill](/quest/m1/read-backfill.md) - changes how the ordered read treats a group below its cursor, where this head was lost diff --git a/quest/m1/push-to-talk/native-comfort-noise.md b/quest/m1/push-to-talk/native-comfort-noise.md deleted file mode 100644 index 24243bff0b8..00000000000 --- a/quest/m1/push-to-talk/native-comfort-noise.md +++ /dev/null @@ -1,31 +0,0 @@ -# [M] Native playback fills declared silence with comfort noise - -## Goal - -`rs/moq-audio` playback, as used by `moq play`, treats a publisher's declared -silence the way `@moq/watch` does: when the ring runs dry at a declared end it -plays level-matched comfort noise instead of plain silence, counts no underrun -or stall, and keeps that silence through the next utterance's refill. A -push-to-talk listener on `moq play` then matches the browser. - -## Plan - -Planned 2026-11-05 (maintainer, /quest-plan of the follow-ups from the -2026-11-05 /quest-complete run), from #7942, whose agent left native playback -with the estimator exception but no comfort noise because its ring has no -notion of a declared end. Milestone: ✅ m1. - -Follow the browser's choices in #7942 unless measurement argues otherwise: the -decoder hands the ring each declared end less the Opus pre-skip, and comfort -noise is white noise matched to the quietest 20 ms block played, capped at --60 dBFS (`js/watch/src/audio/comfort.ts`). Share the level rule with the -browser through `doc/concept/audio-jitter.md` and a corpus case rather than -two descriptions. Cover it in `test/audio-quality`'s native rows or -`just test push-to-talk` if that can drive `moq play`. - -Build on [moq play reaches zero steady-state underruns by -time-stretching](/quest/m1/audio-jitter-target/native-stretch.md), whose -stretch in the same playback ring is on `main` since -[moq-dev/moq#7856](https://github.com/moq-dev/moq/pull/7856). - -Public API: none expected. Wire: none. diff --git a/rs/moq-audio/MIGRATE.md b/rs/moq-audio/MIGRATE.md index 472f955f863..94376708a9f 100644 --- a/rs/moq-audio/MIGRATE.md +++ b/rs/moq-audio/MIGRATE.md @@ -4,6 +4,16 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Playback fills a declared silence with comfort noise + +`Sink::end()` declares that the audio written so far is followed by a silence +its publisher announced, such as a push-to-talk mute: once it has played, the +sink plays comfort noise rather than running dry, and the `Observer` counts no +gap. `Sink::next()` starts the next utterance with the next write, one latency +out, as a new sink's first write would. A player that opened a new sink at each +declared end can keep the one it has, calling `end()` at the marker and `next()` +before the first frame after it. Existing callers need no change. + ### Playback follows a changing latency with received-audio stretch `Sink::set_latency(duration)?` updates the device callback's playout target diff --git a/rs/moq-audio/src/comfort.rs b/rs/moq-audio/src/comfort.rs new file mode 100644 index 00000000000..f049b1be7dd --- /dev/null +++ b/rs/moq-audio/src/comfort.rs @@ -0,0 +1,144 @@ +//! Comfort noise for a silence the publisher declared, by the rule in +//! `doc/concept/audio-jitter.md`, which `@moq/watch` follows too. + +/// The loudest comfort noise, in RMS: -60 dBFS. Audio that never falls quiet, +/// such as music, would otherwise fill a pause with noise as loud as itself. +const MAX_LEVEL: f64 = 0.001; + +/// White noise at the level of the quietest stretch played before a declared +/// silence, so a muted talker's background continues instead of the line going +/// dead. The publisher sends nothing while muted, so the level is measured from +/// what this reader played: the quietest 20 ms block since playback last +/// resumed. A source that was already silent there, such as a noise-suppressed +/// microphone, gets none. +/// +/// Allocates nothing, so it runs on the audio thread. +pub(crate) struct Comfort { + channels: usize, + /// Frames in a block. + block: usize, + /// The block being measured: summed squares across channels, and the frames in it so far. + sum: f64, + frames: usize, + /// The mean square of the quietest whole block since playback last resumed. + quietest: f64, + /// The RMS of the noise for the current silence. + level: f64, + filling: bool, + /// xorshift32 state, never zero. + seed: u32, +} + +impl Comfort { + pub(crate) fn new(rate: u32, channels: usize) -> Self { + Self { + channels, + block: ((rate as f64 / 50.0).round() as usize).max(1), + sum: 0.0, + frames: 0, + quietest: f64::INFINITY, + level: 0.0, + filling: false, + seed: 0x9e37_79b9, + } + } + + /// Measure interleaved `output`, just played. + pub(crate) fn played(&mut self, output: &[f32]) { + if output.len() < self.channels { + return; + } + if std::mem::take(&mut self.filling) { + self.quietest = f64::INFINITY; + self.sum = 0.0; + self.frames = 0; + } + for frame in output.chunks_exact(self.channels) { + for &sample in frame { + let sample = sample as f64; + self.sum += sample * sample; + } + self.frames += 1; + if self.frames < self.block { + continue; + } + self.quietest = self.quietest.min(self.sum / (self.block * self.channels) as f64); + self.sum = 0.0; + self.frames = 0; + } + } + + /// Fill interleaved `output` with comfort noise. + pub(crate) fn fill(&mut self, output: &mut [f32]) { + if !std::mem::replace(&mut self.filling, true) && self.quietest.is_finite() { + // A silence after too little audio to measure keeps the previous level. + self.level = self.quietest.sqrt().min(MAX_LEVEL); + } + // Uniform noise on [-a, a] has an RMS of a / sqrt(3). + let amplitude = (self.level * 3f64.sqrt()) as f32; + for sample in output { + self.seed ^= self.seed << 13; + self.seed ^= self.seed >> 17; + self.seed ^= self.seed << 5; + *sample = (self.seed as f32 / u32::MAX as f32 * 2.0 - 1.0) * amplitude; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(serde::Deserialize)] + struct Case { + name: String, + rate: u32, + channels: usize, + steps: Vec, + } + + #[derive(serde::Deserialize)] + #[serde(untagged)] + enum Step { + Play { play: Vec, frames: usize }, + Fill { fill: f64 }, + } + + /// Each case plays constant runs and checks the level the next silence fills at. + #[test] + fn the_level_conforms_to_the_corpus() { + let cases: Vec = + serde_json::from_str(include_str!("../../../doc/concept/audio-jitter/comfort/corpus.json")).unwrap(); + for case in cases { + let mut comfort = Comfort::new(case.rate, case.channels); + for (index, step) in case.steps.iter().enumerate() { + match step { + Step::Play { play, frames } => comfort.played(&play.repeat(*frames)), + Step::Fill { fill } => { + let mut noise = vec![0.0; 2 * case.channels]; + comfort.fill(&mut noise); + assert!( + (comfort.level - fill).abs() < 1e-9, + "{} step {index}: level {}, want {fill}", + case.name, + comfort.level + ); + } + } + } + } + } + + /// The noise is white and uniform, so its RMS is the level and nothing exceeds a / sqrt(3). + #[test] + fn the_noise_has_the_level_it_was_measured_at() { + let mut comfort = Comfort::new(48_000, 2); + comfort.played(&[0.0002; 960 * 2]); + let mut noise = vec![0.0; 48_000 * 2]; + comfort.fill(&mut noise); + let peak = noise.iter().fold(0.0f32, |peak, sample| peak.max(sample.abs())); + let rms = (noise.iter().map(|&sample| sample as f64 * sample as f64).sum::() / noise.len() as f64).sqrt(); + assert!(peak <= 0.0002 * 3f32.sqrt() * 1.0001, "peak {peak}"); + assert!((rms - 0.0002).abs() < 0.000_005, "rms {rms}"); + } +} diff --git a/rs/moq-audio/src/lib.rs b/rs/moq-audio/src/lib.rs index 147d16353d1..7547e4ca3a6 100644 --- a/rs/moq-audio/src/lib.rs +++ b/rs/moq-audio/src/lib.rs @@ -56,6 +56,8 @@ mod allocation; #[cfg(feature = "aac")] mod aac; mod activity; +#[cfg(any(feature = "bus", feature = "playback"))] +mod comfort; mod error; mod format; mod frame; diff --git a/rs/moq-audio/src/playback/mixer.rs b/rs/moq-audio/src/playback/mixer.rs index e2367a1b9cd..25fb1e5f62e 100644 --- a/rs/moq-audio/src/playback/mixer.rs +++ b/rs/moq-audio/src/playback/mixer.rs @@ -335,7 +335,8 @@ impl Mixer { let read = match status { Read::NotReady => None, Read::Underflow { frames: read } => Some(read), - Read::Ok | Read::Trimmed { .. } => Some(frames), + // A declared silence is played on purpose, so it is no gap. + Read::Ok | Read::Trimmed { .. } | Read::Comfort { .. } => Some(frames), }; match read { Some(read) if entry.completion.writing() => { diff --git a/rs/moq-audio/src/playback/sink.rs b/rs/moq-audio/src/playback/sink.rs index 8797fa4b0ef..87b8b2906bf 100644 --- a/rs/moq-audio/src/playback/sink.rs +++ b/rs/moq-audio/src/playback/sink.rs @@ -218,6 +218,28 @@ impl Sink { Ok(Write::from_accepted(requested_sample_frames, accepted_sample_frames)) } + /// Declare that the audio written so far is followed by a silence its + /// publisher announced, such as a push-to-talk mute. + /// + /// Once that audio has played, the sink plays comfort noise at the level of + /// the quietest 20 ms it played, at most -60 dBFS, rather than running dry, + /// and the [`Observer`](super::Observer) counts no gap. That lasts until the + /// utterance [`next`](Self::next) starts plays. Writes before then, such as a + /// codec's final frames, continue the audio before the silence. + pub fn end(&mut self) { + self.channel.lock().unwrap().prod.end(); + } + + /// Start the next utterance after a declared silence with the next + /// [`write`](Self::write). + /// + /// It plays one [`latency`](Input::latency) after that write, behind + /// whatever is still queued, with comfort noise filling the wait, the way a + /// new sink's first write plays behind its opening silence. + pub fn next(&mut self) { + self.channel.lock().unwrap().prod.next(); + } + /// How much audio is queued between the last [`write`](Self::write) and the /// speaker. /// @@ -439,7 +461,8 @@ impl Registration { let mut current = self.channel.lock().unwrap(); let mut input = self.input.clone(); input.latency = current.prod.latency(); - let (channel, cons) = channel(&input, rate, bus); + let (mut channel, cons) = channel(&input, rate, bus); + channel.prod.carry(¤t.prod); *current = channel; self.pending = Some(cons); } @@ -648,6 +671,82 @@ mod tests { assert!(gap.at <= stall.at && stall.at - gap.at < buffer, "{observation:?}"); } + /// A declared silence plays comfort noise at the level heard before it, through + /// the wait before the next utterance, which then plays one latency after its + /// first write. None of it is a gap or a stall. + #[test] + fn a_declared_silence_plays_comfort_noise_until_the_next_utterance() { + let shared = Arc::new(Shared::default()); + let mut observer = super::super::Observer::new(shared.observe.clone()).unwrap(); + let engine = Arc::new(super::super::Handle { + commands: super::super::driver::Commands::default(), + }); + let latency = Duration::from_millis(40); + let input = Input { + latency, + ..Default::default() + }; + let (mut sink, mut registration) = new(0, 48_000, Layout::Stereo, input, shared.clone(), engine).unwrap(); + let (commands, rx) = std::sync::mpsc::sync_channel(8); + let (retired, _rx) = std::sync::mpsc::sync_channel(8); + let mut mixer = mixer::Mixer::new(rx, retired, 48_000, Layout::Stereo, shared.observe.clone()).unwrap(); + registration.attach(&commands); + let pcm = |sample: f32, frames: usize| -> Vec { + std::iter::repeat_n(sample.to_le_bytes(), frames * 2) + .flatten() + .collect() + }; + let rms = + |samples: &[f32]| (samples.iter().map(|&s| s as f64 * s as f64).sum::() / samples.len() as f64).sqrt(); + let mut out = [0.0; 960]; + mixer.fill(&mut out, Duration::ZERO); + + // 100 ms of a quiet background, then a declared end. + let quiet = 0.0002; + let _ = sink.write(&pcm(quiet, 4_800)).unwrap(); + sink.end(); + let mut heard = Vec::new(); + for _ in 0..150 { + mixer.fill(&mut out, Duration::ZERO); + heard.extend_from_slice(&out); + } + // The primed silence, the background, then a second of noise at its level. + let noise = &heard[heard.len() - 48_000 * 2..]; + assert!(noise.iter().all(|s| s.abs() <= quiet * 3f32.sqrt() * 1.0001)); + assert!((rms(noise) - quiet as f64).abs() < 0.00001, "rms {}", rms(noise)); + + // The next utterance plays one latency after its first write, with noise until then. + sink.next(); + let _ = sink.write(&pcm(0.5, 960)).unwrap(); + let mut until = 0; + loop { + mixer.fill(&mut out, Duration::ZERO); + if let Some(at) = out.iter().position(|s| *s == 0.5) { + until += at / 2; + break; + } + assert!(out.iter().all(|s| s.abs() <= quiet * 3f32.sqrt() * 1.0001 && *s != 0.0)); + until += out.len() / 2; + } + assert_eq!(until, 48_000 * 40 / 1000); + for _ in 0..10 { + let _ = sink.write(&pcm(0.5, 480)).unwrap(); + mixer.fill(&mut out, Duration::ZERO); + } + + let observation = observer.read().unwrap(); + assert!(observation.gaps.is_empty(), "{observation:?}"); + assert!(!observation.stalled && observation.stalls.len() == 1, "{observation:?}"); + + // Once the next utterance plays, running dry is a gap again. + for _ in 0..20 { + mixer.fill(&mut out, Duration::ZERO); + } + let _ = sink.write(&pcm(0.5, 2_400)).unwrap(); + mixer.fill(&mut out, Duration::ZERO); + assert_eq!(observer.read().unwrap().gaps.len(), 1); + } + /// The resampler only converts whole blocks and its filter lags the input, so /// a track whose rate differs from the device's must have its tail pushed /// through on finish: both a partial final block and one that ends exactly on diff --git a/rs/moq-audio/src/ring.rs b/rs/moq-audio/src/ring.rs index 78b8779cf9b..c8b48617615 100644 --- a/rs/moq-audio/src/ring.rs +++ b/rs/moq-audio/src/ring.rs @@ -6,6 +6,8 @@ //! writer falls behind, and can trim back down to it when the writer runs ahead. //! Native playback instead enables received-audio stretch; only an actual //! underrun refills with emergency silence, which never enters the pitch search. +//! A silence the writer declared is not an underrun: the reader fills it with +//! comfort noise, and the next utterance plays one latency after it resumes. //! Neither end allocates, locks, or blocks after construction, so either may //! run on an audio callback. @@ -17,7 +19,7 @@ )] use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::time::Duration; use ringbuf::traits::{Consumer as _, Observer as _, Producer as _, Split as _}; @@ -64,6 +66,9 @@ pub(crate) enum Read { NotReady, /// Only the first `frames` were queued; the rest of the output is silence. Underflow { frames: usize }, + /// Only the first `frames` were queued; the rest is a declared silence, + /// filled with comfort noise. + Comfort { frames: usize }, /// The output was filled, then `frames` were discarded to trim the queue. Trimmed { frames: usize }, } @@ -81,8 +86,21 @@ struct Shared { target: AtomicUsize, finished: AtomicBool, stretch: AtomicBool, + /// Declared ends, counted. See [`Producer::end`]. + ends: AtomicU64, + /// The resume the reader has yet to reach, [`PENDING`] set while there is one, + /// with the count of ends reaching it clears in the other bits. See + /// [`Producer::next`]. + resume: AtomicU64, + /// Where that resume starts, in written frames, and the comfort noise it + /// plays there before the next utterance. Written only while none is pending. + resume_at: AtomicUsize, + resume_frames: AtomicUsize, } +/// Set in [`Shared::resume`] while a resume waits for the reader. +const PENDING: u64 = 1 << 63; + /// A queue resampling `input_rate` to `output_rate`, interleaved at `channels`. /// /// # Panics @@ -123,15 +141,22 @@ pub(crate) fn channel(channels: usize, input_rate: u32, output_rate: u32, config latency: config.latency, written: latency, pad: config.pad.then(|| latency * channels / 4), + declared: false, + next: false, shared: shared.clone(), }; let consumer = Consumer { ring: cons, channels, stretch: None, + staged_prime: false, refill: false, prime: latency, consumed: 0, + position: 0, + cleared: 0, + cushioned: 0, + comfort: crate::comfort::Comfort::new(output_rate, channels), trim: config.trim, shared, }; @@ -154,10 +179,77 @@ pub(crate) struct Producer { written: usize, /// Pad back up to the latency at or below this many samples. pad: Option, + /// An end was declared and no next utterance pushed since. + declared: bool, + /// The next push starts the next utterance. See [`next`](Self::next). + next: bool, shared: Arc, } impl Producer { + /// Declare that what was pushed so far is followed by a silence the writer + /// announced. Once it has played, the reader fills with comfort noise rather + /// than underflowing, until the utterance [`next`](Self::next) starts plays. + /// Pushes before then continue the audio before the silence. + pub(crate) fn end(&mut self) { + self.declared = true; + self.shared.ends.fetch_add(1, Ordering::Release); + } + + /// Start the next utterance with the next push. It plays one latency after + /// that push, behind whatever is still queued, with comfort noise filling the + /// wait, as a fresh queue's first push would behind its primed silence. + pub(crate) fn next(&mut self) { + self.next = true; + } + + /// Carry a declared silence over from the queue this one replaces. + pub(crate) fn carry(&mut self, from: &Producer) { + if from.declared { + self.end(); + } + self.next = from.next; + } + + /// Queue the comfort noise that starts the next utterance one latency out, + /// once the reader reaches what is queued now. + fn resume(&mut self) { + // The resampler's tail belongs to the audio before the silence. + if let Some(convert) = &mut self.convert { + let before = convert.emitted; + convert.flush(&mut self.ring); + self.written = self.written.wrapping_add((convert.emitted - before) as usize); + } + self.declared = false; + let queued = self.written.wrapping_sub(self.shared.consumed.load(Ordering::Relaxed)); + let frames = self.shared.target.load(Ordering::Relaxed).saturating_sub(queued); + let ends = self.shared.ends.load(Ordering::Relaxed); + let mut state = self.shared.resume.load(Ordering::Acquire); + while state & PENDING != 0 { + // The reader has yet to reach the previous resume, so two declared + // silences fall within one latency. That resume clears this end too, + // and it already topped the queue up to the latency, so only a target + // raised since leaves anything to wait, in plain silence: a resume's + // place is published only while none is pending. + match self + .shared + .resume + .compare_exchange(state, PENDING | ends, Ordering::AcqRel, Ordering::Acquire) + { + Ok(_) => { + let samples = self.ring.push_iter(std::iter::repeat_n(0.0, frames * self.channels)); + self.written = self.written.wrapping_add(samples / self.channels); + return; + } + Err(current) => state = current, + } + } + self.shared.resume_at.store(self.written, Ordering::Relaxed); + self.shared.resume_frames.store(frames, Ordering::Relaxed); + self.shared.resume.store(PENDING | ends, Ordering::Release); + self.written = self.written.wrapping_add(frames); + } + /// Queue interleaved input, resampling it on the way in. /// /// Input frames are admitted only while their output has room, so a full @@ -167,6 +259,9 @@ impl Producer { if !self.shared.read.load(Ordering::Relaxed) { return Push::NotReady; } + if input.len() >= self.channels && std::mem::take(&mut self.next) { + self.resume(); + } let frames = input.len() / self.channels; let input = &input[..frames * self.channels]; @@ -350,6 +445,9 @@ impl Convert { return false; } self.input[self.pending * self.channels..].fill(0.0); + // The padding is input too, so a later flush, after the next utterance, + // owes that utterance's tail rather than counting this padding against it. + self.fed += (self.chunk - self.pending) as u64; self.convert(ring); } true @@ -378,12 +476,22 @@ pub(crate) struct Consumer { ring: HeapCons, channels: usize, stretch: Option>, + /// Whether the window staged in the stretcher is startup cushion. + staged_prime: bool, /// An actual underrun waits for received PCM, emitting emergency silence /// directly rather than putting padding through the pitch search. refill: bool, /// The startup cushion is silence, not received audio to stretch. prime: usize, consumed: usize, + /// Frames taken off the ring plus the comfort noise of each resume played + /// out: the writer's `written`, as far as the reader has reached. + position: usize, + /// The count of declared ends the last resume reached clears. + cleared: u64, + /// Comfort noise already played at the pending resume. + cushioned: usize, + comfort: crate::comfort::Comfort, /// Trim only after filling three quarters of the current target's headroom. trim: bool, shared: Arc, @@ -399,22 +507,27 @@ impl Consumer { /// Fill `output` with queued interleaved frames, and silence past them. pub(crate) fn read(&mut self, output: &mut [f32]) -> Read { self.start(); + let whole = output.len() - output.len() % self.channels; + output[whole..].fill(0.0); if !self.shared.pushed.load(Ordering::Relaxed) { + // A silence declared before any audio, as for a listener joining a muted talker. + if self.declared() { + self.comfort.fill(&mut output[..whole]); + return Read::Comfort { frames: 0 }; + } output.fill(0.0); return Read::NotReady; } - let whole = output.len() - output.len() % self.channels; - let read = if self.stretch.is_some() { - self.read_stretched(&mut output[..whole]) - } else { - let samples = self.ring.pop_slice(&mut output[..whole]); - self.consumed = self.consumed.wrapping_add(samples / self.channels); - self.shared.consumed.store(self.consumed, Ordering::Relaxed); - samples - }; - output[read..].fill(0.0); + let read = self.fill(&mut output[..whole]); if read < whole { + if self.declared() { + self.comfort.fill(&mut output[read..whole]); + return Read::Comfort { + frames: read / self.channels, + }; + } + output[read..whole].fill(0.0); self.refill = self.stretch.is_some(); return Read::Underflow { frames: read / self.channels, @@ -436,41 +549,111 @@ impl Consumer { Read::Ok } - fn read_stretched(&mut self, output: &mut [f32]) -> usize { + /// Whether running dry now is a silence the writer declared: an end it has + /// not resumed from, on a queue that is not playing out its last samples. + fn declared(&self) -> bool { + self.shared.ends.load(Ordering::Acquire) != self.cleared && !self.shared.finished.load(Ordering::Relaxed) + } + + /// The pending resume, as the frames queued ahead of it and its comfort noise + /// still to play. + /// + /// Read the ring's length before calling this: the writer publishes a resume + /// before pushing what follows it, so a length that includes those frames + /// guarantees the resume is visible too. + fn resume(&self) -> Option<(usize, usize)> { + if self.shared.resume.load(Ordering::Acquire) & PENDING == 0 { + return None; + } + let at = self.shared.resume_at.load(Ordering::Relaxed); + let frames = self.shared.resume_frames.load(Ordering::Relaxed); + Some((at.wrapping_sub(self.position), frames - self.cushioned)) + } + + /// Play up to `frames` of the pending resume's comfort noise into `output`, + /// or skip them when it is `None`, and clear its ends once it is done. + fn cushion(&mut self, output: Option<&mut [f32]>, frames: usize) -> usize { + let left = self.shared.resume_frames.load(Ordering::Relaxed) - self.cushioned; + let frames = frames.min(left); + if let Some(output) = output { + self.comfort.fill(&mut output[..frames * self.channels]); + } + self.cushioned += frames; + self.consumed = self.consumed.wrapping_add(frames); + if frames == left { + self.position = self.position.wrapping_add(self.cushioned); + self.cushioned = 0; + self.cleared = self.shared.resume.swap(0, Ordering::AcqRel) & !PENDING; + } + frames + } + + /// Fill `output` from the queue, returning the samples written. + fn fill(&mut self, output: &mut [f32]) -> usize { if self.refill && !self.shared.finished.load(Ordering::Relaxed) { - if self.available() < self.shared.target.load(Ordering::Relaxed) { + // Nothing more comes before a declared silence, so its tail plays now. + if !self.declared() && self.available() < self.shared.target.load(Ordering::Relaxed) { return 0; } self.refill = false; } let channels = self.channels; - let stretch = self.stretch.as_mut().unwrap(); let mut done = 0; while done < output.len() { - if stretch.available() == 0 { - let buffered = self.ring.occupied_len() / channels; - let frames = buffered.min(stretch.window); + if let Some(stretch) = self.stretch.as_mut().filter(|stretch| stretch.available() > 0) { + let frames = stretch.read(&mut output[done..]); + if !self.staged_prime { + self.comfort.played(&output[done..done + frames * channels]); + } + done += frames * channels; + self.consumed = self.consumed.wrapping_add(frames); + continue; + } + + let buffered = self.ring.occupied_len() / channels; + let mut frames = buffered; + if let Some((ahead, _)) = self.resume() { + if ahead == 0 { + let room = (output.len() - done) / channels; + done += self.cushion(Some(&mut output[done..]), room) * channels; + continue; + } + frames = frames.min(ahead); + } + let Some(stretch) = self.stretch.as_mut() else { + let frames = frames.min((output.len() - done) / channels); if frames == 0 { break; } - // Never transform startup padding or a finite stream's tail. - let enabled = self.prime == 0 && !self.shared.finished.load(Ordering::Relaxed); - let frames = if self.prime > 0 { frames.min(self.prime) } else { frames }; - self.prime = self.prime.saturating_sub(frames); - let read = self.ring.pop_slice(&mut stretch.input[..frames * channels]); + let read = self.ring.pop_slice(&mut output[done..done + frames * channels]); debug_assert_eq!(read, frames * channels); - stretch.process(frames, buffered, self.shared.target.load(Ordering::Relaxed), enabled); - // The writer still counts the input while it is being analysed. - // Publish only the resulting duration change, then the rendered - // frames. Moving PCM into staging never makes it disappear. - self.consumed = self.consumed.wrapping_add(frames).wrapping_sub(stretch.available()); - self.shared.consumed.store(self.consumed, Ordering::Relaxed); + let prime = frames.min(self.prime); + self.prime -= prime; + self.comfort.played(&output[done + prime * channels..done + read]); + self.position = self.position.wrapping_add(frames); + self.consumed = self.consumed.wrapping_add(frames); + done += read; + continue; + }; + let frames = frames.min(stretch.window); + if frames == 0 { + break; } - let frames = stretch.read(&mut output[done..]); - done += frames * channels; - self.consumed = self.consumed.wrapping_add(frames); - self.shared.consumed.store(self.consumed, Ordering::Relaxed); + // Never transform startup padding or a finite stream's tail. + let enabled = self.prime == 0 && !self.shared.finished.load(Ordering::Relaxed); + let frames = if self.prime > 0 { frames.min(self.prime) } else { frames }; + self.staged_prime = self.prime > 0; + self.prime -= frames.min(self.prime); + let read = self.ring.pop_slice(&mut stretch.input[..frames * channels]); + debug_assert_eq!(read, frames * channels); + self.position = self.position.wrapping_add(frames); + stretch.process(frames, buffered, self.shared.target.load(Ordering::Relaxed), enabled); + // The writer still counts the input while it is being analysed. + // Publish only the resulting duration change, then the rendered + // frames. Moving PCM into staging never makes it disappear. + self.consumed = self.consumed.wrapping_add(frames).wrapping_sub(stretch.available()); } + self.shared.consumed.store(self.consumed, Ordering::Relaxed); done } @@ -481,19 +664,37 @@ impl Consumer { /// Frames ready to read, or zero before the writer's first push. pub(crate) fn available(&self) -> usize { - if self.shared.pushed.load(Ordering::Relaxed) { - self.ring.occupied_len() / self.channels + self.stretch.as_ref().map_or(0, |stretch| stretch.available()) - } else { - 0 + if !self.shared.pushed.load(Ordering::Relaxed) { + return 0; } + let buffered = self.ring.occupied_len() / self.channels; + let cushion = self.resume().map_or(0, |(_, frames)| frames); + buffered + cushion + self.stretch.as_ref().map_or(0, |stretch| stretch.available()) } /// Drop up to `frames` queued frames unread. pub(crate) fn discard(&mut self, frames: usize) { let staged = self.stretch.as_mut().map_or(0, |stretch| stretch.discard(frames)); - let skipped = self.ring.skip((frames - staged) * self.channels) / self.channels; - self.prime = self.prime.saturating_sub(skipped); - self.consumed = self.consumed.wrapping_add(staged + skipped); + self.consumed = self.consumed.wrapping_add(staged); + let mut left = frames - staged; + while left > 0 { + let mut take = left; + if let Some((ahead, _)) = self.resume() { + if ahead == 0 { + left -= self.cushion(None, left); + continue; + } + take = take.min(ahead); + } + let skipped = self.ring.skip(take * self.channels) / self.channels; + if skipped == 0 { + break; + } + self.prime = self.prime.saturating_sub(skipped); + self.position = self.position.wrapping_add(skipped); + self.consumed = self.consumed.wrapping_add(skipped); + left -= skipped; + } self.shared.consumed.store(self.consumed, Ordering::Relaxed); } } @@ -858,4 +1059,155 @@ mod tests { Duration::from_secs_f64(cons.available() as f64 / 48_000.0) ); } + + /// The plain queue's 50 ms, in frames. + const LATENCY: usize = 2400; + + /// A queue that has played its primed silence and all but `tail` frames of a + /// quiet background, then declared an end. + fn ended(tail: usize) -> (Producer, Consumer) { + let (mut prod, mut cons) = queue(48_000, 48_000, Duration::from_secs(1)); + assert_eq!(prod.push(&[0.0002; 4800 * STEREO]), Push::Ok); + let mut out = vec![0.0; (LATENCY + 4800 - tail) * STEREO]; + assert_eq!(cons.read(&mut out), Read::Ok); + prod.end(); + (prod, cons) + } + + fn noise(samples: &[f32]) -> bool { + samples + .iter() + .all(|&sample| sample != 0.0 && sample.abs() <= 0.0002 * 3f32.sqrt() * 1.0001) + } + + /// A declared end plays comfort noise once the queue runs dry, and the next + /// utterance plays one latency after its first push. + #[test] + fn a_declared_silence_is_comfort_noise_until_the_next_utterance() { + let (mut prod, mut cons) = ended(0); + let mut out = vec![0.0; 1000 * STEREO]; + assert_eq!(cons.read(&mut out), Read::Comfort { frames: 0 }); + assert!(noise(&out)); + + prod.next(); + assert_eq!(prod.push(&[0.5; 960 * STEREO]), Push::Ok); + let mut out = vec![0.0; (LATENCY + 960) * STEREO]; + assert_eq!(cons.read(&mut out), Read::Ok); + assert!(noise(&out[..LATENCY * STEREO])); + assert!(out[LATENCY * STEREO..].iter().all(|&sample| sample == 0.5)); + // Running dry after the next utterance is an underflow again. + assert_eq!(cons.read(&mut out), Read::Underflow { frames: 0 }); + } + + /// A next utterance pushed while the end's tail still plays waits behind it, + /// with comfort noise topping the wait up to one latency. + #[test] + fn a_resume_behind_a_queued_tail_starts_one_latency_out() { + let (mut prod, mut cons) = ended(1200); + prod.next(); + assert_eq!(prod.push(&[0.5; 960 * STEREO]), Push::Ok); + assert_eq!(prod.buffered(), Duration::from_millis(70)); + let mut out = vec![0.0; (LATENCY + 960) * STEREO]; + assert_eq!(cons.read(&mut out), Read::Ok); + assert!(out[..1200 * STEREO].iter().all(|&sample| sample == 0.0002)); + assert!(noise(&out[1200 * STEREO..LATENCY * STEREO])); + assert!(out[LATENCY * STEREO..].iter().all(|&sample| sample == 0.5)); + } + + /// A second silence declared before the reader reaches the first resume is + /// cleared by it: the first already topped the queue up to the latency. + #[test] + fn a_second_silence_within_one_latency_clears_with_the_first() { + let (mut prod, mut cons) = ended(600); + prod.next(); + assert_eq!(prod.push(&[0.5; 240 * STEREO]), Push::Ok); + prod.end(); + prod.next(); + assert_eq!(prod.push(&[0.25; 240 * STEREO]), Push::Ok); + let mut out = vec![0.0; (LATENCY + 480) * STEREO]; + assert_eq!(cons.read(&mut out), Read::Ok); + assert!(out[(LATENCY + 240) * STEREO..].iter().all(|&sample| sample == 0.25)); + assert_eq!(cons.read(&mut out), Read::Underflow { frames: 0 }); + } + + /// An end before any audio, such as a listener joining a muted talker, is a + /// declared silence already, with nothing measured to fill it with. + #[test] + fn an_end_before_any_audio_is_a_silence() { + let (mut prod, mut cons) = queue(48_000, 48_000, Duration::from_secs(1)); + prod.end(); + let mut out = vec![1.0; 128 * STEREO]; + assert_eq!(cons.read(&mut out), Read::Comfort { frames: 0 }); + assert!(out.iter().all(|&sample| sample == 0.0)); + prod.next(); + assert_eq!(prod.push(&[0.5; 960 * STEREO]), Push::Ok); + let mut out = vec![0.0; (LATENCY + 960) * STEREO]; + assert_eq!(cons.read(&mut out), Read::Ok); + assert!(out[LATENCY * STEREO..].iter().all(|&sample| sample == 0.5)); + assert_eq!(cons.read(&mut out), Read::Underflow { frames: 0 }); + } + + /// A trim across a resume skips its comfort noise like any other queued audio. + #[test] + fn a_discard_crosses_a_resume() { + let (mut prod, mut cons) = ended(1200); + prod.next(); + assert_eq!(prod.push(&[0.5; 960 * STEREO]), Push::Ok); + cons.discard(1800); + assert_eq!(cons.available(), 600 + 960); + let mut out = vec![0.0; (600 + 960) * STEREO]; + assert_eq!(cons.read(&mut out), Read::Ok); + assert!(noise(&out[..600 * STEREO])); + assert!(out[600 * STEREO..].iter().all(|&sample| sample == 0.5)); + assert_eq!(cons.read(&mut out), Read::Underflow { frames: 0 }); + } + + /// Each resume flushes the resampler, and a later flush still owes the next + /// utterance's whole tail rather than counting the earlier padding against it. + #[test] + fn every_utterance_through_the_resampler_is_heard() { + let (mut prod, mut cons) = queue(44_100, 48_000, Duration::from_secs(1)); + let mut out = vec![0.0; 48_000 * STEREO]; + // The first flush pads nearly a whole chunk, more than the second leaves pending. + for (sample, frames) in [(0.25f32, CHUNK + 1), (0.5, CHUNK + 100)] { + prod.end(); + prod.next(); + assert_eq!(prod.push(&vec![sample; frames * STEREO]), Push::Ok); + } + assert!(prod.flush()); + let _ = cons.read(&mut out); + // Past half amplitude, where the resampled edges cross. + let heard = out.iter().step_by(STEREO).filter(|&&sample| sample > 0.375).count(); + let want = (CHUNK + 100) * 48_000 / 44_100; + assert!(heard.abs_diff(want) <= 1, "heard {heard}, want {want}"); + } + + /// A tail stuck short of an emergency refill when its end is declared plays + /// out, rather than waiting through the silence for audio that is not coming. + #[test] + fn a_declared_end_releases_an_emergency_refill() { + let (mut prod, mut cons) = channel( + STEREO, + 48_000, + 48_000, + Config { + latency: Duration::from_millis(50), + capacity: Duration::from_secs(1), + pad: true, + trim: false, + }, + ); + cons.stretch(48_000); + cons.start(); + assert_eq!(prod.push(&[]), Push::Ok); + let mut out = vec![0.0; LATENCY * STEREO]; + assert_eq!(cons.read(&mut out), Read::Ok); + assert_eq!(cons.read(&mut out[..128 * STEREO]), Read::Underflow { frames: 0 }); + assert_eq!(prod.push(&[0.25; 960 * STEREO]), Push::Ok); + assert_eq!(cons.read(&mut out[..128 * STEREO]), Read::Underflow { frames: 0 }); + prod.end(); + let mut out = vec![0.0; 1000 * STEREO]; + assert_eq!(cons.read(&mut out), Read::Comfort { frames: 960 }); + assert!(out[..960 * STEREO].iter().all(|&sample| sample == 0.25)); + } } diff --git a/rs/moq-cli/src/play/fake.rs b/rs/moq-cli/src/play/fake.rs index 75deca889b4..e0707a582a6 100644 --- a/rs/moq-cli/src/play/fake.rs +++ b/rs/moq-cli/src/play/fake.rs @@ -16,6 +16,8 @@ use super::window::Event; pub(super) struct Played { /// Which sink it went through, in the order they were opened. pub sink: usize, + /// Which utterance on that sink: each [`Sink::next`] starts another. + pub utterance: usize, /// Its first sample, which tells a test's renditions apart. pub sample: f32, /// When it starts sounding. @@ -117,6 +119,9 @@ impl Speaker for Recorder { sample_rate: input.sample_rate, latency: input.latency, refill: false, + declared: false, + next: false, + utterance: 0, prime: Some(input.latency), pending: Vec::new(), queued: Duration::ZERO, @@ -135,6 +140,11 @@ pub(super) struct FakeSink { sample_rate: u32, latency: Duration, refill: bool, + /// Running dry is a declared silence rather than an underrun. + declared: bool, + /// The next write starts the next utterance one latency out. + next: bool, + utterance: usize, prime: Option, pending: Vec<(f32, Duration)>, queued: Duration, @@ -167,8 +177,27 @@ impl Sink for FakeSink { // heard until fresh writes refill the target after a genuine gap. if let Some(prime) = self.prime.take() { self.end = now + prime; + // The opening silence already holds the first utterance one latency out. + if std::mem::take(&mut self.next) { + self.declared = false; + self.utterance += 1; + } + } else if std::mem::take(&mut self.next) { + self.utterance += 1; + // Comfort noise tops the queue up to the latency ahead of the utterance. + if self.refill { + self.resume(); + } + self.declared = false; + let ahead = self.end.saturating_duration_since(now); + self.end = self.end.max(now) + self.latency.saturating_sub(ahead); } else if self.end < now { - self.refill = true; + // Comfort noise filled a declared silence, so this plays as it lands. + if self.declared { + self.end = now; + } else { + self.refill = true; + } } let frames = (samples.len() / self.stride) as u64; @@ -191,6 +220,14 @@ impl Sink for FakeSink { Ok(written) } + fn end(&mut self) { + self.declared = true; + } + + fn next(&mut self) { + self.next = true; + } + fn buffered(&self) -> Duration { if let Some(prime) = self.prime { prime @@ -206,6 +243,7 @@ impl FakeSink { fn play(&mut self, sample: f32, duration: Duration) { self.state.lock().unwrap().played.push(Played { sink: self.id, + utterance: self.utterance, sample, from: self.end, to: self.end + duration, diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 461a0381c71..51564d3cca2 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -541,8 +541,8 @@ async fn play_audio( // and the sink is where it lives: a sample handed over now sounds that much // later. That estimate times each packet when `read` hands it over, so this // receiver never waits on the speaker between reads. Live playback lets - // the callback stretch toward the target; declared ends queue their replacements - // until the old sink drains. The window + // the callback stretch toward the target; a declared end plays comfort noise + // until the old media drains and its replacement is due. The window // schedules video against where the speaker actually is, which keeps the two // together whatever the target does. let role = presentation.track(Kind::Audio); @@ -671,6 +671,10 @@ async fn play_audio( let mut ended = false; let mut deferred = None; let mut sequential = false; + // The sink was told its audio ends in a declared silence and has not been told what follows. + let mut silent = false; + // The frame being written resumes a timeline whose end is already declared. + let mut declare = false; loop { let (event, delay, crossing, _) = loop { if let Some(event) = received.lock().unwrap().pop_front() { @@ -691,10 +695,16 @@ async fn play_audio( ended = false; role.gap(); sequential = false; + // What follows a declared silence starts the next utterance however many + // groups were lost on the way, rather than filling the silence as a hole. + if std::mem::take(&mut silent) { + timeline = AudioTimeline::default(); + sink.next(); + } } Some(Crossing::Resume(transition)) => { - // A declared boundary preserves the old sink, unlike an undeclared hole. - drain(sink).await; + // A declared boundary preserves the old sink, unlike an undeclared hole: + // its tail and then comfort noise play while the old media drains. let first = lead(delay); let mapping = role.resume(transition, frame.timestamp, Some(first)).ready().await?; role.apply(transition, mapping); @@ -711,16 +721,24 @@ async fn play_audio( { role.end(end)?; ended = true; + declare = true; } sequential = true; timeline = AudioTimeline::default(); - sink = speaker.sink(input.clone())?; + // Comfort noise fills the wait, so this frame plays one target out. + silent = false; + sink.next(); } None => {} } frame } Ok(Some(moq_audio::decode::Read::End { end })) => { + // The stopped encoder's final frames may follow, and play before the silence. + if !silent { + sink.end(); + silent = true; + } if playing && ended { deferred = Some(end); } else { @@ -758,6 +776,7 @@ async fn play_audio( // timeline region the player skipped. if timing.reset_sink { sequential = false; + silent = false; drop(sink); presentation.audio_restarted(); sink = speaker.sink(input.clone())?; @@ -807,6 +826,10 @@ async fn play_audio( } let skip = fit.skip as usize * stride; let _ = sink.write(&frame.data[skip.min(frame.data.len())..])?; + if std::mem::take(&mut declare) { + sink.end(); + silent = true; + } // Anchor the playout clock on where the speaker has actually reached, which // is the only half of the pipeline that cannot skip ahead. A move has to @@ -926,7 +949,8 @@ mod tests { producer.write(packet(0, 0.5)).unwrap(); producer.finish().unwrap(); task.await.unwrap(); - assert_eq!(recorder.latencies(), vec![AUDIO_BUFFER_MIN, delay]); + // The resume keeps the sink, which fills the declared silence with comfort noise. + assert_eq!(recorder.latencies(), vec![AUDIO_BUFFER_MIN]); let played = recorder.played(); let old: Vec<_> = played.iter().filter(|part| part.sample == 0.25).collect(); let new: Vec<_> = played.iter().filter(|part| part.sample == 0.5).collect(); @@ -941,6 +965,96 @@ mod tests { assert!(old.last().unwrap().to <= new.first().unwrap().from); } + /// A push-to-talk mute declares its silence and sends nothing until the next + /// press. The sink carries on through it, so the native ring fills it with + /// comfort noise, and the next press plays one delay after it arrives. + #[tokio::test] + async fn a_mute_keeps_the_sink_and_the_next_press_plays_one_delay_out() { + mute(false).await; + } + + /// A group lost after the mute's marker still leaves the next press its own + /// utterance, rather than a hole filled with two seconds of silence. + #[tokio::test] + async fn a_press_after_a_lost_group_still_plays_one_delay_out() { + mute(true).await; + } + + async fn mute(lost: bool) { + tokio::time::pause(); + let delay = Duration::from_millis(300); + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + let mut producer = moq_mux::container::Producer::new(track, Container::Legacy(moq_mux::container::Kind::Audio)); + let mut options = moq_audio::decode::Options::new(); + options.delay = moq_audio::decode::Delay { + min: Some(delay), + max: Some(delay), + }; + options.estimate = false; + options.output.format = moq_audio::Format::F32; + let consumer = moq_audio::decode::Consumer::new( + &broadcast.consume(), + &AudioConfig::new(AudioCodec::Pcm, SAMPLE_RATE, 1), + "audio", + options, + ) + .await + .unwrap(); + let recorder = Recorder::default(); + let playback = AudioPlayback { + changed: Default::default(), + speaker: recorder.clone(), + presentation: Presentation::new(delay), + latency: delay, + output: recorder.clone(), + }; + let task = tokio::spawn(async move { + play_audio(consumer, playback, std::future::pending()) + .await + .unwrap() + .finish() + .await; + }); + for index in 0..10 { + producer.write(packet(index, 0.25)).unwrap(); + kio::clock::sleep(PACKET_DURATION).await; + } + producer.discontinuity().unwrap(); + // One group per packet and one for the marker, so the next press would be group 11. + if lost { + producer.seek(12).unwrap(); + } + // Two seconds muted, with the timeline running on through it. + kio::clock::sleep(Duration::from_secs(2)).await; + let pressed = Instant::now(); + for index in 110..120 { + producer.write(packet(index, 0.5)).unwrap(); + kio::clock::sleep(PACKET_DURATION).await; + } + producer.finish().unwrap(); + task.await.unwrap(); + + assert_eq!(recorder.latencies(), vec![delay], "the mute replaced the sink"); + let played = recorder.played(); + let first = played.iter().find(|part| part.sample == 0.5).unwrap(); + assert_eq!(first.utterance, 1); + let waited = first.from - pressed; + if lost { + // The ordered read first waits within its age budget for the lost group. + assert!(waited > delay && waited < 2 * delay, "{played:?}"); + } else { + assert_eq!(waited, delay, "{played:?}"); + } + let new: Vec<_> = played.iter().filter(|part| part.sample == 0.5).collect(); + assert_eq!( + new.iter().map(|part| part.to - part.from).sum::(), + 10 * PACKET_DURATION + ); + } + #[tokio::test] async fn paired_quick_resumes_keep_audio_and_reordered_video_together() { quick_resumes(true, true, Duration::from_millis(200)).await; @@ -1215,19 +1329,13 @@ mod tests { played .iter() .filter(|part| range.contains(&part.sample)) - .map(|part| part.sink) + .map(|part| (part.sink, part.utterance)) .collect::>() }; let (old, short, new) = (sinks(0.2..0.3), sinks(0.4..0.6), sinks(0.7..0.8)); assert!(!short.is_empty(), "the short timeline played: {played:?}"); - assert!( - old.is_disjoint(&short), - "the short timeline resumed on its own sink: {played:?}" - ); - assert!( - short.is_disjoint(&new), - "the next timeline resumed on its own sink: {played:?}" - ); + assert!(old.is_disjoint(&short), "the short timeline did not resume: {played:?}"); + assert!(short.is_disjoint(&new), "the next timeline did not resume: {played:?}"); } #[tokio::test] @@ -1285,7 +1393,7 @@ mod tests { let resumed: Vec<_> = played.iter().filter(|part| part.sample == 0.5).collect(); assert!(!resumed.is_empty(), "{played:?}"); assert!( - resumed.iter().all(|part| part.sink > 0), + resumed.iter().all(|part| part.utterance > 0), "the replacement resumed its declared transition: {played:?}" ); } diff --git a/rs/moq-cli/src/play/output.rs b/rs/moq-cli/src/play/output.rs index 193b3319760..ec5435508fa 100644 --- a/rs/moq-cli/src/play/output.rs +++ b/rs/moq-cli/src/play/output.rs @@ -44,6 +44,12 @@ pub(super) trait Sink: Send + Sync + 'static { /// Play every queued sample before releasing the sink. fn finish(self) -> impl Future + Send; + + /// The audio written so far is followed by a declared silence. + fn end(&mut self); + + /// The next write starts the next utterance after a declared silence. + fn next(&mut self); } impl Speaker for moq_audio::playback::Engine { @@ -70,4 +76,12 @@ impl Sink for moq_audio::playback::Sink { fn buffered(&self) -> Duration { moq_audio::playback::Sink::buffered(self) } + + fn end(&mut self) { + moq_audio::playback::Sink::end(self); + } + + fn next(&mut self) { + moq_audio::playback::Sink::next(self); + } }