diff --git a/Cargo.lock b/Cargo.lock index f54e14e02..2bb64d9fd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7139,6 +7139,7 @@ dependencies = [ "lazy_static", "libpulse-binding", "libpulse-simple-binding", + "libsamplerate-sys", "libxdo-sys", "mac_address", "magnum-opus", diff --git a/Cargo.toml b/Cargo.toml index 00bb88fae..dfb93c248 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ path = "src/service.rs" [features] inline = [] -use_samplerate = ["samplerate"] +use_samplerate = ["samplerate", "libsamplerate-sys"] use_rubato = ["rubato"] use_dasp = ["dasp"] flutter = ["flutter_rust_bridge"] @@ -67,6 +67,7 @@ magnum-opus = { git = "https://github.com/rustdesk-org/magnum-opus" } dasp = { version = "0.11", features = ["signal", "interpolate-linear", "interpolate"], optional = true } rubato = { version = "0.12", optional = true } samplerate = { version = "0.2", optional = true } +libsamplerate-sys = { version = "0.1.12", optional = true } uuid = { version = "1.3", features = ["v4"] } num_cpus = "1.15" bytes = { version = "1.4", features = ["serde"] } diff --git a/src/audio_resampler.rs b/src/audio_resampler.rs new file mode 100644 index 000000000..a95600c75 --- /dev/null +++ b/src/audio_resampler.rs @@ -0,0 +1,242 @@ +use hbb_common::thiserror; + +#[cfg(test)] +pub(crate) mod allocation_tests; + +#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))] +mod sinc; + +const INTERPOLATION_MARGIN_FRAMES: usize = 2; +const PENDING_PACKET_CAPACITY: usize = 2; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct AudioResamplerConfig { + pub input_rate: u32, + pub output_rate: u32, + pub channels: u16, +} + +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub(crate) enum AudioResamplerError { + #[error( + "invalid audio resampler configuration: input_rate={}, output_rate={}, channels={}", + .0.input_rate, .0.output_rate, .0.channels + )] + InvalidConfig(AudioResamplerConfig), + #[error("invalid resampler output frame size: {output_frames}")] + InvalidOutputFrameSize { output_frames: usize }, + #[error("audio resampler input length {samples} is not divisible by channel count {channels}")] + IncompleteFrame { samples: usize, channels: usize }, + #[error("audio resampler output capacity overflow")] + CapacityOverflow, + #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))] + #[error("audio resampler backend failed: {0}")] + Backend(String), +} + +pub(crate) struct FixedFrameAudioResampler { + resampler: AudioResampler, + output_samples: usize, + pending_samples: Vec, +} + +#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))] +// SAFETY: libsamplerate's src_new state owns heap data and has no thread affinity. +// This wrapper never exposes or shares that state; processing requires &mut self. +unsafe impl Send for FixedFrameAudioResampler {} + +impl FixedFrameAudioResampler { + pub(crate) fn new( + config: AudioResamplerConfig, + output_frames: usize, + ) -> Result { + if output_frames == 0 { + return Err(AudioResamplerError::InvalidOutputFrameSize { output_frames }); + } + let channels = validate_config(config)?; + let output_samples = output_frames + .checked_mul(channels) + .ok_or(AudioResamplerError::CapacityOverflow)?; + let input_frames = output_frames + .checked_mul(config.input_rate as usize) + .ok_or(AudioResamplerError::CapacityOverflow)? + .div_ceil(config.output_rate as usize); + let capacity = output_samples + .checked_mul(PENDING_PACKET_CAPACITY) + .and_then(|samples| samples.checked_add(channels * INTERPOLATION_MARGIN_FRAMES)) + .ok_or(AudioResamplerError::CapacityOverflow)?; + let mut resampler = AudioResampler::new(config)?; + resampler.reserve_input(input_frames)?; + Ok(Self { + resampler, + output_samples, + pending_samples: Vec::with_capacity(capacity), + }) + } + + pub(crate) fn process_with( + &mut self, + input: &[f32], + mut on_packet: impl FnMut(&[f32]), + ) -> Result<(), AudioResamplerError> { + self.resampler + .process_into(input, &mut self.pending_samples)?; + let complete_samples = + self.pending_samples.len() / self.output_samples * self.output_samples; + for packet in self.pending_samples[..complete_samples].chunks_exact(self.output_samples) { + on_packet(packet); + } + self.pending_samples.drain(..complete_samples); + Ok(()) + } + + #[cfg(test)] + pub(crate) fn process(&mut self, input: &[f32]) -> Result>, AudioResamplerError> { + let mut packets = Vec::new(); + self.process_with(input, |packet| packets.push(packet.to_owned()))?; + Ok(packets) + } +} + +pub(crate) struct AudioResampler { + #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))] + backend: StreamingLinearAudioResampler, + #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))] + backend: sinc::SincAudioResampler, +} + +impl AudioResampler { + pub(crate) fn new(config: AudioResamplerConfig) -> Result { + Ok(Self { + #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))] + backend: sinc::SincAudioResampler::new(config)?, + #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))] + backend: StreamingLinearAudioResampler::new(config)?, + }) + } + + pub(crate) fn process(&mut self, input: &[f32]) -> Result, AudioResamplerError> { + let mut output = Vec::new(); + self.process_into(input, &mut output)?; + Ok(output) + } + + // Append samples so capture can retain an incomplete output packet in the same buffer. + fn process_into( + &mut self, + input: &[f32], + output: &mut Vec, + ) -> Result<(), AudioResamplerError> { + self.backend.process_into(input, output) + } + + fn reserve_input(&mut self, _frames: usize) -> Result<(), AudioResamplerError> { + #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))] + { + let capacity = _frames + .checked_add(INTERPOLATION_MARGIN_FRAMES) + .and_then(|frames| frames.checked_mul(self.backend.channels)) + .ok_or(AudioResamplerError::CapacityOverflow)?; + self.backend.buffered_samples.reserve(capacity); + } + Ok(()) + } +} + +#[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))] +struct StreamingLinearAudioResampler { + config: AudioResamplerConfig, + channels: usize, + buffered_samples: Vec, + next_position: u64, +} + +#[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))] +impl StreamingLinearAudioResampler { + fn new(config: AudioResamplerConfig) -> Result { + Ok(Self { + config, + channels: validate_config(config)?, + buffered_samples: Vec::new(), + next_position: 0, + }) + } + + fn process_into( + &mut self, + input: &[f32], + output: &mut Vec, + ) -> Result<(), AudioResamplerError> { + validate_input(input, self.channels)?; + let capacity = self.output_capacity(input.len())?; + output.reserve(capacity); + self.buffered_samples.extend_from_slice(input); + while self.write_next_frame(output) { + self.next_position += self.config.input_rate as u64; + } + self.discard_consumed_frames(); + Ok(()) + } + + fn output_capacity(&self, input_samples: usize) -> Result { + let input_frames = input_samples / self.channels; + let scaled_frames = input_frames + .checked_mul(self.config.output_rate as usize) + .ok_or(AudioResamplerError::CapacityOverflow)? + / self.config.input_rate as usize; + scaled_frames + .checked_add(INTERPOLATION_MARGIN_FRAMES) + .and_then(|frames| frames.checked_mul(self.channels)) + .ok_or(AudioResamplerError::CapacityOverflow) + } + + fn write_next_frame(&self, output: &mut Vec) -> bool { + let output_rate = self.config.output_rate as u64; + let frame_count = self.buffered_samples.len() / self.channels; + let frame = (self.next_position / output_rate) as usize; + let fraction = self.next_position % output_rate; + if frame >= frame_count || (fraction != 0 && frame + 1 >= frame_count) { + return false; + } + let weight = fraction as f32 / output_rate as f32; + for channel in 0..self.channels { + let current = self.buffered_samples[frame * self.channels + channel]; + let next_frame = frame + usize::from(fraction != 0); + let next = self.buffered_samples[next_frame * self.channels + channel]; + output.push(current + (next - current) * weight); + } + true + } + + fn discard_consumed_frames(&mut self) { + let output_rate = self.config.output_rate as u64; + let available_frames = self.buffered_samples.len() / self.channels; + let consumed_frames = ((self.next_position / output_rate) as usize).min(available_frames); + self.buffered_samples + .drain(0..consumed_frames * self.channels); + self.next_position -= consumed_frames as u64 * output_rate; + } +} + +fn validate_config(config: AudioResamplerConfig) -> Result { + if config.input_rate == 0 || config.output_rate == 0 || config.channels == 0 { + return Err(AudioResamplerError::InvalidConfig(config)); + } + Ok(config.channels as usize) +} + +fn validate_input(input: &[f32], channels: usize) -> Result<(), AudioResamplerError> { + if input.len() % channels != 0 { + return Err(AudioResamplerError::IncompleteFrame { + samples: input.len(), + channels, + }); + } + Ok(()) +} + +#[cfg(all(test, not(all(feature = "use_samplerate", not(feature = "use_dasp")))))] +mod tests; + +#[cfg(all(test, feature = "use_samplerate", not(feature = "use_dasp")))] +mod samplerate_tests; diff --git a/src/audio_resampler/allocation_tests.rs b/src/audio_resampler/allocation_tests.rs new file mode 100644 index 000000000..c944d2eff --- /dev/null +++ b/src/audio_resampler/allocation_tests.rs @@ -0,0 +1,142 @@ +use super::{AudioResamplerConfig, FixedFrameAudioResampler}; +use std::alloc::{GlobalAlloc, Layout, System}; +use std::cell::Cell; + +struct CountingAllocator; + +thread_local! { + static ALLOCATIONS: Cell> = const { Cell::new(None) }; +} + +fn record_allocation() { + let _ = ALLOCATIONS.try_with(|count| { + if let Some(value) = count.get() { + count.set(Some(value + 1)); + } + }); +} + +unsafe impl GlobalAlloc for CountingAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + record_allocation(); + unsafe { System.alloc(layout) } + } + + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + record_allocation(); + unsafe { System.alloc_zeroed(layout) } + } + + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 { + record_allocation(); + unsafe { System.realloc(ptr, layout, size) } + } + + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + unsafe { System.dealloc(ptr, layout) } + } +} + +#[global_allocator] +static ALLOCATOR: CountingAllocator = CountingAllocator; + +pub(crate) fn assert_no_allocations(process: impl FnOnce()) { + struct ResetCounter; + impl Drop for ResetCounter { + fn drop(&mut self) { + ALLOCATIONS.with(|count| count.set(None)); + } + } + + ALLOCATIONS.with(|count| assert!(count.replace(Some(0)).is_none())); + let reset = ResetCounter; + process(); + let allocations = ALLOCATIONS.with(|count| count.get().unwrap()); + drop(reset); + assert_eq!( + allocations, 0, + "PCM processing allocated on the capture thread" + ); +} + +#[test] +fn capture_resampling_reuses_buffers() { + const PACKETS_PER_SECOND: usize = 100; + const PACKET_COUNT: usize = 100; + const MAX_STARTUP_DELAY_PACKETS: usize = 1; + const SIGNAL_LEVEL: f32 = 0.25; + const RATE_PAIRS: [(u32, u32); 6] = [ + (32_000, 24_000), + (44_100, 24_000), + (44_100, 48_000), + (48_000, 24_000), + (96_000, 48_000), + (192_000, 48_000), + ]; + + for (input_rate, output_rate) in RATE_PAIRS { + for channels in [1, 2, 4, 6, 8] { + let config = AudioResamplerConfig { + input_rate, + output_rate, + channels, + }; + let input = + vec![SIGNAL_LEVEL; input_rate as usize / PACKETS_PER_SECOND * channels as usize]; + let frames = output_rate as usize / PACKETS_PER_SECOND; + let mut resampler = FixedFrameAudioResampler::new(config, frames).unwrap(); + let mut packets = 0; + let mut energy = 0.0; + assert_no_allocations(|| { + for _ in 0..PACKET_COUNT { + resampler + .process_with(&input, |packet| { + assert_eq!(packet.len(), frames * channels as usize); + energy += packet.iter().map(|sample| sample * sample).sum::(); + packets += 1; + }) + .unwrap(); + } + }); + assert!((PACKET_COUNT - MAX_STARTUP_DELAY_PACKETS..=PACKET_COUNT).contains(&packets)); + assert!(energy > SIGNAL_LEVEL); + } + } +} + +#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))] +#[test] +fn sinc_output_matches_the_existing_backend() { + use super::AudioResampler; + + const INPUT_FRAMES: usize = 2_048; + const CHUNK_FRAMES: usize = 73; + const SIGNAL_STEP: f32 = 0.07; + for (input_rate, output_rate) in [(44_100, 24_000), (44_100, 48_000), (96_000, 48_000)] { + for channels in [1, 2, 4, 6, 8] { + let config = AudioResamplerConfig { + input_rate, + output_rate, + channels, + }; + let input: Vec<_> = (0..INPUT_FRAMES * channels as usize) + .map(|sample| (sample as f32 * SIGNAL_STEP).sin()) + .collect(); + let mut actual = AudioResampler::new(config).unwrap(); + let expected = samplerate::Samplerate::new( + samplerate::ConverterType::SincBestQuality, + input_rate, + output_rate, + channels as usize, + ) + .unwrap(); + for chunk in input.chunks(CHUNK_FRAMES * channels as usize) { + assert_eq!( + actual.process(chunk).unwrap(), + expected.process(chunk).unwrap() + ); + assert_eq!(actual.process(&[]).unwrap(), expected.process(&[]).unwrap()); + } + } + } +} diff --git a/src/audio_resampler/samplerate_tests.rs b/src/audio_resampler/samplerate_tests.rs new file mode 100644 index 000000000..0ba78063a --- /dev/null +++ b/src/audio_resampler/samplerate_tests.rs @@ -0,0 +1,157 @@ +use super::{AudioResampler, AudioResamplerConfig, AudioResamplerError, FixedFrameAudioResampler}; + +const INPUT_RATE: u32 = 44_100; +const OUTPUT_RATE: u32 = 48_000; +const CHANNELS: u16 = 2; +const INPUT_PACKET_FRAMES: usize = INPUT_RATE as usize / PACKETS_PER_SECOND; +const OUTPUT_PACKET_FRAMES: usize = OUTPUT_RATE as usize / PACKETS_PER_SECOND; +const PACKET_COUNT: usize = 20; +const PACKETS_PER_SECOND: usize = 100; +const MIN_CONTINUITY_PACKETS: usize = 2; +const TONE_FREQUENCY_HZ: f32 = 997.0; +const TONE_AMPLITUDE: f32 = 0.5; +const MAX_BOUNDARY_RESIDUAL: f32 = 0.02; +const INCOMPLETE_SAMPLE_COUNT: usize = 1; +const DOWNSAMPLE_RATE: u32 = 24_000; +const REJECTED_TONE_HZ: f64 = 18_000.0; +const MAX_ALIAS_RMS: f64 = 0.01; +const MIN_PASSBAND_RMS: f64 = 0.3; + +fn stereo_tone(frames: usize) -> Vec { + (0..frames) + .flat_map(|frame| { + let phase = + std::f32::consts::TAU * TONE_FREQUENCY_HZ * frame as f32 / INPUT_RATE as f32; + let sample = TONE_AMPLITUDE * phase.sin(); + [sample, sample] + }) + .collect() +} + +fn maximum_boundary_residual(packets: &[Vec]) -> f32 { + packets.windows(2).fold(0.0, |maximum, pair| { + let previous = &pair[0]; + let current = &pair[1]; + let last = previous.len() - CHANNELS as usize; + let penultimate = last - CHANNELS as usize; + (0..CHANNELS as usize).fold(maximum, |maximum, channel| { + let predicted = previous[last + channel] + + (previous[last + channel] - previous[penultimate + channel]); + maximum.max((current[channel] - predicted).abs()) + }) + }) +} + +fn stereo_config() -> AudioResamplerConfig { + AudioResamplerConfig { + input_rate: INPUT_RATE, + output_rate: OUTPUT_RATE, + channels: CHANNELS, + } +} + +#[test] +fn moving_capture_resampler_preserves_pending_audio() { + let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT); + let packet_samples = INPUT_PACKET_FRAMES * CHANNELS as usize; + let mut expected_resampler = + FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap(); + let expected: Vec<_> = input + .chunks(packet_samples) + .flat_map(|packet| expected_resampler.process(packet).unwrap()) + .collect(); + let mut moved_resampler = + FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap(); + let mut output = moved_resampler.process(&input[..packet_samples]).unwrap(); + let remaining = std::thread::spawn(move || { + input[packet_samples..] + .chunks(packet_samples) + .flat_map(|packet| moved_resampler.process(packet).unwrap()) + .collect::>() + }) + .join() + .unwrap(); + output.extend(remaining); + + assert!(output.len() >= MIN_CONTINUITY_PACKETS); + assert!(output + .iter() + .all(|packet| packet.len() == OUTPUT_PACKET_FRAMES * CHANNELS as usize)); + assert!(maximum_boundary_residual(&output) <= MAX_BOUNDARY_RESIDUAL); + assert_eq!(output, expected); +} + +fn downsampled_rms(input: &[f32]) -> f64 { + let config = AudioResamplerConfig { + output_rate: DOWNSAMPLE_RATE, + ..stereo_config() + }; + let output_frames = DOWNSAMPLE_RATE as usize / PACKETS_PER_SECOND; + let mut resampler = FixedFrameAudioResampler::new(config, output_frames).unwrap(); + let output: Vec = input + .chunks(INPUT_PACKET_FRAMES * CHANNELS as usize) + .flat_map(|packet| resampler.process(packet).unwrap().into_iter().flatten()) + .collect(); + + assert!(output.len() >= output_frames * CHANNELS as usize * MIN_CONTINUITY_PACKETS); + let mean_square = output + .iter() + .map(|sample| f64::from(*sample).powi(2)) + .sum::() + / output.len() as f64; + mean_square.sqrt() +} + +#[test] +fn capture_downsampling_filters_out_of_band_audio() { + let input: Vec<_> = (0..INPUT_PACKET_FRAMES * PACKET_COUNT) + .flat_map(|frame| { + let phase = + std::f64::consts::TAU * REJECTED_TONE_HZ * frame as f64 / f64::from(INPUT_RATE); + let sample = (f64::from(TONE_AMPLITUDE) * phase.sin()) as f32; + [sample, sample] + }) + .collect(); + let rms = downsampled_rms(&input); + assert!( + rms < MAX_ALIAS_RMS, + "out-of-band output RMS {rms} exceeded {MAX_ALIAS_RMS}" + ); + + let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT); + let rms = downsampled_rms(&input); + assert!( + rms > MIN_PASSBAND_RMS, + "in-band output RMS {rms} fell below {MIN_PASSBAND_RMS}" + ); +} + +#[test] +fn samplerate_backend_preserves_streaming_continuity() { + let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT); + let mut resampler = AudioResampler::new(stereo_config()).unwrap(); + let packets: Vec<_> = input + .chunks(INPUT_PACKET_FRAMES * CHANNELS as usize) + .map(|packet| resampler.process(packet).unwrap()) + .filter(|packet| !packet.is_empty()) + .collect(); + + assert!(packets.len() >= MIN_CONTINUITY_PACKETS); + assert!(packets + .iter() + .all(|packet| packet.len() % CHANNELS as usize == 0)); + assert!(maximum_boundary_residual(&packets) <= MAX_BOUNDARY_RESIDUAL); +} + +#[test] +fn samplerate_backend_reports_incomplete_frame_context() { + let mut resampler = AudioResampler::new(stereo_config()).unwrap(); + + assert_eq!( + resampler.process(&[0.0]).unwrap_err(), + AudioResamplerError::IncompleteFrame { + samples: INCOMPLETE_SAMPLE_COUNT, + channels: CHANNELS as usize, + } + ); +} diff --git a/src/audio_resampler/sinc.rs b/src/audio_resampler/sinc.rs new file mode 100644 index 000000000..4585d6ca8 --- /dev/null +++ b/src/audio_resampler/sinc.rs @@ -0,0 +1,112 @@ +use super::{AudioResamplerConfig, AudioResamplerError}; +use libsamplerate_sys as sys; +use std::ptr::NonNull; + +const OUTPUT_MARGIN_FRAMES: usize = 1; + +pub(super) struct SincAudioResampler { + state: NonNull, + config: AudioResamplerConfig, +} + +impl SincAudioResampler { + pub(super) fn new(config: AudioResamplerConfig) -> Result { + super::validate_config(config)?; + let ratio = f64::from(config.output_rate) / f64::from(config.input_rate); + if unsafe { sys::src_is_valid_ratio(ratio) } == 0 { + return Err(backend_error( + config, + samplerate::ErrorCode::BadSrcRatio as _, + )); + } + let mut error = 0; + // SAFETY: src_new allocates independent state; this owner releases it in Drop. + let state = unsafe { + sys::src_new( + sys::SRC_SINC_BEST_QUALITY as _, + config.channels.into(), + &mut error, + ) + }; + let state = NonNull::new(state).ok_or_else(|| backend_error(config, error))?; + Ok(Self { state, config }) + } + + pub(super) fn process_into( + &mut self, + input: &[f32], + output: &mut Vec, + ) -> Result<(), AudioResamplerError> { + super::validate_input(input, self.config.channels as usize)?; + let mut consumed = 0; + loop { + let (used, generated) = self.process_block(&input[consumed..], output)?; + consumed += used; + if consumed == input.len() { + return Ok(()); + } + if used == 0 && generated == 0 { + return Err(AudioResamplerError::Backend( + "libsamplerate made no progress while input remained".to_owned(), + )); + } + } + } + + fn process_block( + &mut self, + input: &[f32], + output: &mut Vec, + ) -> Result<(usize, usize), AudioResamplerError> { + let channels = self.config.channels as usize; + let input_frames = input.len() / channels; + let output_frames = input_frames + .checked_mul(self.config.output_rate as usize) + .map(|frames| frames / self.config.input_rate as usize) + .and_then(|frames| frames.checked_add(OUTPUT_MARGIN_FRAMES)) + .ok_or(AudioResamplerError::CapacityOverflow)?; + let start = output.len(); + let end = output_frames + .checked_mul(channels) + .and_then(|samples| start.checked_add(samples)) + .ok_or(AudioResamplerError::CapacityOverflow)?; + let mut data = sys::SRC_DATA { + data_in: input.as_ptr(), + input_frames: input_frames + .try_into() + .map_err(|_| AudioResamplerError::CapacityOverflow)?, + output_frames: output_frames + .try_into() + .map_err(|_| AudioResamplerError::CapacityOverflow)?, + src_ratio: f64::from(self.config.output_rate) / f64::from(self.config.input_rate), + ..Default::default() + }; + output.resize(end, 0.0); + data.data_out = output[start..].as_mut_ptr(); + // SAFETY: state is exclusively owned; disjoint slices cover the declared frame counts. + let error = unsafe { sys::src_process(self.state.as_ptr(), &mut data) }; + let generated = data.output_frames_gen as usize * channels; + output.truncate(start + generated); + if error != 0 { + return Err(backend_error(self.config, error)); + } + Ok((data.input_frames_used as usize * channels, generated)) + } +} + +impl Drop for SincAudioResampler { + fn drop(&mut self) { + // SAFETY: this owner holds the only handle returned by src_new. + unsafe { sys::src_delete(self.state.as_ptr()) }; + } +} + +fn backend_error(config: AudioResamplerConfig, code: i32) -> AudioResamplerError { + AudioResamplerError::Backend(format!( + "input_rate={}, output_rate={}, channels={}: {:?}", + config.input_rate, + config.output_rate, + config.channels, + samplerate::Error::from_int(code) + )) +} diff --git a/src/audio_resampler/tests.rs b/src/audio_resampler/tests.rs new file mode 100644 index 000000000..566af6512 --- /dev/null +++ b/src/audio_resampler/tests.rs @@ -0,0 +1,178 @@ +use super::{AudioResampler, AudioResamplerConfig, FixedFrameAudioResampler}; + +const INPUT_RATE: u32 = 24_000; +const OUTPUT_RATE: u32 = 48_000; +const CHANNELS: u16 = 2; +const CHUNK_FRAMES: usize = 240; +const CHUNK_COUNT: usize = 4; +const TONE_FREQUENCY_HZ: f32 = 997.0; +const TONE_AMPLITUDE: f32 = 0.5; +const MAX_BOUNDARY_RESIDUAL: f32 = 0.02; +const LOOK_AHEAD_OUTPUT_FRAMES: usize = 1; +const UNEVEN_CHUNK_FRAMES: usize = 73; +const MONO_CHANNELS: u16 = 1; +const UNIT_RATE: u32 = 1; +const DOUBLE_RATE: u32 = 2; +const FIRST_DOWNSAMPLE_PACKET: [f32; 3] = [0.0, 1.0, 2.0]; +const SECOND_DOWNSAMPLE_PACKET: [f32; 4] = [3.0, 4.0, 5.0, 6.0]; +const EXPECTED_DOWNSAMPLED_OUTPUT: [f32; 4] = [0.0, 2.0, 4.0, 6.0]; +const PACKETS_PER_SECOND: usize = 100; +const OUTPUT_PACKET_FRAMES: usize = OUTPUT_RATE as usize / PACKETS_PER_SECOND; +const RATE_44_1_KHZ: u32 = 44_100; +const FLOAT_TOLERANCE: f32 = 0.000_001; +const MIN_CONTINUITY_PACKETS: usize = 2; + +fn stereo_tone_at_rate(frames: usize, sample_rate: u32) -> Vec { + (0..frames) + .flat_map(|frame| { + let phase = + std::f32::consts::TAU * TONE_FREQUENCY_HZ * frame as f32 / sample_rate as f32; + let sample = TONE_AMPLITUDE * phase.sin(); + [sample, sample] + }) + .collect() +} + +fn stereo_tone(frames: usize) -> Vec { + stereo_tone_at_rate(frames, INPUT_RATE) +} + +fn maximum_tone_prediction_residual(sample_rate: u32) -> f32 { + let half_step = std::f32::consts::PI * TONE_FREQUENCY_HZ / sample_rate as f32; + 4.0 * TONE_AMPLITUDE * half_step.sin().powi(2) +} + +fn maximum_boundary_residual(chunks: &[Vec]) -> f32 { + chunks.windows(2).fold(0.0, |maximum, pair| { + let previous = &pair[0]; + let current = &pair[1]; + let last = previous.len() - CHANNELS as usize; + let penultimate = last - CHANNELS as usize; + (0..CHANNELS as usize).fold(maximum, |maximum, channel| { + let predicted = previous[last + channel] + + (previous[last + channel] - previous[penultimate + channel]); + maximum.max((current[channel] - predicted).abs()) + }) + }) +} + +fn stereo_config() -> AudioResamplerConfig { + AudioResamplerConfig { + input_rate: INPUT_RATE, + output_rate: OUTPUT_RATE, + channels: CHANNELS, + } +} + +#[test] +fn preserves_decoded_packet_continuity_and_output_ratio() { + let input = stereo_tone(CHUNK_FRAMES * CHUNK_COUNT); + let mut whole_resampler = AudioResampler::new(stereo_config()).unwrap(); + let whole_output = whole_resampler.process(&input).unwrap(); + let expected_frames = CHUNK_FRAMES * CHUNK_COUNT * OUTPUT_RATE as usize / INPUT_RATE as usize + - LOOK_AHEAD_OUTPUT_FRAMES; + + for chunk_frames in [CHUNK_FRAMES, UNEVEN_CHUNK_FRAMES] { + let mut resampler = AudioResampler::new(stereo_config()).unwrap(); + let output: Vec<_> = input + .chunks(chunk_frames * CHANNELS as usize) + .map(|chunk| resampler.process(chunk).unwrap()) + .collect(); + let residual = maximum_boundary_residual(&output); + assert!( + residual <= MAX_BOUNDARY_RESIDUAL, + "packet boundary residual {residual} exceeded {MAX_BOUNDARY_RESIDUAL}, chunk_frames={chunk_frames}" + ); + let output_frames = output.iter().map(Vec::len).sum::() / CHANNELS as usize; + assert_eq!(output_frames, expected_frames); + assert_eq!(output.concat(), whole_output, "chunk_frames={chunk_frames}"); + } +} + +#[test] +fn rejects_incomplete_interleaved_frames() { + let mut resampler = AudioResampler::new(stereo_config()).unwrap(); + + assert!(resampler.process(&[TONE_AMPLITUDE]).is_err()); +} + +#[test] +fn interpolates_mono_samples() { + let config = AudioResamplerConfig { + input_rate: UNIT_RATE, + output_rate: DOUBLE_RATE, + channels: MONO_CHANNELS, + }; + let mut resampler = AudioResampler::new(config).unwrap(); + + assert_eq!( + resampler.process(&[0.0, 1.0, 2.0]).unwrap(), + [0.0, 0.5, 1.0, 1.5, 2.0] + ); +} + +#[test] +fn rejects_zero_rate_configuration() { + let config = AudioResamplerConfig { + input_rate: 0, + output_rate: OUTPUT_RATE, + channels: CHANNELS, + }; + + assert!(AudioResampler::new(config).is_err()); +} + +#[test] +fn downsamples_across_packet_boundaries() { + let config = AudioResamplerConfig { + input_rate: DOUBLE_RATE, + output_rate: UNIT_RATE, + channels: MONO_CHANNELS, + }; + let mut resampler = AudioResampler::new(config).unwrap(); + let mut output = resampler.process(&FIRST_DOWNSAMPLE_PACKET).unwrap(); + output.extend(resampler.process(&SECOND_DOWNSAMPLE_PACKET).unwrap()); + + assert_eq!(output, EXPECTED_DOWNSAMPLED_OUTPUT); +} + +#[test] +fn sender_resampler_emits_only_complete_continuous_frames() { + let input = stereo_tone(CHUNK_FRAMES * CHUNK_COUNT); + let mut resampler = + FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap(); + let output: Vec<_> = input + .chunks(CHUNK_FRAMES * CHANNELS as usize) + .flat_map(|chunk| resampler.process(chunk).unwrap()) + .collect(); + + assert!(output.len() >= MIN_CONTINUITY_PACKETS); + assert!(output + .iter() + .all(|packet| packet.len() == OUTPUT_PACKET_FRAMES * CHANNELS as usize)); + assert!(maximum_boundary_residual(&output) <= MAX_BOUNDARY_RESIDUAL); +} + +#[test] +fn sender_downsampling_preserves_packet_continuity() { + let input_packet_frames = RATE_44_1_KHZ as usize / PACKETS_PER_SECOND; + let input = stereo_tone_at_rate(input_packet_frames * CHUNK_COUNT, RATE_44_1_KHZ); + let config = AudioResamplerConfig { + input_rate: RATE_44_1_KHZ, + output_rate: INPUT_RATE, + channels: CHANNELS, + }; + let mut resampler = + FixedFrameAudioResampler::new(config, INPUT_RATE as usize / PACKETS_PER_SECOND).unwrap(); + let packets: Vec<_> = input + .chunks(input_packet_frames * CHANNELS as usize) + .flat_map(|packet| resampler.process(packet).unwrap()) + .collect(); + let residual = maximum_boundary_residual(&packets); + + assert_eq!(packets.len(), CHUNK_COUNT); + assert!( + residual <= maximum_tone_prediction_residual(INPUT_RATE) + FLOAT_TOLERANCE, + "sender packet boundary residual {residual} exceeded the tone curvature" + ); +} diff --git a/src/client.rs b/src/client.rs index 10330cfd6..156fc2e0c 100644 --- a/src/client.rs +++ b/src/client.rs @@ -93,6 +93,10 @@ use crate::ui_session_interface::SessionPermissionConfig; pub use super::lang::*; +#[cfg(not(target_os = "linux"))] +mod audio_playback; +#[cfg(all(test, not(target_os = "linux")))] +mod audio_state_tests; pub mod file_trait; pub mod helper; pub mod io_loop; @@ -2053,6 +2057,8 @@ pub struct AudioHandler { simple: Option, #[cfg(not(target_os = "linux"))] audio_buffer: AudioBuffer, + #[cfg(not(target_os = "linux"))] + audio_resampler: Option, sample_rate: (u32, u32), #[cfg(not(target_os = "linux"))] audio_stream: Option>, @@ -2060,7 +2066,55 @@ pub struct AudioHandler { #[cfg(not(target_os = "linux"))] device_channel: u16, #[cfg(not(target_os = "linux"))] - ready: Arc>, + playback_status: Arc, +} + +#[cfg(not(target_os = "linux"))] +#[derive(Clone, Copy)] +struct DecodedAudioConfig { + sample_rate: u32, + input_channels: u16, + output_channels: u16, +} + +#[cfg(not(target_os = "linux"))] +fn create_audio_resampler( + input_rate: u32, + output_rate: u32, + channels: u16, +) -> ResultType> { + if input_rate == output_rate { + return Ok(None); + } + Ok(Some(crate::audio_resampler::AudioResampler::new( + crate::audio_resampler::AudioResamplerConfig { + input_rate, + output_rate, + channels, + }, + )?)) +} + +#[cfg(not(target_os = "linux"))] +fn prepare_decoded_audio( + input: &[f32], + resampler: Option<&mut crate::audio_resampler::AudioResampler>, + config: DecodedAudioConfig, +) -> Result, crate::audio_resampler::AudioResamplerError> { + let mut output = match resampler { + Some(resampler) => resampler.process(input)?, + None => input.to_owned(), + }; + if config.input_channels != config.output_channels { + output = crate::audio_rechannel( + output, + config.sample_rate, + config.sample_rate, + config.input_channels, + config.output_channels, + ); + } + Ok(output) } #[cfg(not(target_os = "linux"))] @@ -2068,6 +2122,7 @@ struct AudioBuffer( pub Arc>>, usize, [usize; 30], + Arc, ); #[cfg(not(target_os = "linux"))] @@ -2079,6 +2134,7 @@ impl Default for AudioBuffer { )), 48000 * 2, [0; 30], + Arc::new(std::sync::atomic::AtomicUsize::new(0)), ) } } @@ -2153,27 +2209,36 @@ impl AudioBuffer { let skip = (cap * max / (30 * N) + 1) & (!1); if (having > skip * 3) && (skip > 0) { lock.skip(skip); - log::info!("skip {skip}, based {max} {zero}"); + let generation = self.signal_discontinuity(); + drop(lock); + log::info!("skip {skip}, based {max} {zero}, generation={generation}"); } } + /// The caller must hold the PCM buffer lock while signaling the discard. + fn signal_discontinuity(&self) -> usize { + self.3 + .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + .wrapping_add(1) + } + /// append pcm to audio buffer, if buffered data /// exceeds AUDIO_BUFFER_MS, only AUDIO_BUFFER_MS /// will be kept. fn append_pcm2(&self, buffer: &[f32]) -> usize { let mut lock = self.0.lock().unwrap(); let cap = lock.capacity(); - if buffer.len() > cap { - lock.push_slice_overwrite(buffer); - return cap; - } - let having = lock.occupied_len() + buffer.len(); - if having > cap { - lock.skip(having - cap); - } lock.push_slice_overwrite(buffer); - lock.occupied_len() + let discard = (having > cap).then(|| (having - cap, self.signal_discontinuity())); + let occupied = lock.occupied_len(); + drop(lock); + if let Some((discarded, generation)) = discard { + log::debug!( + "Audio buffer capacity discard: samples={discarded}, generation={generation}" + ); + } + occupied } /// append pcm to audio buffer, trying to drop data @@ -2185,6 +2250,41 @@ impl AudioBuffer { } } +#[cfg(all(test, not(target_os = "linux")))] +mod audio_buffer_discontinuity_tests { + use super::AudioBuffer; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }; + + const BUFFER_CAPACITY: usize = 4; + const BUFFER_LEVELS: usize = 30; + const FIRST_INPUT: [f32; 2] = [0.1, 0.2]; + const OVERFLOWING_INPUT: [f32; 3] = [0.3, 0.4, 0.5]; + const OVERSIZED_INPUT: [f32; 5] = [0.6, 0.7, 0.8, 0.9, 1.0]; + + #[test] + fn capacity_discards_signal_discontinuities() { + let audio_buffer = AudioBuffer( + Arc::new(Mutex::new(ringbuf::HeapRb::new(BUFFER_CAPACITY))), + BUFFER_CAPACITY, + [0; BUFFER_LEVELS], + Arc::new(AtomicUsize::new(0)), + ); + + assert_eq!(audio_buffer.append_pcm2(&FIRST_INPUT), FIRST_INPUT.len()); + assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 0); + assert_eq!( + audio_buffer.append_pcm2(&OVERFLOWING_INPUT), + BUFFER_CAPACITY + ); + assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 1); + assert_eq!(audio_buffer.append_pcm2(&OVERSIZED_INPUT), BUFFER_CAPACITY); + assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 2); + } +} + impl AudioHandler { #[cfg(target_os = "linux")] fn start_audio(&mut self, format0: AudioFormat) -> ResultType<()> { @@ -2238,6 +2338,9 @@ impl AudioHandler { } self.sample_rate = (format0.sample_rate, config.sample_rate.0); + let audio_resampler = create_audio_resampler( + format0.sample_rate, config.sample_rate.0, format0.channels as _, + )?; let mut build_output_stream = |config: StreamConfig| match sample_format { cpal::SampleFormat::I8 => self.build_output_stream::(&config, &device), cpal::SampleFormat::I16 => self.build_output_stream::(&config, &device), @@ -2262,6 +2365,7 @@ impl AudioHandler { } else { build_output_stream(config)?; } + self.audio_resampler = audio_resampler; Ok(()) } @@ -2274,10 +2378,17 @@ impl AudioHandler { } match AudioDecoder::new(f.sample_rate, if f.channels > 1 { Stereo } else { Mono }) { Ok(d) => { + #[cfg(target_os = "linux")] + let keep_existing_stream = self.simple.is_some() + && self.sample_rate.0 == f.sample_rate + && u32::from(self.channels) == f.channels; + #[cfg(not(target_os = "linux"))] + let keep_existing_stream = false; let buffer = vec![0.; f.sample_rate as usize * f.channels as usize]; self.audio_decoder = Some((d, buffer)); self.channels = f.channels as _; - allow_err!(self.start_audio(f)); + let result = self.start_audio(f); + self.handle_audio_start_result(result, keep_existing_stream); } Err(err) => { log::error!("Failed to create audio decoder: {}", err); @@ -2285,11 +2396,31 @@ impl AudioHandler { } } + fn handle_audio_start_result(&mut self, result: ResultType<()>, keep_existing_stream: bool) { + if let Err(error) = result { + if keep_existing_stream { + log::error!( + "Failed to replace audio playback stream; keeping the existing compatible stream: {error:#}" + ); + } else { + *self = Self::default(); + log::error!("Failed to start audio playback: {error:#}"); + } + } + } + /// Handle audio frame and play it. #[inline] pub fn handle_frame(&mut self, frame: AudioFrame) { #[cfg(not(target_os = "linux"))] - if self.audio_stream.is_none() || !self.ready.lock().unwrap().clone() { + self.playback_status.report_errors(); + #[cfg(not(target_os = "linux"))] + if self.audio_stream.is_none() + || !self + .playback_status + .ready + .load(std::sync::atomic::Ordering::Acquire) + { return; } #[cfg(target_os = "linux")] @@ -2298,39 +2429,40 @@ impl AudioHandler { return; } self.audio_decoder.as_mut().map(|(d, buffer)| { - if let Ok(n) = d.decode_float(&frame.data, buffer, false) { - let channels = self.channels; - let n = n * (channels as usize); - #[cfg(not(target_os = "linux"))] - { - let sample_rate0 = self.sample_rate.0; - let sample_rate = self.sample_rate.1; - let mut buffer = buffer[0..n].to_owned(); - if sample_rate != sample_rate0 { - buffer = crate::audio_resample( - &buffer[0..n], - sample_rate0, - sample_rate, - channels, - ); - } - if self.channels != self.device_channel { - buffer = crate::audio_rechannel( - buffer, - sample_rate, - sample_rate, - self.channels, - self.device_channel, - ); - } - self.audio_buffer.append_pcm(&buffer); - } - #[cfg(target_os = "linux")] - { - let data_u8 = - unsafe { std::slice::from_raw_parts::(buffer.as_ptr() as _, n * 4) }; - self.simple.as_mut().map(|x| x.write(data_u8)); + let decoded_frames = match d.decode_float(&frame.data, buffer, false) { + Ok(decoded_frames) => decoded_frames, + Err(error) => { + log::warn!("Failed to decode audio frame: {error:?}"); + return; } + }; + let channels = self.channels; + let n = decoded_frames * channels as usize; + #[cfg(not(target_os = "linux"))] + { + let config = DecodedAudioConfig { + sample_rate: self.sample_rate.1, + input_channels: self.channels, + output_channels: self.device_channel, + }; + let buffer = match prepare_decoded_audio( + &buffer[0..n], + self.audio_resampler.as_mut(), + config, + ) { + Ok(output) => output, + Err(error) => { + log::error!("Failed to resample decoded audio: {error:#}"); + return; + } + }; + self.audio_buffer.append_pcm(&buffer); + } + #[cfg(target_os = "linux")] + { + let data_u8 = + unsafe { std::slice::from_raw_parts::(buffer.as_ptr() as _, n * 4) }; + self.simple.as_mut().map(|x| x.write(data_u8)); } }); } @@ -2350,63 +2482,28 @@ impl AudioHandler { self.audio_buffer .resize(config.sample_rate.0 as _, config.channels as _); let audio_buffer = self.audio_buffer.0.clone(); - let ready = self.ready.clone(); + let discontinuity_generation = self.audio_buffer.3.clone(); + let mut playback_writer = audio_playback::AudioPlaybackWriter::new( + audio_playback::AudioPlaybackConfig { + sample_rate: config.sample_rate.0, + channels: config.channels as usize, + }, + audio_buffer, + discontinuity_generation, + )?; + let playback_status = playback_writer.status.clone(); let timeout = None; let stream = device.build_output_stream( config, - move |data: &mut [T], info: &cpal::OutputCallbackInfo| { - if !*ready.lock().unwrap() { - *ready.lock().unwrap() = true; - } - - let mut n = data.len(); - let mut lock = audio_buffer.lock().unwrap(); - let mut having = lock.occupied_len(); - // android two timestamps, one from zero, another not - #[cfg(not(target_os = "android"))] - if having < n { - let tms = info.timestamp(); - let how_long = tms - .playback - .duration_since(&tms.callback) - .unwrap_or(Duration::from_millis(0)); - - // must long enough to fight back scheuler delay - if how_long > Duration::from_millis(6) && how_long < Duration::from_millis(3000) - { - drop(lock); - std::thread::sleep(how_long.div_f32(1.2)); - lock = audio_buffer.lock().unwrap(); - having = lock.occupied_len(); - } - - if having < n { - n = having; - } - } - #[cfg(target_os = "android")] - if having < n { - n = having; - } - let mut elems = vec![0.0f32; n]; - if n > 0 { - lock.pop_slice(&mut elems); - } - drop(lock); - - let mut input = elems.into_iter(); - for sample in data.iter_mut() { - *sample = match input.next() { - Some(x) => T::from_sample(x), - _ => T::from_sample(0.), - }; - } + move |data: &mut [T], _: &cpal::OutputCallbackInfo| { + playback_writer.write_output(data); }, err_fn, timeout, )?; stream.play()?; self.audio_stream = Some(Box::new(stream)); + self.playback_status = playback_status; Ok(()) } } @@ -2426,6 +2523,27 @@ mod audio_format_tests { assert!(!is_supported_audio_channel_count(0)); assert!(!is_supported_audio_channel_count(u32::MAX)); } + + #[test] + fn failed_audio_start_discards_format_state() { + use super::{anyhow, AudioDecoder, AudioHandler, Stereo}; + + const SAMPLE_RATE: u32 = 48_000; + const CHANNELS: u16 = 2; + let decoder = AudioDecoder::new(SAMPLE_RATE, Stereo).unwrap(); + let mut handler = AudioHandler { + audio_decoder: Some((decoder, Vec::new())), + sample_rate: (SAMPLE_RATE, SAMPLE_RATE), + channels: CHANNELS, + ..Default::default() + }; + + handler.handle_audio_start_result(Err(anyhow!("Injected playback startup failure")), false); + + assert!(handler.audio_decoder.is_none()); + assert_eq!(handler.channels, 0); + assert_eq!(handler.sample_rate, (0, 0)); + } } /// Video handler for the [`Client`]. diff --git a/src/client/audio_playback.rs b/src/client/audio_playback.rs new file mode 100644 index 000000000..8e2cb0894 --- /dev/null +++ b/src/client/audio_playback.rs @@ -0,0 +1,218 @@ +use hbb_common::{log, thiserror}; +use ringbuf::{ring_buffer::RbBase, Rb}; +use std::sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + TryLockError, +}; + +pub(super) const UNDERRUN_DECLICK_MS: usize = 5; +const MILLISECONDS_PER_SECOND: usize = 1_000; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct AudioPlaybackConfig { + pub sample_rate: u32, + pub channels: usize, +} + +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub(super) enum AudioPlaybackError { + #[error( + "invalid audio playback configuration: sample_rate={}, channels={}", + .0.sample_rate, .0.channels + )] + InvalidConfig(AudioPlaybackConfig), + #[error("audio playback frame has {samples} samples for {channels} channels")] + IncompleteFrame { samples: usize, channels: usize }, + #[error("audio playback transition frame count overflow")] + FrameCountOverflow, +} + +pub(super) struct AudioPlaybackRecovery { + channels: usize, + transition_frames: usize, + transition_frame: usize, + had_input: bool, + transition_start: Vec, + output_frame: Vec, +} + +#[derive(Default)] +pub(super) struct AudioPlaybackStatus { + pub(super) ready: AtomicBool, + contentions: AtomicUsize, + buffer_poisoned: AtomicBool, +} + +impl AudioPlaybackStatus { + pub(super) fn report_errors(&self) { + let contentions = self.contentions.swap(0, Ordering::Relaxed); + if contentions != 0 { + log::debug!("Audio playback PCM buffer contention: callbacks={contentions}"); + } + if self.buffer_poisoned.swap(false, Ordering::Relaxed) { + log::error!("Audio playback stopped reading a poisoned PCM buffer"); + } + } +} + +pub(super) struct AudioPlaybackWriter { + audio_buffer: std::sync::Arc>>, + discontinuity_generation: std::sync::Arc, + observed_discontinuity_generation: usize, + buffered_input: Vec, + recovery: AudioPlaybackRecovery, + pub(super) status: std::sync::Arc, + buffer_failed: bool, +} + +impl AudioPlaybackWriter { + pub(super) fn new( + config: AudioPlaybackConfig, + audio_buffer: std::sync::Arc>>, + discontinuity_generation: std::sync::Arc, + ) -> Result { + let recovery = AudioPlaybackRecovery::new(config)?; + let buffer_capacity = audio_buffer.lock().unwrap().capacity(); + let observed_discontinuity_generation = discontinuity_generation.load(Ordering::Relaxed); + Ok(Self { + audio_buffer, + discontinuity_generation, + observed_discontinuity_generation, + buffered_input: vec![0.0; buffer_capacity], + recovery, + status: Default::default(), + buffer_failed: false, + }) + } + + fn read_buffer(&mut self, requested_samples: usize) -> usize { + if self.buffer_failed { + return 0; + } + let mut buffer = match self.audio_buffer.try_lock() { + Ok(buffer) => buffer, + Err(TryLockError::WouldBlock) => { + // Keep queued PCM and its generation for the next successful read. + self.status.contentions.fetch_add(1, Ordering::Relaxed); + return 0; + } + Err(TryLockError::Poisoned(_)) => { + self.buffer_failed = true; + self.status.ready.store(false, Ordering::Release); + self.status.buffer_poisoned.store(true, Ordering::Relaxed); + return 0; + } + }; + let generation = self.discontinuity_generation.load(Ordering::Relaxed); + let channels = self.recovery.channels; + let samples = buffer.occupied_len().min(requested_samples) / channels * channels; + buffer.pop_slice(&mut self.buffered_input[..samples]); + drop(buffer); + if generation != self.observed_discontinuity_generation { + self.recovery.begin_discontinuity(); + self.observed_discontinuity_generation = generation; + } + samples + } + + pub(super) fn write_output(&mut self, output: &mut [T]) + where + T: cpal::Sample + cpal::FromSample, + { + self.status + .ready + .store(!self.buffer_failed, Ordering::Release); + let requested_samples = output.len().min(self.buffered_input.len()); + let channel_count = self.recovery.channels; + let available_samples = self.read_buffer(requested_samples); + let available_frames = available_samples / channel_count; + for (frame_index, output_frame) in output.chunks_mut(channel_count).enumerate() { + let input = if frame_index < available_frames { + let start = frame_index * channel_count; + Some(&self.buffered_input[start..start + channel_count]) + } else { + None + }; + match self.recovery.process_frame(input) { + Ok(recovered) => { + for (output, sample) in output_frame.iter_mut().zip(recovered) { + *output = T::from_sample(*sample); + } + } + Err(error) => { + log::error!("Failed to recover audio underflow: {error}"); + output_frame.fill(T::from_sample(0.0)); + } + } + } + } +} + +impl AudioPlaybackRecovery { + pub(super) fn new(config: AudioPlaybackConfig) -> Result { + if config.sample_rate == 0 || config.channels == 0 { + return Err(AudioPlaybackError::InvalidConfig(config)); + } + let transition_frames = (config.sample_rate as usize) + .checked_mul(UNDERRUN_DECLICK_MS) + .ok_or(AudioPlaybackError::FrameCountOverflow)? + / MILLISECONDS_PER_SECOND; + if transition_frames == 0 { + return Err(AudioPlaybackError::InvalidConfig(config)); + } + Ok(Self { + channels: config.channels, + transition_frames, + transition_frame: transition_frames, + had_input: false, + transition_start: vec![0.0; config.channels], + output_frame: vec![0.0; config.channels], + }) + } + + pub(super) fn process_frame( + &mut self, + input: Option<&[f32]>, + ) -> Result<&[f32], AudioPlaybackError> { + if input.is_some_and(|frame| frame.len() != self.channels) { + return Err(AudioPlaybackError::IncompleteFrame { + samples: input.map_or(0, <[f32]>::len), + channels: self.channels, + }); + } + self.begin_transition(input.is_some()); + let target_weight = self.advance_transition(); + for channel in 0..self.channels { + let target = input.map_or(0.0, |frame| frame[channel]); + self.output_frame[channel] = + self.transition_start[channel] * (1.0 - target_weight) + target * target_weight; + } + Ok(&self.output_frame) + } + + pub(super) fn begin_discontinuity(&mut self) { + self.transition_start.copy_from_slice(&self.output_frame); + self.transition_frame = 0; + } + + fn begin_transition(&mut self, has_input: bool) { + if has_input == self.had_input { + return; + } + self.transition_start.copy_from_slice(&self.output_frame); + self.transition_frame = 0; + self.had_input = has_input; + } + + fn advance_transition(&mut self) -> f32 { + if self.transition_frame >= self.transition_frames { + return 1.0; + } + self.transition_frame += 1; + self.transition_frame as f32 / self.transition_frames as f32 + } +} + +#[cfg(test)] +#[path = "audio_playback_tests.rs"] +mod tests; diff --git a/src/client/audio_playback_tests.rs b/src/client/audio_playback_tests.rs new file mode 100644 index 000000000..ce528a732 --- /dev/null +++ b/src/client/audio_playback_tests.rs @@ -0,0 +1,218 @@ +use super::{AudioPlaybackConfig, AudioPlaybackError, AudioPlaybackRecovery, AudioPlaybackWriter}; +use ringbuf::{ring_buffer::RbBase, Rb}; +use std::{ + sync::{atomic::Ordering, mpsc, Arc, Mutex}, + time::Duration, +}; + +const SAMPLE_RATE: u32 = 48_000; +const CHANNELS: usize = 2; +const ACTIVE_FRAME: [f32; CHANNELS] = [0.8, -0.8]; +const OPPOSITE_ACTIVE_FRAME: [f32; CHANNELS] = [-0.8, 0.8]; +const ACTIVE_FRAMES: usize = 300; +const SILENT_FRAMES: usize = 300; +const TRANSITION_FRAMES: usize = + SAMPLE_RATE as usize * super::UNDERRUN_DECLICK_MS / super::MILLISECONDS_PER_SECOND; +const MAX_SAMPLE_STEP: f32 = 0.01; + +#[test] +fn writing_audio_observes_discard_and_releases_buffer_lock() { + const INPUT: [f32; 4] = [0.1, 0.2, 0.3, 0.4]; + const GENERATION: usize = 7; + let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(INPUT.len()))); + let generation = Arc::new(super::AtomicUsize::new(0)); + let config = AudioPlaybackConfig { + sample_rate: SAMPLE_RATE, + channels: CHANNELS, + }; + let mut writer = AudioPlaybackWriter::new(config, buffer.clone(), generation.clone()).unwrap(); + { + let mut buffer = buffer.lock().unwrap(); + buffer.push_slice(&INPUT); + generation.store(GENERATION, Ordering::Relaxed); + } + let mut output = [0.0_f32; INPUT.len()]; + + writer.write_output(&mut output); + + assert_eq!(writer.buffered_input, INPUT); + assert_eq!(writer.observed_discontinuity_generation, GENERATION); + assert_eq!(buffer.try_lock().unwrap().occupied_len(), 0); +} + +fn maximum_sample_step(samples: &[f32]) -> f32 { + samples + .windows(CHANNELS + 1) + .map(|window| (window[CHANNELS] - window[0]).abs()) + .fold(0.0, f32::max) +} + +#[test] +fn smooths_underflow_and_explicit_audio_discontinuities() { + for explicit_discontinuity in [false, true] { + let config = AudioPlaybackConfig { + sample_rate: SAMPLE_RATE, + channels: CHANNELS, + }; + let mut recovery = AudioPlaybackRecovery::new(config).unwrap(); + let mut output = Vec::new(); + for _ in 0..ACTIVE_FRAMES { + output.extend_from_slice(recovery.process_frame(Some(&ACTIVE_FRAME)).unwrap()); + } + let transition_end = TRANSITION_FRAMES * CHANNELS; + assert_eq!( + &output[transition_end - CHANNELS..transition_end], + ACTIVE_FRAME.as_slice(), + "explicit_discontinuity={explicit_discontinuity}" + ); + let resumed_frame = if explicit_discontinuity { + recovery.begin_discontinuity(); + &OPPOSITE_ACTIVE_FRAME + } else { + for _ in 0..SILENT_FRAMES { + output.extend_from_slice(recovery.process_frame(None).unwrap()); + } + &ACTIVE_FRAME + }; + for _ in 0..ACTIVE_FRAMES { + output.extend_from_slice(recovery.process_frame(Some(resumed_frame)).unwrap()); + } + let maximum = maximum_sample_step(&output); + assert!( + maximum <= MAX_SAMPLE_STEP, + "step {maximum} exceeded {MAX_SAMPLE_STEP}, explicit={explicit_discontinuity}" + ); + assert_eq!( + &output[output.len() - CHANNELS..], + resumed_frame, + "explicit_discontinuity={explicit_discontinuity}" + ); + } +} + +#[test] +fn validates_configuration_and_frame_size() { + let invalid_config = AudioPlaybackConfig { + sample_rate: 0, + channels: CHANNELS, + }; + assert_eq!( + AudioPlaybackRecovery::new(invalid_config).err(), + Some(AudioPlaybackError::InvalidConfig(invalid_config)) + ); + + let config = AudioPlaybackConfig { + sample_rate: SAMPLE_RATE, + channels: CHANNELS, + }; + let mut recovery = AudioPlaybackRecovery::new(config).unwrap(); + assert_eq!( + recovery.process_frame(Some(&[0.5])).err(), + Some(AudioPlaybackError::IncompleteFrame { + samples: 1, + channels: CHANNELS, + }) + ); +} + +const CALLBACK_SAMPLES: usize = 64; +const CALLBACK_TIMEOUT: Duration = Duration::from_secs(2); +const DISCARD_GENERATION: usize = 1; + +fn write_while_buffer_is_locked( + mut writer: AudioPlaybackWriter, + buffer: &Arc>>, + generation: &Arc, +) -> (AudioPlaybackWriter, [f32; CALLBACK_SAMPLES]) { + let mut guard = buffer.lock().unwrap(); + let queued = OPPOSITE_ACTIVE_FRAME.repeat(ACTIVE_FRAMES); + guard.push_slice(&queued); + generation.store(DISCARD_GENERATION, Ordering::Relaxed); + let (completed_tx, completed_rx) = mpsc::channel(); + let callback = std::thread::spawn(move || { + let mut output = [0.0; CALLBACK_SAMPLES]; + crate::audio_resampler::allocation_tests::assert_no_allocations(|| { + writer.write_output(&mut output); + }); + completed_tx.send((writer, output)).unwrap(); + }); + let completed = completed_rx.recv_timeout(CALLBACK_TIMEOUT); + let retained = guard.occupied_len(); + drop(guard); + callback.join().unwrap(); + let result = completed.expect("playback callback waited for the buffer owner"); + assert_eq!(retained, queued.len()); + result +} + +#[test] +fn playback_contention_preserves_queued_audio_and_recovers_after_release() { + let samples = ACTIVE_FRAMES * CHANNELS; + let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(samples))); + let generation = Arc::new(super::AtomicUsize::new(0)); + let config = AudioPlaybackConfig { + sample_rate: SAMPLE_RATE, + channels: CHANNELS, + }; + let mut writer = AudioPlaybackWriter::new(config, buffer.clone(), generation.clone()).unwrap(); + buffer + .lock() + .unwrap() + .push_slice(&ACTIVE_FRAME.repeat(ACTIVE_FRAMES)); + let mut output = vec![0.0; samples]; + writer.write_output(&mut output); + assert_eq!(&output[samples - CHANNELS..], &ACTIVE_FRAME); + + let (mut writer, gap) = write_while_buffer_is_locked(writer, &buffer, &generation); + + assert_eq!(writer.status.contentions.load(Ordering::Relaxed), 1); + assert!(writer.status.ready.load(Ordering::Acquire)); + assert_eq!(writer.observed_discontinuity_generation, 0); + assert!(maximum_sample_step(&gap) <= MAX_SAMPLE_STEP); + assert!(gap[0] > 0.0 && gap[0] < ACTIVE_FRAME[0]); + assert_eq!(gap[1], -gap[0]); + assert!(gap[CALLBACK_SAMPLES - CHANNELS] > 0.0); + writer.write_output(&mut output); + let mut transition = gap[gap.len() - CHANNELS..].to_vec(); + transition.extend_from_slice(&output[..TRANSITION_FRAMES * CHANNELS]); + assert!(maximum_sample_step(&transition) <= MAX_SAMPLE_STEP); + assert_eq!( + writer.buffered_input, + OPPOSITE_ACTIVE_FRAME.repeat(ACTIVE_FRAMES) + ); + assert_eq!(writer.observed_discontinuity_generation, DISCARD_GENERATION); + assert_eq!(&output[samples - CHANNELS..], &OPPOSITE_ACTIVE_FRAME); + assert_eq!(buffer.lock().unwrap().occupied_len(), 0); +} + +#[test] +fn poisoned_playback_buffer_reports_once_without_panicking_in_the_callback() { + let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(CALLBACK_SAMPLES))); + let config = AudioPlaybackConfig { + sample_rate: SAMPLE_RATE, + channels: CHANNELS, + }; + let mut writer = + AudioPlaybackWriter::new(config, buffer.clone(), Arc::new(super::AtomicUsize::new(0))) + .unwrap(); + assert!(std::thread::spawn(move || { + let _guard = buffer.lock().unwrap(); + panic!("Injected PCM buffer failure"); + }) + .join() + .is_err()); + let mut output = [ACTIVE_FRAME[0]; CALLBACK_SAMPLES]; + + crate::audio_resampler::allocation_tests::assert_no_allocations(|| { + writer.write_output(&mut output); + }); + + assert_eq!(output, [0.0; CALLBACK_SAMPLES]); + assert!(!writer.status.ready.load(Ordering::Acquire)); + assert!(writer.status.buffer_poisoned.load(Ordering::Relaxed)); + writer.status.report_errors(); + writer.write_output(&mut output); + assert!(!writer.status.buffer_poisoned.load(Ordering::Relaxed)); + assert_eq!(writer.status.contentions.load(Ordering::Relaxed), 0); + assert!(!writer.status.ready.load(Ordering::Acquire)); +} diff --git a/src/client/audio_state_tests.rs b/src/client/audio_state_tests.rs new file mode 100644 index 000000000..3c2924f5e --- /dev/null +++ b/src/client/audio_state_tests.rs @@ -0,0 +1,113 @@ +use super::{create_audio_resampler, AudioDecoder, AudioFrame, AudioHandler, Stereo}; +use cpal::traits::StreamTrait; +use hbb_common::anyhow::anyhow; +use magnum_opus::{Application::LowDelay, Encoder}; +use ringbuf::{ring_buffer::RbBase, Rb}; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; + +const INPUT_RATE: u32 = 24_000; +const OUTPUT_RATE: u32 = 48_000; +const CHANNELS: u16 = 2; +const PACKETS_PER_SECOND: usize = 100; +const MAX_PACKET_BYTES: usize = 4_096; +const SAMPLE_VALUE: f32 = 0.25; + +struct TrackedAudioStream(Arc); + +impl StreamTrait for TrackedAudioStream { + fn play(&self) -> Result<(), cpal::PlayStreamError> { + Ok(()) + } + + fn pause(&self) -> Result<(), cpal::PauseStreamError> { + Ok(()) + } +} + +impl Drop for TrackedAudioStream { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } +} + +fn decoder(sample_rate: u32) -> (AudioDecoder, Vec) { + ( + AudioDecoder::new(sample_rate, Stereo).unwrap(), + vec![0.0; sample_rate as usize * CHANNELS as usize], + ) +} + +fn active_handler(input_rate: u32) -> (AudioHandler, Arc) { + let dropped = Arc::new(AtomicBool::new(false)); + let handler = AudioHandler { + audio_decoder: Some(decoder(input_rate)), + audio_resampler: create_audio_resampler(input_rate, OUTPUT_RATE, CHANNELS).unwrap(), + sample_rate: (input_rate, OUTPUT_RATE), + audio_stream: Some(Box::new(TrackedAudioStream(dropped.clone()))), + channels: CHANNELS, + device_channel: CHANNELS, + ..Default::default() + }; + handler.playback_status.ready.store(true, Ordering::Release); + (handler, dropped) +} + +fn audio_frame() -> AudioFrame { + let samples = OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize; + let mut encoder = Encoder::new(OUTPUT_RATE, Stereo, LowDelay).unwrap(); + AudioFrame { + data: encoder + .encode_vec_float(&vec![SAMPLE_VALUE; samples], MAX_PACKET_BYTES) + .unwrap() + .into(), + ..Default::default() + } +} + +#[test] +fn failed_format_change_discards_old_playback_state() { + let (mut handler, dropped) = active_handler(INPUT_RATE); + handler + .audio_buffer + .0 + .lock() + .unwrap() + .push_slice(&[SAMPLE_VALUE; CHANNELS as usize]); + handler.audio_decoder = Some(decoder(OUTPUT_RATE)); + handler.sample_rate = (OUTPUT_RATE, OUTPUT_RATE); + + handler.handle_audio_start_result( + Err(anyhow!("Injected output stream startup failure")), + false, + ); + + assert!(dropped.load(Ordering::SeqCst)); + assert!(handler.audio_stream.is_none()); + assert!(handler.audio_resampler.is_none()); + assert!(handler.audio_decoder.is_none()); + assert!(!handler.playback_status.ready.load(Ordering::Acquire)); + handler.handle_frame(audio_frame()); + assert_eq!(handler.audio_buffer.0.lock().unwrap().occupied_len(), 0); +} + +#[test] +fn successful_start_or_compatible_failure_preserves_audio_packet_duration() { + for result in [ + Ok(()), + Err(anyhow!("Injected compatible stream replacement failure")), + ] { + let (mut handler, dropped) = active_handler(OUTPUT_RATE); + + handler.handle_audio_start_result(result, true); + handler.handle_frame(audio_frame()); + + assert!(!dropped.load(Ordering::SeqCst)); + assert_eq!( + handler.audio_buffer.0.lock().unwrap().occupied_len(), + OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize + ); + } +} diff --git a/src/common.rs b/src/common.rs index 00a409a63..9dadb1c86 100644 --- a/src/common.rs +++ b/src/common.rs @@ -408,6 +408,11 @@ pub fn resample_channels( } } +#[cfg(all(feature = "use_dasp", feature = "use_samplerate"))] +compile_error!( + "features `use_dasp` and `use_samplerate` are mutually exclusive; disable default features before selecting `use_samplerate`" +); + #[cfg(feature = "use_dasp")] pub fn audio_resample( data: &[f32], @@ -444,7 +449,7 @@ pub fn audio_resample( } } -#[cfg(feature = "use_samplerate")] +#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))] pub fn audio_resample( data: &[f32], sample_rate0: u32, diff --git a/src/lib.rs b/src/lib.rs index 1bfe2698c..649f3ef95 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,5 @@ +#[cfg(any(test, not(target_os = "linux")))] +mod audio_resampler; mod keyboard; /// cbindgen:ignore pub mod platform; diff --git a/src/server/audio_service.rs b/src/server/audio_service.rs index 52f2eff2e..4f17115ea 100644 --- a/src/server/audio_service.rs +++ b/src/server/audio_service.rs @@ -15,7 +15,9 @@ use super::*; #[cfg(not(any(target_os = "linux", target_os = "android")))] use hbb_common::anyhow::anyhow; -use magnum_opus::{Application::*, Channels::*, Encoder}; +#[cfg(any(target_os = "linux", target_os = "android"))] +use magnum_opus::Application::LowDelay; +use magnum_opus::{Channels::*, Encoder}; use std::sync::atomic::{AtomicBool, Ordering}; pub const NAME: &'static str = "audio"; @@ -97,10 +99,11 @@ mod pa_impl { RESTARTING.store(false, Ordering::SeqCst); #[cfg(target_os = "linux")] let mut stream = crate::ipc::connect(1000, "_pa").await?; - unsafe { - AUDIO_ZERO_COUNT = 0; - } - let mut encoder = Encoder::new(crate::platform::PA_SAMPLE_RATE, Stereo, LowDelay)?; + let mut encoder = AudioEncoder::new(Encoder::new( + crate::platform::PA_SAMPLE_RATE, + Stereo, + LowDelay, + )?); #[cfg(target_os = "linux")] allow_err!( stream @@ -172,8 +175,11 @@ pub fn is_screen_capture_kit_available() -> bool { } #[cfg(not(any(target_os = "linux", target_os = "android")))] -#[path = "audio_capture_error.rs"] +mod audio_capture; +#[cfg(not(any(target_os = "linux", target_os = "android")))] mod audio_capture_error; +#[cfg(not(any(target_os = "linux", target_os = "android")))] +mod audio_capture_queue; #[cfg(not(any(target_os = "linux", target_os = "android")))] mod cpal_impl { @@ -182,14 +188,15 @@ mod cpal_impl { use super::*; use cpal::{ traits::{DeviceTrait, HostTrait, StreamTrait}, - BufferSize, Device, Host, InputCallbackInfo, StreamConfig, SupportedStreamConfig, + Device, Host, InputCallbackInfo, SupportedStreamConfig, }; lazy_static::lazy_static! { static ref HOST: Host = cpal::default_host(); - static ref INPUT_BUFFER: Arc>> = Default::default(); } + const AUDIO_PACKETS_PER_SECOND: usize = 100; + #[cfg(feature = "screencapturekit")] lazy_static::lazy_static! { static ref HOST_SCREEN_CAPTURE_KIT: Result = cpal::host_from_id(cpal::HostId::ScreenCaptureKit); @@ -197,7 +204,20 @@ mod cpal_impl { #[derive(Default)] pub struct State { - stream: Option<(Box, Arc, CaptureErrorHandler)>, + stream: Option, + } + + struct ActiveCaptureStream { + stream: Option>, + format: Arc, + _encoder_worker: audio_capture_queue::CaptureEncoderWorker, + errors: CaptureErrorHandler, + } + + impl Drop for ActiveCaptureStream { + fn drop(&mut self) { + self.stream.take(); + } } impl super::service::Reset for State { @@ -215,8 +235,8 @@ mod cpal_impl { } _ => {} } - if let Some((_, format, _)) = &state.stream { - sp.send_shared(format.clone()); + if let Some(stream) = &state.stream { + sp.send_shared(stream.format.clone()); #[cfg(target_os = "macos")] log::info!("Audio capture stream recreated; replacement format sent"); } @@ -232,8 +252,8 @@ mod cpal_impl { } _ => {} } - if let Some((_, format, _)) = &state.stream { - sps.send_shared(format.clone()); + if let Some(stream) = &state.stream { + sps.send_shared(stream.format.clone()); } Ok(()) })?; @@ -241,10 +261,10 @@ mod cpal_impl { } pub fn run(sp: EmptyExtraFieldService, state: &mut State) -> ResultType<()> { - if let Some((_, _, errors)) = &state.stream { - if errors.needs_restart() { - // Recreate on the service thread, outside the backend's error callback. - log::warn!("Recreating interrupted audio capture stream"); + if let Some(stream) = &state.stream { + if stream.errors.needs_restart() { + // Recreate on the service thread, outside the capture callbacks. + log::warn!("Recreating audio capture stream after an error"); super::restart(); } } @@ -255,29 +275,89 @@ mod cpal_impl { } } - fn send( - data: Vec, - sample_rate0: u32, - sample_rate: u32, + #[derive(Clone, Copy)] + struct CaptureFrameProcessorConfig { + input_rate: u32, + output_rate: u32, device_channel: u16, encode_channel: u16, - encoder: &mut Encoder, - sp: &GenericService, - ) { - let mut data = data; - if sample_rate0 != sample_rate { - data = crate::common::audio_resample(&data, sample_rate0, sample_rate, device_channel); + } + + struct CaptureFrameProcessor { + config: CaptureFrameProcessorConfig, + resampler: Option, + sender: audio_capture_queue::CapturePcmSender, + rechannel_buffer: Vec, + } + + struct CaptureStreamOutput { + sender: audio_capture_queue::CapturePcmSender, + sample_rate: u32, + encode_channel: magnum_opus::Channels, + } + + impl CaptureFrameProcessor { + fn new( + config: CaptureFrameProcessorConfig, + sender: audio_capture_queue::CapturePcmSender, + ) -> ResultType { + let resampler = if config.input_rate == config.output_rate { + None + } else { + let output_frames = config.output_rate as usize / AUDIO_PACKETS_PER_SECOND; + Some(crate::audio_resampler::FixedFrameAudioResampler::new( + crate::audio_resampler::AudioResamplerConfig { + input_rate: config.input_rate, + output_rate: config.output_rate, + channels: config.device_channel, + }, + output_frames, + )?) + }; + Ok(Self { + config, + resampler, + sender, + rechannel_buffer: Vec::with_capacity( + capture_packet_layout(config.output_rate, config.encode_channel)?.1, + ), + }) } - if device_channel != encode_channel { - data = crate::common::audio_rechannel( - data, - sample_rate, - sample_rate, - device_channel, - encode_channel, + + fn process(&mut self, data: &[f32]) -> ResultType<()> { + let config = self.config; + let sender = &mut self.sender; + let rechannel_buffer = &mut self.rechannel_buffer; + let mut send_packet = |packet: &[f32]| { + let packet = + audio_capture::rechannel(packet, config.device_channel, rechannel_buffer); + sender.submit(packet); + }; + if let Some(resampler) = self.resampler.as_mut() { + resampler.process_with(data, send_packet).with_context(|| { + format!( + "Failed to resample captured audio from {} Hz to {} Hz", + config.input_rate, config.output_rate + ) + })?; + } else { + send_packet(data); + } + Ok(()) + } + } + + fn capture_packet_layout(sample_rate: u32, channels: u16) -> ResultType<(usize, usize)> { + if sample_rate < AUDIO_PACKETS_PER_SECOND as u32 || channels == 0 { + bail!("Invalid audio capture layout: sample_rate={sample_rate}, channels={channels}"); + } + let frames = sample_rate as usize / AUDIO_PACKETS_PER_SECOND; + let samples = frames.checked_mul(channels as usize).with_context(|| { + format!( + "Audio capture frame size overflow: sample_rate={sample_rate}, channels={channels}" ) - } - send_f32(&data, encoder, sp); + })?; + Ok((frames, samples)) } #[cfg(feature = "screencapturekit")] @@ -367,9 +447,7 @@ mod cpal_impl { Ok((device, format)) } - fn play( - sp: &GenericService, - ) -> ResultType<(Box, Arc, CaptureErrorHandler)> { + fn play(sp: &GenericService) -> ResultType { use cpal::SampleFormat::*; let (device, config) = get_device()?; let sp = sp.clone(); @@ -387,109 +465,274 @@ mod cpal_impl { 48000 }; let ch = if config.channels() > 1 { Stereo } else { Mono }; + let max_channels = config.channels().max(ch as u16); + let (_, max_packet_samples) = capture_packet_layout(sample_rate, max_channels)?; + let encoder_config = audio_capture_queue::CaptureEncoderConfig { + sample_rate, + encode_channel: ch, + max_packet_samples, + }; + let (sender, encoder_worker) = + audio_capture_queue::start_capture_encoder(encoder_config, sp)?; + let output = CaptureStreamOutput { + sender, + sample_rate, + encode_channel: ch, + }; let (stream, errors) = match config.sample_format() { - I8 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - I16 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - I32 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - I64 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - U8 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - U16 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - U32 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - U64 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - F32 => build_input_stream::(device, &config, sp, sample_rate, ch)?, - F64 => build_input_stream::(device, &config, sp, sample_rate, ch)?, + I8 => build_input_stream::(device, &config, output)?, + I16 => build_input_stream::(device, &config, output)?, + I32 => build_input_stream::(device, &config, output)?, + I64 => build_input_stream::(device, &config, output)?, + U8 => build_input_stream::(device, &config, output)?, + U16 => build_input_stream::(device, &config, output)?, + U32 => build_input_stream::(device, &config, output)?, + U64 => build_input_stream::(device, &config, output)?, + F32 => build_input_stream::(device, &config, output)?, + F64 => build_input_stream::(device, &config, output)?, f => bail!("unsupported audio format: {:?}", f), }; stream.play()?; #[cfg(target_os = "macos")] log::info!("Audio capture start call succeeded"); - Ok(( - Box::new(stream), - Arc::new(create_format_msg(sample_rate, ch as _)), + Ok(ActiveCaptureStream { + stream: Some(Box::new(stream)), + format: Arc::new(create_format_msg(sample_rate, ch as _)), + _encoder_worker: encoder_worker, errors, - )) + }) + } + + fn convert_input_samples(data: &[T]) -> impl Iterator + '_ + where + T: cpal::SizedSample, + f32: cpal::FromSample, + { + data.iter() + .map(|sample| >::from_sample_(*sample)) + } + + #[cfg(target_os = "macos")] + fn log_capture_startup( + data: &[T], + received_samples: bool, + received_signal: bool, + ) -> (bool, bool) + where + T: cpal::SizedSample, + f32: cpal::FromSample, + { + // Starting capture does not guarantee sample delivery or audible data. + if !received_samples && !data.is_empty() { + log::info!( + "Audio capture received first PCM block: {} samples", + data.len() + ); + } + let has_signal = received_signal + || convert_input_samples(data).any(|sample| sample.is_finite() && sample != 0.0); + if !received_signal && has_signal { + log::info!("Audio capture received first nonzero PCM"); + } + (received_samples || !data.is_empty(), has_signal) } fn build_input_stream( device: cpal::Device, config: &cpal::SupportedStreamConfig, - sp: GenericService, - sample_rate: u32, - encode_channel: magnum_opus::Channels, + output: CaptureStreamOutput, ) -> ResultType<(cpal::Stream, CaptureErrorHandler)> where - T: cpal::SizedSample + dasp::sample::ToSample, + T: cpal::SizedSample, + f32: cpal::FromSample, { let errors = CaptureErrorHandler::default(); let callback_errors = errors.clone(); let err_fn = move |err| callback_errors.handle(err); + let processor_errors = errors.clone(); #[cfg(target_os = "macos")] let (mut received_samples, mut received_signal) = (false, false); let sample_rate_0 = config.sample_rate().0; - log::debug!("Audio sample rate : {}", sample_rate); - unsafe { - AUDIO_ZERO_COUNT = 0; - } + log::debug!("Audio sample rate : {}", output.sample_rate); let device_channel = config.channels(); - let mut encoder = Encoder::new(sample_rate, encode_channel, LowDelay)?; - // https://www.opus-codec.org/docs/html_api/group__opusencoder.html#gace941e4ef26ed844879fde342ffbe546 - // https://chromium.googlesource.com/chromium/deps/opus/+/1.1.1/include/opus.h - // Do not set `frame_size = sample_rate as usize / 100;` - // Because we find `sample_rate as usize / 100` will cause encoder error in `encoder.encode_vec_float()` sometimes. - // https://github.com/xiph/opus/blob/2554a89e02c7fc30a980b4f7e635ceae1ecba5d6/src/opus_encoder.c#L725 - let frame_size = sample_rate_0 as usize / 100; // 10 ms - let encode_len = frame_size * encode_channel as usize; - let rechannel_len = encode_len * device_channel as usize / encode_channel as usize; - INPUT_BUFFER.lock().unwrap().clear(); - let timeout = None; - let stream_config = StreamConfig { - channels: device_channel, - sample_rate: config.sample_rate(), - buffer_size: BufferSize::Default, + let (_, capture_frame_samples) = capture_packet_layout(sample_rate_0, device_channel)?; + let mut frame = audio_capture::CaptureFrameBuffer::new(capture_frame_samples)?; + let processor_config = CaptureFrameProcessorConfig { + input_rate: sample_rate_0, + output_rate: output.sample_rate, + device_channel, + encode_channel: output.encode_channel as _, }; + let mut processor = CaptureFrameProcessor::new(processor_config, output.sender)?; + let timeout = None; let stream = device.build_input_stream( - &stream_config, + &config.config(), move |data: &[T], _: &InputCallbackInfo| { - let buffer: Vec = data.iter().map(|s| T::to_sample(*s)).collect(); + if processor_errors.needs_restart() { + return; + } #[cfg(target_os = "macos")] { - // Starting capture does not guarantee sample delivery or audible data. - if !received_samples && !buffer.is_empty() { - received_samples = true; - log::info!( - "Audio capture received first PCM block: {} samples", - buffer.len() - ); - } - if !received_signal - && buffer - .iter() - .any(|sample| sample.is_finite() && *sample != 0.0) - { - received_signal = true; - log::info!("Audio capture received first nonzero PCM"); - } - } - let mut lock = INPUT_BUFFER.lock().unwrap(); - lock.extend(buffer); - while lock.len() >= rechannel_len { - let frame: Vec = lock.drain(0..rechannel_len).collect(); - send( - frame, - sample_rate_0, - sample_rate, - device_channel, - encode_channel as _, - &mut encoder, - &sp, - ); + (received_samples, received_signal) = + log_capture_startup(data, received_samples, received_signal); } + frame.process(convert_input_samples(data), |frame| { + processor_errors.process_frame(|| processor.process(frame)); + }); }, err_fn, timeout, )?; Ok((stream, errors)) } + + #[cfg(test)] + mod tests { + use super::super::audio_capture_queue::{ + new_pcm_handoff, start_capture_encoder, CaptureEncoderConfig, + }; + use super::{ + capture_packet_layout, convert_input_samples, CaptureFrameProcessor, + CaptureFrameProcessorConfig, + }; + use crate::audio_resampler::allocation_tests::assert_no_allocations; + use crate::server::EmptyExtraFieldService; + use magnum_opus::Channels::{Mono, Stereo}; + + const INVALID_CAPTURE_RATE: u32 = 99; + const RATE_24_KHZ: u32 = 24_000; + const RATE_44_1_KHZ: u32 = 44_100; + const RATE_48_KHZ: u32 = 48_000; + const MONO_CHANNELS: u16 = 1; + const NEGATIVE_FULL_SCALE_LIMIT: f32 = -0.99; + const POSITIVE_FULL_SCALE_LIMIT: f32 = 0.99; + const STEREO_CHANNELS: u16 = 2; + const SURROUND_CHANNELS: u16 = 6; + const ZERO_CHANNELS: u16 = 0; + + #[test] + fn capture_sample_conversion_uses_cpal_traits() { + let input = [i16::MIN, 0, i16::MAX]; + let output: Vec<_> = convert_input_samples(&input).collect(); + + assert_eq!(output.len(), input.len()); + assert!(output[0] <= NEGATIVE_FULL_SCALE_LIMIT); + assert_eq!(output[1], 0.0); + assert!(output[2] >= POSITIVE_FULL_SCALE_LIMIT); + } + + #[test] + fn capture_packet_layout_validates_rate_and_channels() { + let expected_frames = RATE_48_KHZ as usize / super::AUDIO_PACKETS_PER_SECOND; + assert_eq!( + capture_packet_layout(RATE_48_KHZ, STEREO_CHANNELS).unwrap(), + (expected_frames, expected_frames * STEREO_CHANNELS as usize) + ); + assert!(capture_packet_layout(INVALID_CAPTURE_RATE, MONO_CHANNELS).is_err()); + assert!(capture_packet_layout(RATE_48_KHZ, ZERO_CHANNELS).is_err()); + } + + #[test] + fn capture_callback_pipeline_does_not_allocate_after_warmup() { + for (input_rate, output_rate, device_channel, encode_channel) in [ + (RATE_48_KHZ, RATE_48_KHZ, MONO_CHANNELS, MONO_CHANNELS), + (RATE_48_KHZ, RATE_48_KHZ, STEREO_CHANNELS, STEREO_CHANNELS), + (RATE_44_1_KHZ, RATE_24_KHZ, STEREO_CHANNELS, STEREO_CHANNELS), + (RATE_48_KHZ, RATE_48_KHZ, SURROUND_CHANNELS, STEREO_CHANNELS), + ] { + assert_capture_processor_does_not_allocate(CaptureFrameProcessorConfig { + input_rate, + output_rate, + device_channel, + encode_channel, + }); + } + } + + #[test] + fn capture_pcm_handoff_reuses_buffers_and_accounts_for_loss() { + const QUEUE_CAPACITY: usize = 2; + const PACKET_SAMPLES: usize = 4; + const FIRST: [f32; PACKET_SAMPLES] = [1.0; PACKET_SAMPLES]; + const SECOND: [f32; PACKET_SAMPLES] = [2.0; PACKET_SAMPLES]; + const THIRD: [f32; PACKET_SAMPLES] = [3.0; PACKET_SAMPLES]; + const OVERSIZED_SAMPLES: usize = PACKET_SAMPLES + 1; + const OVERSIZED: [f32; OVERSIZED_SAMPLES] = [1.0; OVERSIZED_SAMPLES]; + + let (mut sender, receiver) = new_pcm_handoff(QUEUE_CAPACITY, PACKET_SAMPLES).unwrap(); + sender.set_wake_thread(std::thread::current()).unwrap(); + assert_no_allocations(|| { + sender.submit(&FIRST); + sender.submit(&SECOND); + sender.submit(&THIRD); + }); + + let loss = receiver.take_loss(); + assert_eq!(loss.dropped, 1); + assert_eq!(loss.oversized, 0); + assert_eq!(loss.recycle_failures, 0); + let second = receiver.pop().unwrap(); + let third = receiver.pop().unwrap(); + assert_eq!(second, SECOND); + assert_eq!(third, THIRD); + receiver.recycle(second); + receiver.recycle(third); + assert!(receiver.is_empty()); + + assert_no_allocations(|| sender.submit(&OVERSIZED)); + let loss = receiver.take_loss(); + assert_eq!(loss.dropped, 0); + assert_eq!(loss.oversized, 1); + assert_eq!(loss.recycle_failures, 0); + assert!(receiver.is_empty()); + } + + #[test] + fn capture_pcm_handoff_rejects_invalid_layouts() { + assert!(new_pcm_handoff(0, 1).is_err()); + assert!(new_pcm_handoff(1, 0).is_err()); + } + + fn assert_capture_processor_does_not_allocate(config: CaptureFrameProcessorConfig) { + const INPUT_LEVEL: f32 = 0.25; + const TEST_SERVICE_NAME: &str = "audio-allocation-test"; + + let service = EmptyExtraFieldService::new(TEST_SERVICE_NAME.to_owned(), true).sp; + let encode_channel = if config.encode_channel == MONO_CHANNELS { + Mono + } else { + Stereo + }; + let encoder_config = CaptureEncoderConfig { + sample_rate: config.output_rate, + encode_channel, + max_packet_samples: config.output_rate as usize / super::AUDIO_PACKETS_PER_SECOND + * config.device_channel.max(config.encode_channel) as usize, + }; + let (sender, worker) = start_capture_encoder(encoder_config, service).unwrap(); + let mut processor = CaptureFrameProcessor::new(config, sender).unwrap(); + let errors = super::CaptureErrorHandler::default(); + let input = vec![ + INPUT_LEVEL; + config.input_rate as usize / super::AUDIO_PACKETS_PER_SECOND + * config.device_channel as usize + ]; + let mut frame_buffer = + super::audio_capture::CaptureFrameBuffer::new(input.len()).unwrap(); + + frame_buffer.process(convert_input_samples(&input), |frame| { + errors.process_frame(|| processor.process(frame)); + }); + assert_no_allocations(|| { + frame_buffer.process(convert_input_samples(&input), |frame| { + errors.process_frame(|| processor.process(frame)); + }); + }); + assert!(!errors.needs_restart()); + drop(processor); + drop(worker); + } + } } fn create_format_msg(sample_rate: u32, channels: u16) -> Message { @@ -505,28 +748,43 @@ fn create_format_msg(sample_rate: u32, channels: u16) -> Message { msg } -// use AUDIO_ZERO_COUNT for the Noise(Zero) Gate Attack Time +// Use a per-encoder counter for the Noise(Zero) Gate Attack Time. // every audio data length is set to 480 // MAX_AUDIO_ZERO_COUNT=800 is similar as Gate Attack Time 3~5s(Linux) || 6~8s(Windows) const MAX_AUDIO_ZERO_COUNT: u16 = 800; -static mut AUDIO_ZERO_COUNT: u16 = 0; -fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) { - if data.iter().filter(|x| **x != 0.).next().is_some() { - unsafe { - AUDIO_ZERO_COUNT = 0; +struct AudioEncoder { + encoder: Encoder, + zero_count: u16, +} + +impl AudioEncoder { + fn new(encoder: Encoder) -> Self { + Self { + encoder, + zero_count: 0, } - } else { - unsafe { - if AUDIO_ZERO_COUNT > MAX_AUDIO_ZERO_COUNT { - if AUDIO_ZERO_COUNT == MAX_AUDIO_ZERO_COUNT + 1 { - log::debug!("Audio Zero Gate Attack"); - AUDIO_ZERO_COUNT += 1; - } - return; + } + + fn should_encode(&mut self, data: &[f32]) -> bool { + if data.iter().filter(|x| **x != 0.).next().is_some() { + self.zero_count = 0; + } else if self.zero_count > MAX_AUDIO_ZERO_COUNT { + if self.zero_count == MAX_AUDIO_ZERO_COUNT + 1 { + log::debug!("Audio Zero Gate Attack"); + self.zero_count += 1; } - AUDIO_ZERO_COUNT += 1; + return false; + } else { + self.zero_count += 1; } + true + } +} + +fn send_f32(data: &[f32], encoder: &mut AudioEncoder, sp: &GenericService) { + if !encoder.should_encode(data) { + return; } #[cfg(target_os = "android")] { @@ -539,6 +797,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) { let n = input_size / BATCH_SIZE; for i in 0..n { match encoder + .encoder .encode_vec_float(&data[i * BATCH_SIZE..(i + 1) * BATCH_SIZE], BATCH_SIZE) { Ok(data) => { @@ -549,7 +808,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) { }); sp.send(msg_out); } - Err(_) => {} + Err(error) => log::warn!("Failed to encode audio frame: {error:?}"), } } } else { @@ -559,7 +818,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) { } #[cfg(not(target_os = "android"))] - match encoder.encode_vec_float(data, data.len() * 6) { + match encoder.encoder.encode_vec_float(data, data.len() * 6) { Ok(data) => { let mut msg_out = Message::new(); msg_out.set_audio_frame(AudioFrame { @@ -568,6 +827,6 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) { }); sp.send(msg_out); } - Err(_) => {} + Err(error) => log::warn!("Failed to encode audio frame: {error:?}"), } } diff --git a/src/server/audio_service/audio_capture.rs b/src/server/audio_service/audio_capture.rs new file mode 100644 index 000000000..27b259835 --- /dev/null +++ b/src/server/audio_service/audio_capture.rs @@ -0,0 +1,144 @@ +use hbb_common::anyhow::{bail, Result}; + +const STEREO_CHANNELS: usize = 2; + +pub(super) struct CaptureFrameBuffer { + samples: Vec, + filled: usize, +} + +impl CaptureFrameBuffer { + pub(super) fn new(samples: usize) -> Result { + if samples == 0 { + bail!("Audio capture frame must contain at least one sample"); + } + Ok(Self { + samples: vec![0.0; samples], + filled: 0, + }) + } + + pub(super) fn process( + &mut self, + input: impl Iterator, + mut on_frame: impl FnMut(&[f32]), + ) { + for sample in input { + self.samples[self.filled] = sample; + self.filled += 1; + if self.filled == self.samples.len() { + self.filled = 0; + on_frame(&self.samples); + } + } + } +} + +pub(super) fn rechannel<'a>( + input: &'a [f32], + channels: u16, + output: &'a mut Vec, +) -> &'a [f32] { + let input = if channels > STEREO_CHANNELS as u16 { + &input[..input.len() / channels as usize * channels as usize] + } else { + input + }; + output.clear(); + match channels { + 3 => rechannel_frame::<3>(input, output), + 4 => rechannel_frame::<4>(input, output), + 5 => rechannel_frame::<5>(input, output), + 6 => rechannel_frame::<6>(input, output), + 7 => rechannel_frame::<7>(input, output), + 8 => rechannel_frame::<8>(input, output), + // Preserve the existing passthrough for mono/stereo and unsupported layouts. + _ => return input, + } + output +} + +fn rechannel_frame(input: &[f32], output: &mut Vec) { + use fon::{ + chan::{Ch32, Channel}, + Frame, + }; + + for samples in input.chunks_exact(CHANNELS) { + let mut frame = Frame::::default(); + for (channel, sample) in frame.channels_mut().iter_mut().zip(samples) { + *channel = (*sample).into(); + } + // Match the same-rate Stream::pipe conversion before fon's SinkTo conversion. + let stereo = frame.to::().to::(); + output.extend(stereo.channels().iter().map(|channel| channel.to_f32())); + } +} + +#[cfg(test)] +mod tests { + use super::{rechannel, CaptureFrameBuffer}; + use crate::audio_resampler::allocation_tests::assert_no_allocations; + + #[test] + fn capture_channel_conversion_reuses_storage_and_preserves_mapping() { + const SAMPLE_RATE: u32 = 48_000; + const FRAMES: usize = 7; + const STEREO_CHANNELS: u16 = 2; + const MIN_SAMPLE: f32 = -1.25; + const SAMPLE_STEP: f32 = 0.2; + + for channels in [1, 2, 3, 4, 5, 6, 7, 8, 16] { + let input: Vec<_> = (0..FRAMES * channels as usize + 1) + .map(|sample| MIN_SAMPLE + sample as f32 * SAMPLE_STEP) + .collect(); + let expected = crate::common::audio_rechannel( + input.clone(), + SAMPLE_RATE, + SAMPLE_RATE, + channels, + channels.min(STEREO_CHANNELS), + ); + let mut output = Vec::with_capacity(FRAMES * STEREO_CHANNELS as usize); + assert_no_allocations(|| { + let actual = rechannel(&input, channels, &mut output); + assert_eq!( + actual, expected, + "channel mapping changed for {channels} channels" + ); + }); + } + } + + #[test] + fn capture_framing_retains_partial_input_without_allocating() { + const FRAME_SAMPLES: usize = 6; + const CALLBACK_SIZES: [usize; 7] = [0, 1, 17, 2, 257, 492, 5]; + let input: Vec<_> = (0..CALLBACK_SIZES.iter().sum::()) + .map(|sample| sample as f32) + .collect(); + let mut buffer = CaptureFrameBuffer::new(FRAME_SAMPLES).unwrap(); + let mut input_position = 0; + let mut output_position = 0; + + assert!(CaptureFrameBuffer::new(0).is_err()); + assert_no_allocations(|| { + for samples in CALLBACK_SIZES { + let end = input_position + samples; + buffer.process(input[input_position..end].iter().copied(), |frame| { + assert_eq!( + frame, + &input[output_position..output_position + FRAME_SAMPLES] + ); + output_position += FRAME_SAMPLES; + }); + input_position = end; + assert_eq!( + output_position, + input_position / FRAME_SAMPLES * FRAME_SAMPLES + ); + } + }); + assert_eq!(output_position, input.len()); + } +} diff --git a/src/server/audio_service/audio_capture_encoder.rs b/src/server/audio_service/audio_capture_encoder.rs new file mode 100644 index 000000000..f5f8b3722 --- /dev/null +++ b/src/server/audio_service/audio_capture_encoder.rs @@ -0,0 +1,128 @@ +use super::super::AudioEncoder; +use super::{ + send_f32, CaptureEncoderConfig, CaptureEncoderContext, CapturePcmReceiver, CapturePcmStats, + CAPTURE_PCM_QUEUE_PACKETS, +}; +use hbb_common::log; +use magnum_opus::Channels; +use std::{ + sync::atomic::Ordering, + time::{Duration, Instant}, +}; + +const CAPTURE_DECLICK_MS: usize = 5; +const CAPTURE_PACKET_MS: usize = 10; +const MILLISECONDS_PER_SECOND: usize = 1_000; +const MAX_ENCODE_CHANNELS: usize = Channels::Stereo as usize; +const CAPTURE_STATS_LOG_INTERVAL: Duration = Duration::from_secs(5); + +struct CaptureEncoderState { + channels: usize, + expected_sequence: usize, + fade_frames: usize, + last_frame: [f32; MAX_ENCODE_CHANNELS], + reporter: CaptureStatsReporter, +} + +impl CaptureEncoderState { + fn new(sample_rate: u32, channels: Channels) -> Self { + // The encoder has already validated the supported Opus rate and channel count. + let fade_frames = sample_rate as usize * CAPTURE_DECLICK_MS / MILLISECONDS_PER_SECOND; + Self { + channels: channels as usize, + expected_sequence: 0, + fade_frames, + last_frame: [0.0; MAX_ENCODE_CHANNELS], + reporter: CaptureStatsReporter::new(), + } + } + + fn next_packet(&mut self, receiver: &CapturePcmReceiver) -> Option> { + self.reporter.pending.add(receiver.take_stats()); + self.reporter.report(false); + let (sequence, mut packet) = receiver.pop_packet()?; + self.smooth_packet(sequence, &mut packet); + Some(packet) + } + + fn smooth_packet(&mut self, sequence: usize, packet: &mut [f32]) { + if sequence != self.expected_sequence { + // Capture packets contain 10 ms of PCM, so the transition fits in this packet. + for (index, frame) in packet + .chunks_exact_mut(self.channels) + .take(self.fade_frames) + .enumerate() + { + let weight = (index + 1) as f32 / self.fade_frames as f32; + for (channel, sample) in frame.iter_mut().enumerate() { + *sample = self.last_frame[channel] * (1.0 - weight) + *sample * weight; + } + } + } + self.expected_sequence = sequence.wrapping_add(1); + if let Some(frame) = packet.chunks_exact(self.channels).next_back() { + self.last_frame[..self.channels].copy_from_slice(frame); + } + } +} + +struct CaptureStatsReporter { + pending: CapturePcmStats, + last_report: Instant, +} + +impl CaptureStatsReporter { + fn new() -> Self { + Self { + pending: Default::default(), + last_report: Instant::now(), + } + } + + fn report(&mut self, force: bool) { + if self.pending.is_empty() { + return; + } + if !force && self.last_report.elapsed() < CAPTURE_STATS_LOG_INTERVAL { + return; + } + let stats = std::mem::take(&mut self.pending); + log::debug!( + "Audio capture PCM handoff stats: observed_max_queued_packets={}, approx_queued_audio_ms={}, capacity_packets={}", + stats.max_queued_packets, + stats.max_queued_packets.saturating_mul(CAPTURE_PACKET_MS), + CAPTURE_PCM_QUEUE_PACKETS + ); + if !stats.loss.is_empty() { + log::warn!( + "Audio capture PCM handoff loss: dropped={}, contention_dropped={}, oversized={}, recycle_failures={}", + stats.loss.dropped, + stats.loss.contention_dropped, + stats.loss.oversized, + stats.loss.recycle_failures + ); + } + self.last_report = Instant::now(); + } +} + +pub(super) fn run_capture_encoder(context: CaptureEncoderContext, config: CaptureEncoderConfig) { + let mut encoder = AudioEncoder::new(context.encoder); + let mut state = CaptureEncoderState::new(config.sample_rate, config.encode_channel); + loop { + while let Some(packet) = state.next_packet(&context.receiver) { + send_f32(&packet, &mut encoder, &context.service); + context.receiver.recycle(packet); + } + if context.stop.load(Ordering::Acquire) && context.receiver.is_empty() { + state.reporter.pending.add(context.receiver.take_stats()); + state.reporter.report(true); + return; + } + std::thread::park_timeout(CAPTURE_STATS_LOG_INTERVAL); + } +} + +#[cfg(test)] +#[path = "audio_capture_encoder_tests.rs"] +mod tests; diff --git a/src/server/audio_service/audio_capture_encoder_tests.rs b/src/server/audio_service/audio_capture_encoder_tests.rs new file mode 100644 index 000000000..3c52c318f --- /dev/null +++ b/src/server/audio_service/audio_capture_encoder_tests.rs @@ -0,0 +1,186 @@ +use super::super::{new_pcm_handoff, CAPTURE_PCM_QUEUE_PACKETS}; +use super::*; +use crate::audio_resampler::allocation_tests::assert_no_allocations; +use magnum_opus::{Application::LowDelay, Decoder, Encoder}; + +const SAMPLE_RATE: u32 = 48_000; +const PACKETS_PER_SECOND: usize = 100; +const PACKET_COUNT: usize = 120; +const PAUSE_PACKET: usize = 50; +const DROPPED_PACKETS: usize = 6; +const SIGNAL_FREQUENCY: f64 = 97.0; +const SIGNAL_AMPLITUDE: f64 = 0.5; +const ACTIVE_LEVEL: f32 = 0.8; +const SAMPLE_TOLERANCE: f32 = 0.000001; +const MAX_ENCODE_BYTES_PER_SAMPLE: usize = 6; + +fn signal_packet(index: usize, channels: usize) -> Vec { + let frames = SAMPLE_RATE as usize / PACKETS_PER_SECOND; + (0..frames) + .flat_map(|frame| { + let phase = std::f64::consts::TAU * SIGNAL_FREQUENCY * (index * frames + frame) as f64 + / f64::from(SAMPLE_RATE); + (0..channels) + .map(move |channel| (SIGNAL_AMPLITUDE * (phase + channel as f64).sin()) as f32) + }) + .collect() +} + +fn encoded_audio(dropped: usize, channels: Channels) -> Vec { + let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND * channels as usize; + let (mut sender, receiver) = new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, samples).unwrap(); + let mut state = CaptureEncoderState::new(SAMPLE_RATE, channels); + let mut encoder = Encoder::new(SAMPLE_RATE, channels, LowDelay).unwrap(); + let mut decoder = Decoder::new(SAMPLE_RATE, channels).unwrap(); + let mut decoded = vec![0.0; samples]; + let mut output = Vec::new(); + let resume_packet = PAUSE_PACKET + CAPTURE_PCM_QUEUE_PACKETS + dropped - 1; + for index in 0..PACKET_COUNT { + sender.submit(&signal_packet(index, channels as usize)); + if (PAUSE_PACKET..resume_packet).contains(&index) { + continue; + } + if index == resume_packet { + let stats = receiver.take_stats(); + assert_eq!(stats.max_queued_packets, CAPTURE_PCM_QUEUE_PACKETS); + assert_eq!(stats.loss.dropped, dropped); + assert_eq!(stats.loss.contention_dropped, 0); + assert_eq!(stats.loss.oversized, 0); + assert_eq!(stats.loss.recycle_failures, 0); + } + while let Some(packet) = state.next_packet(&receiver) { + let encoded = encoder + .encode_vec_float(&packet, samples * MAX_ENCODE_BYTES_PER_SAMPLE) + .unwrap(); + let frames = decoder.decode_float(&encoded, &mut decoded, false).unwrap(); + assert_eq!(frames * channels as usize, samples); + output.extend_from_slice(&decoded); + receiver.recycle(packet); + } + } + assert_eq!(output.len(), (PACKET_COUNT - dropped) * samples); + output +} + +fn maximum_join_step(pcm: &[f32], channels: usize) -> f32 { + let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND * channels; + pcm[(PAUSE_PACKET - 1) * samples..(PAUSE_PACKET + 2) * samples] + .windows(channels + 1) + .map(|window| (window[channels] - window[0]).abs()) + .fold(0.0, f32::max) +} + +#[test] +fn capture_overflow_is_smoothed_before_encoding() { + const MAX_STEP_RATIO: f32 = 2.0; + for channels in [Channels::Mono, Channels::Stereo] { + let clean = encoded_audio(0, channels); + let overflow = encoded_audio(DROPPED_PACKETS, channels); + let clean_step = maximum_join_step(&clean, channels as usize); + let overflow_step = maximum_join_step(&overflow, channels as usize); + assert!(clean_step > 0.0); + assert!( + overflow_step < clean_step * MAX_STEP_RATIO, + "capture discard introduced a sharp join: clean={clean_step}, overflow={overflow_step}" + ); + } +} + +#[test] +fn rejected_packets_mark_the_gap_after_already_queued_audio() { + const CAPACITY: usize = 3; + let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND; + let (mut sender, receiver) = new_pcm_handoff(CAPACITY, samples).unwrap(); + let mut state = CaptureEncoderState::new(SAMPLE_RATE, Channels::Mono); + let active = vec![ACTIVE_LEVEL; samples]; + let opposite = vec![-ACTIVE_LEVEL; samples]; + sender.submit(&active); + let packet = state.next_packet(&receiver).unwrap(); + assert_eq!(packet, active); + receiver.recycle(packet); + sender.submit(&active); + sender.submit(&active); + sender.submit(&vec![ACTIVE_LEVEL; samples + 1]); + sender.submit(&opposite); + assert_eq!(receiver.take_loss().oversized, 1); + assert_no_allocations(|| { + for _ in 0..CAPACITY - 1 { + let packet = state.next_packet(&receiver).unwrap(); + assert_eq!(packet, active); + receiver.recycle(packet); + } + let packet = state.next_packet(&receiver).unwrap(); + let expected = ACTIVE_LEVEL * (1.0 - 2.0 / state.fade_frames as f32); + assert!((packet[0] - expected).abs() < SAMPLE_TOLERANCE); + assert_eq!(&packet[samples / 2..], &opposite[samples / 2..]); + receiver.recycle(packet); + }); +} + +#[test] +fn gaps_and_sequence_wrap_preserve_channel_history() { + const RATE_8_KHZ: u32 = 8_000; + for (rate, channels) in [ + (RATE_8_KHZ, Channels::Mono), + (RATE_8_KHZ, Channels::Stereo), + (SAMPLE_RATE, Channels::Mono), + (SAMPLE_RATE, Channels::Stereo), + ] { + let channels_count = channels as usize; + let mut state = CaptureEncoderState::new(rate, channels); + let samples = rate as usize / PACKETS_PER_SECOND * channels_count; + let active: Vec<_> = [ACTIVE_LEVEL, -ACTIVE_LEVEL][..channels_count] + .iter() + .copied() + .cycle() + .take(samples) + .collect(); + let mut first = active.clone(); + state.smooth_packet(0, &mut first); + assert_eq!(first, active); + let mut opposite: Vec<_> = active.iter().map(|sample| -sample).collect(); + let mut resumed = active.clone(); + assert_no_allocations(|| { + state.smooth_packet(2, &mut opposite); + state.smooth_packet(4, &mut resumed); + }); + for channel in 0..channels_count { + let expected = active[channel] * (1.0 - 2.0 / state.fade_frames as f32); + assert!((opposite[channel] - expected).abs() < SAMPLE_TOLERANCE); + let previous = opposite[samples - channels_count + channel]; + let expected = previous + (active[channel] - previous) / state.fade_frames as f32; + assert!((resumed[channel] - expected).abs() < SAMPLE_TOLERANCE); + } + assert_eq!(&resumed[samples / 2..], &active[samples / 2..]); + state.expected_sequence = usize::MAX; + let mut last = active.clone(); + let mut wrapped: Vec<_> = active.iter().map(|sample| -sample).collect(); + let expected = wrapped.clone(); + assert_no_allocations(|| { + state.smooth_packet(usize::MAX, &mut last); + state.smooth_packet(0, &mut wrapped); + }); + assert_eq!(last, active); + assert_eq!(wrapped, expected); + } +} + +#[test] +fn capture_loss_is_reported_while_packets_remain_queued() { + const CAPACITY: usize = 2; + let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND; + let (mut sender, receiver) = new_pcm_handoff(CAPACITY, samples).unwrap(); + let mut state = CaptureEncoderState::new(SAMPLE_RATE, Channels::Mono); + let before_report = Instant::now() - CAPTURE_STATS_LOG_INTERVAL; + state.reporter.last_report = before_report; + let packet = vec![ACTIVE_LEVEL; samples]; + for _ in 0..=CAPACITY { + sender.submit(&packet); + } + let packet = state.next_packet(&receiver).unwrap(); + assert!(!receiver.is_empty()); + assert!(state.reporter.last_report > before_report); + assert!(state.reporter.pending.is_empty()); + assert!(receiver.take_loss().is_empty()); + receiver.recycle(packet); +} diff --git a/src/server/audio_capture_error.rs b/src/server/audio_service/audio_capture_error.rs similarity index 54% rename from src/server/audio_capture_error.rs rename to src/server/audio_service/audio_capture_error.rs index 34eec584d..a1106f737 100644 --- a/src/server/audio_capture_error.rs +++ b/src/server/audio_service/audio_capture_error.rs @@ -26,6 +26,20 @@ impl CaptureErrorHandler { } } + pub(super) fn process_frame(&self, process: impl FnOnce() -> hbb_common::ResultType<()>) { + // Defensive recovery: persistent processing failures with valid capture input + // have not been reproduced. Stop using a failed processor until stream replacement. + if self.needs_restart() { + return; + } + if let Err(error) = process() { + self.interrupted.store(true, Ordering::Relaxed); + hbb_common::log::error!( + "Failed to process captured audio frame; requesting stream restart: {error:#}" + ); + } + } + pub(super) fn needs_restart(&self) -> bool { self.interrupted.load(Ordering::Relaxed) } @@ -72,4 +86,44 @@ mod tests { }); assert!(!errors.needs_restart()); } + + #[test] + fn processing_failure_skips_remaining_frames_until_stream_replacement() { + const FRAME_SAMPLES: usize = 2; + const FRAMES_PER_CALLBACK: usize = 3; + const CALLBACK_COUNT: usize = 2; + const FAILURE_CALL: usize = 2; + let errors = CaptureErrorHandler::default(); + let callback_errors = errors.clone(); + let mut framer = + super::super::audio_capture::CaptureFrameBuffer::new(FRAME_SAMPLES).unwrap(); + let input = [0.0; FRAME_SAMPLES * FRAMES_PER_CALLBACK]; + let mut processed = 0; + let mut failures = 0; + + // Inject an error to test recovery; this is not a valid-input backend failure reproduction. + for _ in 0..CALLBACK_COUNT { + framer.process(input.iter().copied(), |_| { + callback_errors.process_frame(|| { + processed += 1; + if processed >= FAILURE_CALL { + failures += 1; + hbb_common::anyhow::bail!("Injected capture processing failure"); + } + Ok(()) + }); + }); + } + + assert_eq!(processed, FAILURE_CALL); + assert_eq!(failures, 1); + assert!(errors.needs_restart()); + let replacement = CaptureErrorHandler::default(); + replacement.process_frame(|| { + processed += 1; + Ok(()) + }); + assert_eq!(processed, FAILURE_CALL + 1); + assert!(!replacement.needs_restart()); + } } diff --git a/src/server/audio_service/audio_capture_queue.rs b/src/server/audio_service/audio_capture_queue.rs new file mode 100644 index 000000000..e46798e5b --- /dev/null +++ b/src/server/audio_service/audio_capture_queue.rs @@ -0,0 +1,289 @@ +use super::{send_f32, GenericService}; +use hbb_common::{ + anyhow::{bail, Context, Result}, + log, +}; +use magnum_opus::{Application::LowDelay, Channels, Encoder}; +use std::{ + collections::VecDeque, + sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Arc, Mutex, OnceLock, TryLockError, + }, + thread::{JoinHandle, Thread}, +}; + +#[path = "audio_capture_encoder.rs"] +mod encoder; + +const CAPTURE_PCM_QUEUE_PACKETS: usize = 10; +const CAPTURE_ENCODER_THREAD_NAME: &str = "audio-encoder"; + +#[derive(Debug, Default, PartialEq, Eq)] +pub(super) struct CapturePcmLoss { + pub(super) dropped: usize, + pub(super) contention_dropped: usize, + pub(super) oversized: usize, + pub(super) recycle_failures: usize, +} + +impl CapturePcmLoss { + pub(super) fn is_empty(&self) -> bool { + self.dropped == 0 && self.oversized == 0 && self.recycle_failures == 0 + } + + pub(super) fn add(&mut self, other: Self) { + self.dropped += other.dropped; + self.contention_dropped += other.contention_dropped; + self.oversized += other.oversized; + self.recycle_failures += other.recycle_failures; + } +} + +#[derive(Debug, Default, PartialEq, Eq)] +pub(super) struct CapturePcmStats { + pub(super) loss: CapturePcmLoss, + pub(super) max_queued_packets: usize, +} + +impl CapturePcmStats { + pub(super) fn is_empty(&self) -> bool { + self.loss.is_empty() && self.max_queued_packets == 0 + } + + pub(super) fn add(&mut self, other: Self) { + self.loss.add(other.loss); + self.max_queued_packets = self.max_queued_packets.max(other.max_queued_packets); + } +} + +struct CapturePcmHandoff { + buffers: Mutex, + wake_thread: OnceLock, + other_dropped: AtomicUsize, + contention_dropped: AtomicUsize, + oversized: AtomicUsize, + recycle_failures: AtomicUsize, + max_queued_packets: AtomicUsize, + max_samples: usize, +} + +struct CapturePcmBuffers { + available: Vec>, + ready: VecDeque<(usize, Vec)>, +} + +pub(super) struct CapturePcmSender { + handoff: Arc, + sequence: usize, +} + +pub(super) struct CapturePcmReceiver { + handoff: Arc, +} + +pub(super) struct CaptureEncoderConfig { + pub(super) sample_rate: u32, + pub(super) encode_channel: Channels, + pub(super) max_packet_samples: usize, +} + +pub(super) struct CaptureEncoderWorker { + stop: Arc, + handle: Option>, +} + +impl Drop for CaptureEncoderWorker { + fn drop(&mut self) { + self.stop.store(true, Ordering::Release); + if let Some(handle) = self.handle.take() { + handle.thread().unpark(); + if let Err(error) = handle.join() { + log::error!("Failed to join audio encoder thread: {error:?}"); + } + } + } +} + +struct CaptureEncoderContext { + receiver: CapturePcmReceiver, + encoder: Encoder, + service: GenericService, + stop: Arc, +} + +pub(super) fn new_pcm_handoff( + capacity: usize, + max_samples: usize, +) -> Result<(CapturePcmSender, CapturePcmReceiver)> { + if capacity == 0 || max_samples == 0 { + bail!("Audio capture PCM handoff requires nonzero capacity and packet size"); + } + let handoff = Arc::new(CapturePcmHandoff { + buffers: Mutex::new(CapturePcmBuffers { + available: Vec::with_capacity(capacity), + ready: VecDeque::with_capacity(capacity), + }), + wake_thread: OnceLock::new(), + other_dropped: AtomicUsize::new(0), + contention_dropped: AtomicUsize::new(0), + oversized: AtomicUsize::new(0), + recycle_failures: AtomicUsize::new(0), + max_queued_packets: AtomicUsize::new(0), + max_samples, + }); + { + // Initialize the mutex on this thread, including on platforms with lazy allocation. + let mut buffers = handoff.buffers.lock().unwrap(); + for _ in 0..capacity { + buffers.available.push(Vec::with_capacity(max_samples)); + } + } + Ok(( + CapturePcmSender { + handoff: handoff.clone(), + sequence: 0, + }, + CapturePcmReceiver { handoff }, + )) +} + +impl CapturePcmSender { + pub(super) fn set_wake_thread(&self, thread: Thread) -> Result<()> { + if self.handoff.wake_thread.set(thread).is_err() { + bail!("Audio capture PCM wake thread is already configured"); + } + Ok(()) + } + + pub(super) fn submit(&mut self, input: &[f32]) { + let sequence = self.sequence; + self.sequence = self.sequence.wrapping_add(1); + if input.len() > self.handoff.max_samples { + self.handoff.oversized.fetch_add(1, Ordering::Relaxed); + self.wake(); + return; + } + // Do not wait for a descheduled worker. Contention rejects the current + // packet even if buffers are available; this is separate from drop-oldest + // when the buffer pool is exhausted. + let mut buffers = match self.handoff.buffers.try_lock() { + Ok(buffers) => buffers, + Err(TryLockError::WouldBlock) => { + self.handoff + .contention_dropped + .fetch_add(1, Ordering::Relaxed); + self.wake(); + return; + } + Err(TryLockError::Poisoned(error)) => { + self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed); + log::error!("Audio capture PCM handoff is poisoned: {error}"); + self.wake(); + return; + } + }; + if let Some(mut buffer) = self.take_buffer(&mut buffers) { + buffer.clear(); + buffer.extend_from_slice(input); + buffers.ready.push_back((sequence, buffer)); + self.handoff + .max_queued_packets + .fetch_max(buffers.ready.len(), Ordering::Relaxed); + } else { + self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed); + } + drop(buffers); + self.wake(); + } + + fn take_buffer(&self, buffers: &mut CapturePcmBuffers) -> Option> { + buffers.available.pop().or_else(|| { + let buffer = buffers.ready.pop_front().map(|(_, buffer)| buffer); + if buffer.is_some() { + self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed); + } + buffer + }) + } + + fn wake(&self) { + if let Some(thread) = self.handoff.wake_thread.get() { + thread.unpark(); + } + } +} + +impl CapturePcmReceiver { + #[cfg(test)] + pub(super) fn pop(&self) -> Option> { + self.pop_packet().map(|(_, buffer)| buffer) + } + + fn pop_packet(&self) -> Option<(usize, Vec)> { + self.handoff.buffers.lock().unwrap().ready.pop_front() + } + + pub(super) fn recycle(&self, mut buffer: Vec) { + buffer.clear(); + let mut buffers = self.handoff.buffers.lock().unwrap(); + if buffers.available.len() == buffers.available.capacity() { + self.handoff + .recycle_failures + .fetch_add(1, Ordering::Relaxed); + } else { + buffers.available.push(buffer); + } + } + + pub(super) fn is_empty(&self) -> bool { + self.handoff.buffers.lock().unwrap().ready.is_empty() + } + + pub(super) fn take_loss(&self) -> CapturePcmLoss { + let contention_dropped = self.handoff.contention_dropped.swap(0, Ordering::Relaxed); + CapturePcmLoss { + dropped: self.handoff.other_dropped.swap(0, Ordering::Relaxed) + contention_dropped, + contention_dropped, + oversized: self.handoff.oversized.swap(0, Ordering::Relaxed), + recycle_failures: self.handoff.recycle_failures.swap(0, Ordering::Relaxed), + } + } + + pub(super) fn take_stats(&self) -> CapturePcmStats { + CapturePcmStats { + loss: self.take_loss(), + max_queued_packets: self.handoff.max_queued_packets.swap(0, Ordering::Relaxed), + } + } +} + +pub(super) fn start_capture_encoder( + config: CaptureEncoderConfig, + service: GenericService, +) -> Result<(CapturePcmSender, CaptureEncoderWorker)> { + let (sender, receiver) = new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, config.max_packet_samples)?; + let encoder = Encoder::new(config.sample_rate, config.encode_channel, LowDelay)?; + let stop = Arc::new(AtomicBool::new(false)); + let context = CaptureEncoderContext { + receiver, + encoder, + service, + stop: stop.clone(), + }; + let handle = std::thread::Builder::new() + .name(CAPTURE_ENCODER_THREAD_NAME.to_owned()) + .spawn(move || encoder::run_capture_encoder(context, config)) + .with_context(|| "Failed to start audio encoder thread")?; + let wake_thread = handle.thread().clone(); + let worker = CaptureEncoderWorker { + stop, + handle: Some(handle), + }; + sender.set_wake_thread(wake_thread)?; + Ok((sender, worker)) +} + +#[cfg(test)] +#[path = "audio_capture_queue_tests.rs"] +mod tests; diff --git a/src/server/audio_service/audio_capture_queue_tests.rs b/src/server/audio_service/audio_capture_queue_tests.rs new file mode 100644 index 000000000..90caf2ed5 --- /dev/null +++ b/src/server/audio_service/audio_capture_queue_tests.rs @@ -0,0 +1,169 @@ +use super::*; +use crate::audio_resampler::allocation_tests::assert_no_allocations; +use std::{sync::mpsc, time::Duration}; + +const PACKET_SAMPLES: usize = 4; +const TEST_TIMEOUT: Duration = Duration::from_secs(2); +const CONCURRENT_PACKETS: usize = 10_000; + +#[derive(Clone, Copy)] +enum PausePoint { + Recycle, + Consume, +} + +struct WorkerPause { + entered: mpsc::Sender<()>, + resume: mpsc::Receiver<()>, +} + +fn paused_worker( + receiver: CapturePcmReceiver, + point: PausePoint, + pause: WorkerPause, +) -> CapturePcmReceiver { + let recycled = match point { + PausePoint::Recycle => Some(receiver.pop_packet().unwrap().1), + PausePoint::Consume => None, + }; + let consumed = { + let mut buffers = receiver.handoff.buffers.lock().unwrap(); + let consumed = match recycled { + Some(mut packet) => { + packet.clear(); + buffers.available.push(packet); + None + } + None => Some(buffers.ready.pop_front().unwrap().1), + }; + pause.entered.send(()).unwrap(); + pause.resume.recv().unwrap(); + consumed + }; + if let Some(packet) = consumed { + receiver.recycle(packet); + } + receiver +} + +fn assert_callback_progress(point: PausePoint) { + let (mut sender, receiver) = + new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, PACKET_SAMPLES).unwrap(); + for sequence in 0..CAPTURE_PCM_QUEUE_PACKETS { + sender.submit(&[sequence as f32; PACKET_SAMPLES]); + } + let (entered_tx, entered_rx) = mpsc::channel(); + let (resume_tx, resume_rx) = mpsc::channel(); + let pause = WorkerPause { + entered: entered_tx, + resume: resume_rx, + }; + let worker = std::thread::spawn(move || paused_worker(receiver, point, pause)); + sender.set_wake_thread(worker.thread().clone()).unwrap(); + entered_rx.recv_timeout(TEST_TIMEOUT).unwrap(); + let (completed_tx, completed_rx) = mpsc::channel(); + let callback = std::thread::spawn(move || { + let packet = [CAPTURE_PCM_QUEUE_PACKETS as f32; PACKET_SAMPLES]; + assert_no_allocations(|| sender.submit(&packet)); + completed_tx.send(()).unwrap(); + sender + }); + let completed = completed_rx.recv_timeout(TEST_TIMEOUT); + // Release and join both threads before asserting progress, including on failure. + resume_tx.send(()).unwrap(); + let receiver = worker.join().unwrap(); + let mut sender = callback.join().unwrap(); + assert!(completed.is_ok(), "callback waited for the encoder worker"); + assert_eq!( + receiver.take_loss(), + CapturePcmLoss { + dropped: 1, + contention_dropped: 1, + ..Default::default() + } + ); + assert_packets_after_contention(&mut sender, &receiver); +} + +fn assert_packets_after_contention(sender: &mut CapturePcmSender, receiver: &CapturePcmReceiver) { + let next_sequence = CAPTURE_PCM_QUEUE_PACKETS + 1; + sender.submit(&[next_sequence as f32; PACKET_SAMPLES]); + for expected in (1..CAPTURE_PCM_QUEUE_PACKETS).chain(std::iter::once(next_sequence)) { + let (sequence, packet) = receiver.pop_packet().unwrap(); + assert_eq!(sequence, expected); + assert_eq!(packet, [expected as f32; PACKET_SAMPLES]); + receiver.recycle(packet); + } + assert!(receiver.is_empty()); + assert!(receiver.take_loss().is_empty()); + assert_pool_restored(receiver, CAPTURE_PCM_QUEUE_PACKETS); +} + +fn assert_pool_restored(receiver: &CapturePcmReceiver, capacity: usize) { + let buffers = receiver.handoff.buffers.lock().unwrap(); + assert_eq!(buffers.available.len(), capacity); + assert!(buffers + .available + .iter() + .all(|buffer| buffer.capacity() >= PACKET_SAMPLES)); +} + +#[test] +fn callback_finishes_while_worker_recycles_a_buffer() { + assert_callback_progress(PausePoint::Recycle); +} + +#[test] +fn callback_finishes_while_worker_releases_a_ready_packet() { + assert_callback_progress(PausePoint::Consume); +} + +fn consume_concurrently( + receiver: CapturePcmReceiver, + finished: Arc, + first_sequence: usize, +) -> (CapturePcmReceiver, usize) { + let mut received = 0; + let mut next_ordinal = 0; + loop { + if let Some((sequence, packet)) = receiver.pop_packet() { + let ordinal = sequence.wrapping_sub(first_sequence); + assert!(ordinal >= next_ordinal && ordinal < CONCURRENT_PACKETS); + assert_eq!(packet, [ordinal as f32; PACKET_SAMPLES]); + next_ordinal = ordinal + 1; + received += 1; + receiver.recycle(packet); + } else if finished.load(Ordering::Acquire) && receiver.is_empty() { + return (receiver, received); + } else { + std::thread::yield_now(); + } + } +} + +#[test] +fn concurrent_handoff_preserves_order_buffers_and_loss_counts_across_sequence_wrap() { + const FIRST_SEQUENCE: usize = usize::MAX - CONCURRENT_PACKETS / 2; + for capacity in [1, 2, CAPTURE_PCM_QUEUE_PACKETS] { + let (mut sender, receiver) = new_pcm_handoff(capacity, PACKET_SAMPLES).unwrap(); + sender.sequence = FIRST_SEQUENCE; + let finished = Arc::new(AtomicBool::new(false)); + let worker_finished = finished.clone(); + let worker = std::thread::spawn(move || { + consume_concurrently(receiver, worker_finished, FIRST_SEQUENCE) + }); + assert_no_allocations(|| { + for ordinal in 0..CONCURRENT_PACKETS { + sender.submit(&[ordinal as f32; PACKET_SAMPLES]); + } + }); + finished.store(true, Ordering::Release); + let (receiver, received) = worker.join().unwrap(); + let stats = receiver.take_stats(); + assert_eq!(received + stats.loss.dropped, CONCURRENT_PACKETS); + assert_eq!(stats.loss.oversized, 0); + assert_eq!(stats.loss.recycle_failures, 0); + assert!(stats.max_queued_packets <= capacity); + assert_pool_restored(&receiver, capacity); + } +}