diff --git a/CHANGELOG.md b/CHANGELOG.md index 3aa2ebd..e7d0beb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -93,8 +93,31 @@ Use `vp9` or `vp9_svc` for camera video and `vp8` or `vp8_simulcast` for screen share. The `gpt_realtime_bot` example is voice-only, because OpenAI Realtime accepts only H.264 video. +### PCM has a fixed sample rate + +`write_pcm` takes the sample rate of the first frame. A later frame at another +rate gets `RtcError::PcmRateMismatch { expected, actual }` and is not written. +The track still averages the channels of a frame to mono. + +`StreamResampler::new(out_rate, channels)` replaces `new(out_rate)` and +`to_opus_mono()`. The first `push` sets the input rate. A frame with another +rate gets `RtcError::PcmRateMismatch`, and a frame with another channel count +gets `RtcError::IllegalState`. `push` and `flush` return +`RtcResult` with the same channels as the input. The resampler no +longer downmixes to mono. + ## New Features +### The pacer plays the end of the PCM + +`StreamResampler` holds the newest 128 input frames in its filter: 8 ms at +16 kHz, 5.3 ms at 24 kHz. `StreamResampler::flush` returns them. When the queue +of a `LocalAudioTrack` holds less than one 20 ms frame, the pacer first adds +that audio, so the end of the written audio plays without the next write. A +track at 48 kHz holds no audio. Each time the queue runs short in the middle of +the audio, the filter starts again: the gap is at least 11 ms at 16 kHz and +23 ms at 8 kHz. + ### Pacing control for PCM tracks `LocalAudioTrackConfig::pace` (default `true`) starts pacing at the first @@ -192,6 +215,11 @@ on `Call` (`get_call_participant_session_metrics`, packets arrive but no frame is decoded for 3 s, and again after each further 3 s, as libwebrtc does. Before, a stream whose frames could not be assembled stayed frozen and sent no keyframe request. +- `StreamResampler` uses a windowed-sinc filter (rubato) and removes the audio + above the lower Nyquist frequency. Before, it interpolated linearly: from + 48 kHz to 16 kHz, a 12 kHz tone came out at 4 kHz at full level. +- `LocalAudioTrack::flush` also drops the audio in the resampler. A flush during + a `write_pcm` can no longer put audio from before the flush in the queue. # v0.1.0-preview.2 diff --git a/Cargo.lock b/Cargo.lock index bc8f83f..54031be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -144,6 +144,39 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" +[[package]] +name = "audio-codec-algorithms" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1254ebf6529f3763c491acfb5ab6e960809e3c75a38584cc664f8c5667fb7107" + +[[package]] +name = "audioadapter" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1292ef9edf681b7426ed089004b021a42896492f58bd0c909ad44e5faec9ac4" + +[[package]] +name = "audioadapter-buffers" +version = "5.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46289a81a3bfa26d0f8b2415f8ca9decde9a66308a150568af5c102381b3c8ce" +dependencies = [ + "audioadapter", + "audioadapter-sample", + "num-traits", +] + +[[package]] +name = "audioadapter-sample" +version = "5.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5258faecf4edbe35ec483bda9b1f7438e6b391891819269c6b5461aeaa0e8d7d" +dependencies = [ + "audio-codec-algorithms", + "num-traits", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -1150,6 +1183,7 @@ dependencies = [ "prost-types", "protox", "reqwest", + "rubato", "rustls", "serde", "serde_json", @@ -2729,6 +2763,20 @@ dependencies = [ "webrtc-util", ] +[[package]] +name = "rubato" +version = "5.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc1e951b9f5432ec1422f4dcfa5733fc501265902612edf5ef8bd8ecdd960930" +dependencies = [ + "audioadapter", + "audioadapter-buffers", + "num-integer", + "num-traits", + "visibility", + "windowfunctions", +] + [[package]] name = "rustc-hash" version = "1.1.0" @@ -3657,6 +3705,17 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "visibility" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d674d135b4a8c1d7e813e2f8d1c9a58308aee4a680323066025e53132218bd91" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "waitgroup" version = "0.1.2" @@ -4027,6 +4086,15 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windowfunctions" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90628d739333b7c5d2ee0b70210b97b8cddc38440c682c96fd9e2c24c2db5f3a" +dependencies = [ + "num-traits", +] + [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index 9a8e2ca..24c06e5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,6 +52,9 @@ rustdoc-args = ["--cfg", "docsrs", "--generate-link-to-definition"] # `cmake_minimum_required(VERSION 3.16)`, which satisfies CMake 4's # minimum-policy floor without any build-time environment override. opus = "0.4" +# Sinc resampler for PCM written at a rate other than 48 kHz. Only the `Async` +# resampler is used, so the default FFT feature is off. +rubato = { version = "5.0.1", default-features = false } base64 = "0.23.1" bytes = "1.12.1" futures-util = "0.3.34" diff --git a/benches/media_baseline.rs b/benches/media_baseline.rs index a477f7c..fce31e3 100644 --- a/benches/media_baseline.rs +++ b/benches/media_baseline.rs @@ -62,11 +62,11 @@ fn encoded_keyframe(codec: VpxCodec, width: u32, height: u32) -> Vec { fn bench_resampling(criterion: &mut Criterion) { let mut group = criterion.benchmark_group("resample"); for (name, input) in [ - ("44k1_stereo_to_48k_mono_20ms", tone(44_100, 2, 20)), + ("44k1_stereo_to_48k_stereo_20ms", tone(44_100, 2, 20)), ("24k_mono_to_48k_mono_20ms", tone(24_000, 1, 20)), ] { group.throughput(Throughput::Elements(input.frames() as u64)); - let mut resampler = StreamResampler::to_opus_mono(); + let mut resampler = StreamResampler::new(48_000, input.channels); group.bench_function(name, |bencher| { bencher.iter(|| black_box(resampler.push(black_box(&input)))); }); diff --git a/src/rtc/error.rs b/src/rtc/error.rs index 29be991..c1c6f10 100644 --- a/src/rtc/error.rs +++ b/src/rtc/error.rs @@ -158,6 +158,16 @@ pub enum RtcError { capacity_samples: usize, }, + /// A PCM frame is not at the sample rate of the first frame. The frame was + /// not written. + #[error("pcm frame at {actual} Hz, the first frame was at {expected} Hz")] + PcmRateMismatch { + /// Sample rate of the first frame, in Hz. + expected: u32, + /// Sample rate of the rejected frame, in Hz. + actual: u32, + }, + /// Token minting failed for the participant path. /// /// Retained for source compatibility. New typed token-boundary failures use diff --git a/src/rtc/pcm/resample.rs b/src/rtc/pcm/resample.rs index e1ae04f..205b27d 100644 --- a/src/rtc/pcm/resample.rs +++ b/src/rtc/pcm/resample.rs @@ -6,16 +6,26 @@ //! the rate ratio alone, so a 20 ms block in is a 20 ms block out — what a //! model API expecting exact frame sizes needs. Ported from stream-py's //! `Resampler`. -//! - [`StreamResampler`] carries a fractional read position and one sample of -//! history between calls, so a continuous stream joins without clicks at block -//! boundaries. This is what the publish pacer uses. +//! - [`StreamResampler`] runs a windowed-sinc filter (rubato) whose state carries +//! between calls, so a continuous stream joins without clicks at block +//! boundaries and does not alias. This is what the publish pacer uses. //! -//! Both interpolate linearly. That is accurate enough for voice and for the -//! synthesized test tones, and keeps the dependency surface small; a -//! higher-order kernel can replace either one without an API change. +//! [`Resampler`] interpolates linearly, so it aliases when it downsamples. + +use rubato::audioadapter_buffers::direct::InterleavedSlice; +use rubato::{ + Async, FixedAsync, Resampler as _, Resizable as _, SincInterpolationParameters, WindowFunction, +}; use super::OPUS_SAMPLE_RATE; use super::PcmFrame; +use crate::rtc::error::{Result, RtcError}; + +/// Sinc filter length in input frames. The filter delays audio by half of it. +const SINC_LEN: usize = 256; +/// Zero frames that push all held input out of the filter. An output sample +/// reads `SINC_LEN + 1` input frames from a fractional position. +const FLUSH_FRAMES: usize = SINC_LEN + 4; /// Convert one PCM block to a target sample rate and channel count. /// @@ -170,98 +180,163 @@ fn clamp_to_i16(v: f64) -> i16 { v.round().clamp(f64::from(i16::MIN), f64::from(i16::MAX)) as i16 } -/// A streaming linear resampler that downmixes to mono and converts to a fixed -/// output rate (48 kHz by default). +/// A streaming sinc resampler to a fixed output rate and channel count. The +/// first pushed frame sets the input rate, and later frames must keep it. +/// +/// Feed input blocks of any length with [`StreamResampler::push`]; the filter +/// state carries across calls, so consecutive blocks resample continuously. The +/// filter delays audio by 128 input frames (8 ms at 16 kHz) and holds them until +/// more input or [`flush`](Self::flush) pushes them out. Input already at the +/// output rate passes through unfiltered. Use this for a live stream, where +/// block boundaries are arbitrary; use [`Resampler`] when each block must +/// convert to an exact length on its own. +/// +/// ``` +/// use getstream::rtc::{PcmFrame, StreamResampler}; /// -/// Feed input blocks with [`StreamResampler::push`]; it retains a one-sample -/// history and a fractional read position so consecutive blocks resample -/// continuously. Use this for a live stream, where block boundaries are -/// arbitrary; use [`Resampler`] when each block must convert to an exact length -/// on its own. +/// let mut r = StreamResampler::new(48_000, 1); +/// let mut out = r.push(&PcmFrame::mono(vec![0; 320], 16_000))?; +/// out.append(&r.flush()?); +/// assert_eq!(out.sample_rate, 48_000); +/// # Ok::<(), getstream::rtc::RtcError>(()) +/// ``` #[derive(Debug)] pub struct StreamResampler { out_rate: u32, - in_rate: u32, - in_channels: u16, - /// Carried mono input (f32), including one history sample at index 0. - inbuf: Vec, - /// Fractional read position within `inbuf`. - pos: f64, + channels: u16, + /// The rate of the first pushed frame. + in_rate: Option, + /// `None` before the first frame and when the input rate equals the output + /// rate. + filter: Option>, + /// Interleaved input of the current call. + input: Vec, + /// Interleaved filter output, `output_frames_max` frames long. + output: Vec, + /// Set when input went into the filter after the last flush. + holds_audio: bool, } impl StreamResampler { - /// A resampler targeting 48 kHz mono. - pub fn to_opus_mono() -> Self { - Self::new(OPUS_SAMPLE_RATE) - } - - /// A resampler targeting `out_rate` mono. - pub fn new(out_rate: u32) -> Self { + /// A resampler to `out_rate` for `channels` interleaved channels. + pub fn new(out_rate: u32, channels: u16) -> Self { Self { out_rate: out_rate.max(1), - in_rate: 0, - in_channels: 1, - inbuf: Vec::new(), - pos: 0.0, + channels: channels.max(1), + in_rate: None, + filter: None, + input: Vec::new(), + output: Vec::new(), + holds_audio: false, } } - /// Resample and downmix `frame`, returning mono s16 samples at the target - /// rate. Returns an empty vector when there is not yet enough input to emit - /// a sample (the remainder is buffered for the next call). - pub fn push(&mut self, frame: &PcmFrame) -> Vec { - let in_rate = frame.sample_rate.max(1); - let in_channels = frame.channels.max(1); - - // A format change invalidates the carried history; restart cleanly. - if in_rate != self.in_rate || in_channels != self.in_channels { - self.in_rate = in_rate; - self.in_channels = in_channels; - self.inbuf.clear(); - self.pos = 0.0; + /// Resample `frame` to the output rate. The first frame sets the input + /// rate. The newest 128 input frames stay in the filter until the next call + /// or a [`flush`](Self::flush). + /// + /// # Errors + /// + /// Returns [`RtcError::PcmRateMismatch`] if `frame` has another rate than + /// the first frame, [`RtcError::IllegalState`] if it has another channel + /// count than the resampler, and [`RtcError::Media`] if the filter rejects + /// the rate or a buffer. + pub fn push(&mut self, frame: &PcmFrame) -> Result { + let in_rate = self.in_rate.unwrap_or(frame.sample_rate); + if frame.sample_rate != in_rate { + return Err(RtcError::PcmRateMismatch { + expected: in_rate, + actual: frame.sample_rate, + }); } - - // Downmix the incoming block to mono f32 and append. - let ch = in_channels as usize; - self.inbuf.reserve(frame.frames()); - for chunk in frame.samples.chunks_exact(ch) { - let sum: f32 = chunk.iter().map(|&s| f32::from(s)).sum(); - self.inbuf.push(sum / ch as f32); + if frame.channels != self.channels { + return Err(RtcError::IllegalState(format!( + "pcm frame with {} channels, resampler input has {} channels", + frame.channels, self.channels + ))); + } + if self.in_rate.is_none() { + self.start(in_rate)?; } + let samples = frame.frames() * usize::from(self.channels); + self.input.clear(); + self.input + .extend(frame.samples[..samples].iter().map(|&s| f32::from(s))); + self.run() + } - // Fast path: identical rate → no interpolation, just emit and reset. - if in_rate == self.out_rate { - let out: Vec = self - .inbuf - .drain(..) - .map(|v| v.round().clamp(f32::from(i16::MIN), f32::from(i16::MAX)) as i16) - .collect(); - self.pos = 0.0; - return out; + /// Push the audio the filter still holds out with silence and return it. + /// The filter then holds only silence. Returns an empty frame when no audio + /// went in since the last flush, or when the input runs at the output rate. + /// + /// # Errors + /// + /// Returns [`RtcError::Media`] if the filter rejects a buffer. + pub fn flush(&mut self) -> Result { + if !self.holds_audio { + return Ok(PcmFrame::new(Vec::new(), self.out_rate, self.channels)); } + self.input.clear(); + self.input + .resize(FLUSH_FRAMES * usize::from(self.channels), 0.0); + let out = self.run()?; + self.holds_audio = false; + Ok(out) + } - let step = f64::from(in_rate) / f64::from(self.out_rate); - let mut out = Vec::new(); - // Need `floor(pos) + 1` to exist to interpolate. - while self.pos + 1.0 < self.inbuf.len() as f64 { - let idx = self.pos.floor() as usize; - let frac = (self.pos - idx as f64) as f32; - let a = self.inbuf[idx]; - let b = self.inbuf[idx + 1]; - let v = a + (b - a) * frac; - out.push(v.round().clamp(f32::from(i16::MIN), f32::from(i16::MAX)) as i16); - self.pos += step; + /// Build the filter for `in_rate` and fix the input rate. A rate the filter + /// rejects leaves the input rate unset. + fn start(&mut self, in_rate: u32) -> Result<()> { + if in_rate != self.out_rate { + let params = + SincInterpolationParameters::new(SINC_LEN, WindowFunction::BlackmanHarris2); + // The largest block one filter call takes; `run` splits longer input. + let max_chunk = (in_rate as usize / 10).max(1); + let filter = Async::new_sinc( + f64::from(self.out_rate) / f64::from(in_rate), + 1.0, + ¶ms, + max_chunk, + usize::from(self.channels), + FixedAsync::Input, + ) + .map_err(|e| RtcError::Media(e.to_string()))?; + self.output = vec![0.0; filter.output_frames_max() * usize::from(self.channels)]; + self.filter = Some(filter); } + self.in_rate = Some(in_rate); + Ok(()) + } - // Drop fully-consumed input, keeping one sample of history so the next - // block interpolates across the boundary. - let consumed = self.pos.floor() as usize; - if consumed > 0 { - let keep_from = consumed.min(self.inbuf.len()); - self.inbuf.drain(..keep_from); - self.pos -= consumed as f64; + /// Pass `self.input` through the filter. + fn run(&mut self) -> Result { + let ch = usize::from(self.channels); + let mut out = Vec::new(); + if let Some(filter) = self.filter.as_mut() { + for piece in self.input.chunks(filter.input_frames_max() * ch) { + let frames = piece.len() / ch; + filter + .set_chunk_size(frames) + .map_err(|e| RtcError::Media(e.to_string()))?; + let input = InterleavedSlice::new(piece, ch, frames) + .map_err(|e| RtcError::Media(e.to_string()))?; + let capacity = self.output.len() / ch; + let mut output = InterleavedSlice::new_mut(&mut self.output[..], ch, capacity) + .map_err(|e| RtcError::Media(e.to_string()))?; + let (_, written) = filter + .process_into_buffer(&input, &mut output, None) + .map_err(|e| RtcError::Media(e.to_string()))?; + out.extend( + self.output[..written * ch] + .iter() + .map(|&v| clamp_to_i16(f64::from(v))), + ); + self.holds_audio = true; + } + } else { + out.extend(self.input.iter().map(|&v| clamp_to_i16(f64::from(v)))); } - out + Ok(PcmFrame::new(out, self.out_rate, self.channels)) } } @@ -366,27 +441,135 @@ mod tests { assert_eq!(*out.samples.last().unwrap(), i16::MAX); } + /// A silent 20 ms block at `rate` whose last 2 ms carry a loud 1 kHz tone. + fn loud_end(rate: u32) -> PcmFrame { + let frames = rate as usize / 50; + let quiet = frames - rate as usize / 500; + let samples = (0..frames) + .map(|i| { + if i < quiet { + return 0; + } + let time = i as f64 / f64::from(rate); + (20_000.0 * (std::f64::consts::TAU * 1_000.0 * time).sin()) as i16 + }) + .collect(); + PcmFrame::mono(samples, rate) + } + + fn peak(samples: &[i16]) -> i16 { + samples + .iter() + .map(|s| s.saturating_abs()) + .max() + .unwrap_or(0) + } + #[test] - fn stream_identity_rate_passes_through_downmixed() { - let mut r = StreamResampler::to_opus_mono(); - // Stereo 48k -> mono 48k: averages channels, same length per channel. - let frame = PcmFrame::new(vec![100, 200, 300, 400], 48_000, 2); - let out = r.push(&frame); - assert_eq!(out, vec![150, 350]); + fn stream_flush_returns_the_audio_the_filter_still_holds() { + let mut r = StreamResampler::new(48_000, 1); + + let pushed = r.push(&loud_end(16_000)).expect("push"); + let flushed = r.flush().expect("flush"); + + assert!( + peak(&pushed.samples) < 1_000, + "the loud end left before the flush" + ); + assert!( + peak(&flushed.samples) > 15_000, + "the flush did not return the loud end" + ); + } + + #[test] + fn stream_flush_leaves_no_old_audio_for_the_next_block() { + for rate in [8_000, 16_000, 22_050, 24_000, 32_000, 44_100, 96_000] { + let mut r = StreamResampler::new(48_000, 1); + r.push(&loud_end(rate)).expect("push"); + r.flush().expect("flush"); + + let next = r + .push(&PcmFrame::silence(rate as usize / 10, rate, 1)) + .expect("push"); + + assert_eq!( + peak(&next.samples), + 0, + "{rate} Hz: old audio after the flush" + ); + } + } + + #[test] + fn stream_keeps_the_rate_of_the_first_frame_and_its_channels() { + let mut r = StreamResampler::new(48_000, 1); + r.push(&PcmFrame::silence(320, 16_000, 1)).expect("push"); + + let rate = r + .push(&PcmFrame::silence(480, 24_000, 1)) + .expect_err("rate differs"); + let channels = r + .push(&PcmFrame::silence(320, 16_000, 2)) + .expect_err("channels differ"); + + assert!( + matches!( + rate, + RtcError::PcmRateMismatch { + expected: 16_000, + actual: 24_000 + } + ), + "error was: {rate}" + ); + assert!( + matches!(channels, RtcError::IllegalState(_)), + "error was: {channels}" + ); + } + + #[test] + fn stream_at_the_same_rate_copies_every_channel_and_holds_nothing() { + let mut r = StreamResampler::new(48_000, 2); + let frame = PcmFrame::new(vec![100, -200, 300, -400], 48_000, 2); + + let out = r.push(&frame).expect("push"); + + assert_eq!(out, frame); + assert!(r.flush().expect("flush").is_empty()); + } + + #[test] + fn stream_resamples_each_channel_on_its_own() { + let mut r = StreamResampler::new(48_000, 2); + let left = loud_end(16_000).samples; + let stereo: Vec = left.iter().flat_map(|&l| [l, 0]).collect(); + + let mut out = r.push(&PcmFrame::new(stereo, 16_000, 2)).expect("push"); + out.append(&r.flush().expect("flush")); + + assert_eq!((out.sample_rate, out.channels), (48_000, 2)); + let right: Vec = out.samples.iter().skip(1).step_by(2).copied().collect(); + assert!( + peak(&out.samples) > 15_000, + "the left channel lost its audio" + ); + assert_eq!(peak(&right), 0, "the silent right channel picked up audio"); } #[test] fn stream_downsample_halves_sample_count() { // 96k mono -> 48k mono ≈ half as many samples. - let mut r = StreamResampler::to_opus_mono(); + let mut r = StreamResampler::new(48_000, 1); let input: Vec = (0..960).map(|i| (i % 100) as i16).collect(); let frame = PcmFrame::mono(input, 96_000); - let out = r.push(&frame); + let out = r.push(&frame).expect("push"); // ~480 output samples (± a couple for boundary handling). assert!( - (475..=480).contains(&out.len()), + (475..=480).contains(&out.frames()), "unexpected out len {}", - out.len() + out.frames() ); } @@ -394,13 +577,15 @@ mod tests { fn stream_upsample_is_continuous_across_blocks() { // 24k -> 48k across two blocks should roughly double total samples and // not panic on the boundary. - let mut r = StreamResampler::to_opus_mono(); + let mut r = StreamResampler::new(48_000, 1); let block: Vec = (0..240).map(|i| i as i16).collect(); - let a = r.push(&PcmFrame::mono(block.clone(), 24_000)); - let b = r.push(&PcmFrame::mono(block, 24_000)); + let a = r + .push(&PcmFrame::mono(block.clone(), 24_000)) + .expect("push"); + let b = r.push(&PcmFrame::mono(block, 24_000)).expect("push"); assert!(!a.is_empty() && !b.is_empty()); // 480 input samples @2x ≈ 960 output (allow slack for warm-up). - let total = a.len() + b.len(); + let total = a.frames() + b.frames(); assert!((950..=960).contains(&total), "unexpected total {total}"); } } diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index b9fafe4..d4b4049 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -4,9 +4,11 @@ //! Rust analog of videosdk's `track.Local` (`WriteRTP` / `WriteSample` / //! `StartWrite`). Three write paths feed the same outbound track: //! -//! - [`LocalAudioTrack::write_pcm`] — raw [`PcmFrame`]s. Resampled to 48 kHz -//! mono, queued, and paced into 20 ms Opus frames by a background task that -//! emits silence on starve (stream-py `AudioStreamTrack` pacing). Pacing +//! - [`LocalAudioTrack::write_pcm`] — raw [`PcmFrame`]s at the rate of the +//! first frame. Averaged to mono, resampled to 48 kHz, queued, and paced +//! into 20 ms Opus frames by a background task that +//! emits silence on starve (stream-py `AudioStreamTrack` pacing). Before +//! the silence, it takes the audio that the resampler still holds. Pacing //! starts at the first write, or at [`LocalAudioTrack::start_pacing`] when //! [`LocalAudioTrackConfig::pace`] is off. A published track holds its queue //! until the SFU publisher first connects and does not pause on a later @@ -346,6 +348,8 @@ struct AudioInner { /// Resampled 48 kHz mono PCM awaiting the 20 ms pacer. pcm: StdMutex>, pcm_capacity_samples: usize, + /// Locked before `pcm`. A writer keeps it until its audio is queued, so a + /// flush cannot fall between the resample and the queue. resampler: StdMutex, encoder: StdMutex, pacer: StdMutex>>, @@ -361,6 +365,38 @@ struct AudioInner { write_guard: tokio::sync::Mutex<()>, } +impl AudioInner { + /// Fill `frame` from the queue and then with silence. A queue shorter than + /// `frame` first gets the audio that the resampler still holds, so the end + /// of the written audio plays without a gap. + fn take_pcm(&self, frame: &mut [i16]) { + let mut buf = self.pcm.lock().unwrap_or_else(|e| e.into_inner()); + if buf.len() < frame.len() { + drop(buf); + let mut resampler = self.resampler.lock().unwrap_or_else(|e| e.into_inner()); + buf = self.pcm.lock().unwrap_or_else(|e| e.into_inner()); + if buf.len() < frame.len() { + match resampler.flush() { + Ok(mut tail) => { + // The silence after the tail would delay the next write. + let end = tail + .samples + .iter() + .rposition(|&s| s != 0) + .map_or(0, |i| i + 1); + tail.samples.truncate(end); + push_bounded_pcm(&mut buf, tail.samples, self.pcm_capacity_samples); + } + Err(e) => tracing::debug!(error = %e, "stream.rtc.audio.flush_failed"), + } + } + } + for slot in frame.iter_mut() { + *slot = buf.pop_front().unwrap_or(0); + } + } +} + /// Encoder settings for a locally encoded Opus track. #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[non_exhaustive] @@ -498,7 +534,7 @@ impl LocalAudioTrack { core, pcm: StdMutex::new(VecDeque::new()), pcm_capacity_samples, - resampler: StdMutex::new(StreamResampler::to_opus_mono()), + resampler: StdMutex::new(StreamResampler::new(OPUS_SAMPLE_RATE, 1)), encoder: StdMutex::new(encoder), pacer: StdMutex::new(None), pacer_started: AtomicBool::new(false), @@ -513,12 +549,16 @@ impl LocalAudioTrack { /// Queue a PCM frame for the paced 20 ms Opus encoder. /// - /// The frame is resampled to 48 kHz mono and buffered up to + /// The first frame sets the sample rate of the track, and later frames must + /// keep it. The channels of a frame are averaged to mono, and the mono + /// audio is resampled to 48 kHz and buffered up to /// [`LocalAudioTrackConfig::pcm_queue_capacity`]. The first write starts /// pacing when [`LocalAudioTrackConfig::pace`] is set (the default); /// otherwise [`start_pacing`](Self::start_pacing) does. While pacing runs, a /// background task emits one Opus packet every 20 ms, writing silence when - /// the buffer runs dry. This method + /// the buffer runs dry. Before the silence, it takes the audio that the + /// resampler still holds (8 ms at 16 kHz), so the end of the written audio + /// plays without the next write. This method /// does not backpressure a producer: overflow drops the oldest queued /// samples and retains the newest audio. [`flush`](Self::flush) still drops /// all unsent samples immediately for barge-in. @@ -528,7 +568,8 @@ impl LocalAudioTrack { /// Returns [`RtcError::PcmQueueOverflow`] after retaining the newest audio /// when this write exceeds the queue capacity. The caller may continue /// writing; the typed error makes overload observable without allowing - /// stale audio to accumulate. + /// stale audio to accumulate. Returns [`RtcError::PcmRateMismatch`] when + /// the frame is not at the rate of the first frame. pub async fn write_pcm(&self, frame: PcmFrame) -> Result<()> { if self.inner.core.stopped.load(Ordering::SeqCst) { return Err(RtcError::IllegalState( @@ -536,39 +577,54 @@ impl LocalAudioTrack { )); } self.inner.pcm_pacing.store(true, Ordering::SeqCst); - let resampled = { + let frame = if frame.channels > 1 { + let channels = usize::from(frame.channels); + let samples = frame + .samples + .chunks_exact(channels) + .map(|chunk| { + let sum: f32 = chunk.iter().map(|&s| f32::from(s)).sum(); + (sum / channels as f32).round() as i16 + }) + .collect(); + PcmFrame::mono(samples, frame.sample_rate) + } else { + frame + }; + let queued = { let mut r = self .inner .resampler .lock() .unwrap_or_else(|e| e.into_inner()); - r.push(&frame) - }; - let dropped = { - let mut buf = self.inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); - let dropped = push_bounded_pcm(&mut buf, resampled, self.inner.pcm_capacity_samples); - if dropped > 0 { - tracing::debug!( - dropped_samples = dropped, - capacity_samples = self.inner.pcm_capacity_samples, - "stream.rtc.audio.pcm_queue_overflow" - ); - } - dropped + let resampled = r.push(&frame)?.samples; + self.queue_pcm(resampled) }; // Only a pacer that never started starts here, so a `pause_pacing` // stays in force. if self.inner.pace && !self.inner.pacer_started.load(Ordering::SeqCst) { self.start_pacing().await; } - if dropped > 0 { - Err(RtcError::PcmQueueOverflow { - dropped_samples: dropped, - capacity_samples: self.inner.pcm_capacity_samples, - }) - } else { - Ok(()) + queued + } + + /// Append 48 kHz mono samples to the pacer queue, dropping the oldest + /// audio above the capacity. Callers hold the `resampler` lock. + fn queue_pcm(&self, samples: Vec) -> Result<()> { + let mut buf = self.inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); + let dropped = push_bounded_pcm(&mut buf, samples, self.inner.pcm_capacity_samples); + if dropped == 0 { + return Ok(()); } + tracing::debug!( + dropped_samples = dropped, + capacity_samples = self.inner.pcm_capacity_samples, + "stream.rtc.audio.pcm_queue_overflow" + ); + Err(RtcError::PcmQueueOverflow { + dropped_samples: dropped, + capacity_samples: self.inner.pcm_capacity_samples, + }) } /// Write an already-encoded Opus frame of `duration` (packetized immediately; @@ -634,8 +690,17 @@ impl LocalAudioTrack { self.inner.core.forward_rtp(&packet).await } - /// Drop any buffered-but-unsent PCM (barge-in). Does not stop the track. + /// Drop any buffered-but-unsent PCM (barge-in), including the audio that + /// the resampler still holds. Does not stop the track. pub fn flush(&self) { + let mut r = self + .inner + .resampler + .lock() + .unwrap_or_else(|e| e.into_inner()); + if let Err(e) = r.flush() { + tracing::debug!(error = %e, "stream.rtc.audio.flush_failed"); + } let mut buf = self.inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); buf.clear(); } @@ -769,12 +834,7 @@ async fn pace_audio(track: Weak) { if !inner.pcm_pacing.load(Ordering::SeqCst) { continue; } - { - let mut buf = inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); - for slot in scratch.iter_mut() { - *slot = buf.pop_front().unwrap_or(0); - } - } + inner.take_pcm(&mut scratch); // A muted track sends nothing, so its audio is dropped without an encode. // The RTP clock keeps time, so the first packet after the mute shows the // gap (RFC 3550 §5.1). @@ -2294,6 +2354,83 @@ mod tests { track.stop(); } + /// A silent 20 ms block at 16 kHz whose last 2 ms carry a loud 1 kHz tone. + fn loud_end_16k() -> PcmFrame { + let samples = (0..320) + .map(|index| { + if index < 288 { + return 0; + } + let time = index as f64 / 16_000.0; + (20_000.0 * (std::f64::consts::TAU * 1_000.0 * time).sin()) as i16 + }) + .collect(); + PcmFrame::mono(samples, 16_000) + } + + fn unpaced_track() -> LocalAudioTrack { + LocalAudioTrack::opus_with_config(LocalAudioTrackConfig::default().with_pace(false)) + .expect("opus track") + } + + fn peak(samples: &[i16]) -> i16 { + samples + .iter() + .map(|s| s.saturating_abs()) + .max() + .unwrap_or(0) + } + + #[tokio::test] + async fn a_short_queue_takes_the_end_of_the_written_audio() { + let track = unpaced_track(); + track.write_pcm(loud_end_16k()).await.expect("write_pcm"); + let mut first = vec![0; FRAME_SAMPLES_20MS]; + let mut second = vec![0; FRAME_SAMPLES_20MS]; + + track.inner.take_pcm(&mut first); + track.inner.take_pcm(&mut second); + + assert!(peak(&first) < 1_000, "the loud end came before its time"); + assert!(peak(&second) > 15_000, "the loud end did not play"); + track.stop(); + } + + #[tokio::test] + async fn the_end_of_the_written_audio_leaves_no_silence_in_the_queue() { + let track = unpaced_track(); + let tone_5ms = (0..80) + .map(|index| { + let time = index as f64 / 16_000.0; + (12_000.0 * (std::f64::consts::TAU * 440.0 * time).sin()) as i16 + }) + .collect(); + track + .write_pcm(PcmFrame::mono(tone_5ms, 16_000)) + .await + .expect("write_pcm"); + + track.inner.take_pcm(&mut vec![0; FRAME_SAMPLES_20MS]); + + let queue = track.inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); + assert!(queue.is_empty(), "{} samples of silence wait", queue.len()); + drop(queue); + track.stop(); + } + + #[tokio::test] + async fn flush_drops_the_audio_the_resampler_still_holds() { + let track = unpaced_track(); + track.write_pcm(loud_end_16k()).await.expect("write_pcm"); + + track.flush(); + let mut frame = vec![0; FRAME_SAMPLES_20MS]; + track.inner.take_pcm(&mut frame); + + assert_eq!(peak(&frame), 0, "audio from before the flush played"); + track.stop(); + } + #[test] fn pcm_queue_overflow_drops_oldest_and_flush_still_clears() { let track = LocalAudioTrack::opus().expect("opus track");