From f17473eb22fe887d3076c6e869f936c345f4558b Mon Sep 17 00:00:00 2001 From: oiupoyt Date: Tue, 29 Sep 2026 21:18:44 +0530 Subject: [PATCH] fix(audio): implement frame-aware ring buffer, handle registry to prevent double-free, and high-fidelity DFT loopback test --- crates/molotov-core/Cargo.toml | 1 + crates/molotov-core/src/audio/buffer.rs | 130 +++--- crates/molotov-core/src/audio/builder.rs | 3 + crates/molotov-core/src/audio/c_api.rs | 395 ++++++++++++++---- crates/molotov-core/src/audio/pipewire.rs | 69 +-- crates/molotov-core/tests/audio_fixes_test.rs | 74 ++++ .../tests/pipewire_lifecycle_test.rs | 10 +- tests/ci_audio_loopback.rs | 152 +++++-- 8 files changed, 637 insertions(+), 197 deletions(-) create mode 100644 crates/molotov-core/tests/audio_fixes_test.rs diff --git a/crates/molotov-core/Cargo.toml b/crates/molotov-core/Cargo.toml index 221da2a..2f1565a 100644 --- a/crates/molotov-core/Cargo.toml +++ b/crates/molotov-core/Cargo.toml @@ -8,3 +8,4 @@ license.workspace = true tracing = { workspace = true } thiserror = { workspace = true } libc = { workspace = true } +lazy_static = "1.5" diff --git a/crates/molotov-core/src/audio/buffer.rs b/crates/molotov-core/src/audio/buffer.rs index e468b17..2c1cbbb 100644 --- a/crates/molotov-core/src/audio/buffer.rs +++ b/crates/molotov-core/src/audio/buffer.rs @@ -3,72 +3,106 @@ use std::sync::atomic::{AtomicUsize, Ordering}; pub struct AudioRingBuffer { data: Vec, - capacity: usize, - read_pos: AtomicUsize, - write_pos: AtomicUsize, + capacity_frames: usize, + channels: usize, + read_frame: AtomicUsize, + write_frame: AtomicUsize, } impl AudioRingBuffer { - pub fn new(capacity: usize) -> Self { + pub fn new(capacity_samples: usize) -> Self { + Self::new_framed(capacity_samples.max(1), 1) + } + + pub fn new_framed(capacity_frames: usize, channels: usize) -> Self { + let ch = channels.max(1); + // +1 slot so capacity_frames can be stored simultaneously without write == read collision + let cap = capacity_frames.max(1) + 1; Self { - data: vec![0.0; capacity], - capacity, - read_pos: AtomicUsize::new(0), - write_pos: AtomicUsize::new(0), + data: vec![0.0; cap * ch], + capacity_frames: cap, + channels: ch, + read_frame: AtomicUsize::new(0), + write_frame: AtomicUsize::new(0), + } + } + + pub fn available_read_frames(&self) -> usize { + let r = self.read_frame.load(Ordering::Acquire); + let w = self.write_frame.load(Ordering::Acquire); + if w >= r { + w - r + } else { + self.capacity_frames - (r - w) } } pub fn available_read(&self) -> usize { - let r = self.read_pos.load(Ordering::Acquire); - let w = self.write_pos.load(Ordering::Acquire); - if w >= r { - w - r - } else { - self.capacity - (r - w) - } + self.available_read_frames() * self.channels } - pub fn available_write(&self) -> usize { - self.capacity.saturating_sub(self.available_read() + 1) + pub fn write_frames(&mut self, samples: &[f32]) -> usize { + let frames_to_write = samples.len() / self.channels; + let mut w = self.write_frame.load(Ordering::Relaxed); + let mut r = self.read_frame.load(Ordering::Acquire); + + for f in 0..frames_to_write { + let next_w = (w + 1) % self.capacity_frames; + if next_w == r { + // Buffer full: advance read frame to drop oldest frame (preserving channel alignment) + r = (r + 1) % self.capacity_frames; + self.read_frame.store(r, Ordering::Release); + } + + let slot_offset = w * self.channels; + let sample_offset = f * self.channels; + self.data[slot_offset..slot_offset + self.channels] + .copy_from_slice(&samples[sample_offset..sample_offset + self.channels]); + + w = next_w; + } + + self.write_frame.store(w, Ordering::Release); + frames_to_write + } + + pub fn read_frames(&mut self, dest: &mut [f32]) -> usize { + let frames_requested = dest.len() / self.channels; + let mut r = self.read_frame.load(Ordering::Relaxed); + let w = self.write_frame.load(Ordering::Acquire); + let mut frames_read = 0; + + while r != w && frames_read < frames_requested { + let slot_offset = r * self.channels; + let dest_offset = frames_read * self.channels; + dest[dest_offset..dest_offset + self.channels] + .copy_from_slice(&self.data[slot_offset..slot_offset + self.channels]); + + r = (r + 1) % self.capacity_frames; + frames_read += 1; + } + + self.read_frame.store(r, Ordering::Release); + + // Fill remaining underrun space with pure silence + let samples_read = frames_read * self.channels; + if samples_read < dest.len() { + dest[samples_read..].fill(0.0); + } + + frames_read } pub fn write(&mut self, samples: &[f32]) -> usize { - let mut written = 0; - let mut w = self.write_pos.load(Ordering::Relaxed); - - for &sample in samples { - let next_w = (w + 1) % self.capacity; - let r = self.read_pos.load(Ordering::Acquire); - if next_w == r { - // Buffer full: drop oldest sample to avoid stalling capture - let _ = self.read_pos.fetch_add(1, Ordering::Release); - } - self.data[w] = sample; - w = next_w; - written += 1; - } - - self.write_pos.store(w, Ordering::Release); - written + self.write_frames(samples) * self.channels } pub fn read(&mut self, dest: &mut [f32]) -> usize { - let mut read = 0; - let mut r = self.read_pos.load(Ordering::Relaxed); - let w = self.write_pos.load(Ordering::Acquire); - - while r != w && read < dest.len() { - dest[read] = self.data[r]; - r = (r + 1) % self.capacity; - read += 1; - } - - self.read_pos.store(r, Ordering::Release); - read + self.read_frames(dest) * self.channels } pub fn clear(&mut self) { - self.read_pos.store(0, Ordering::Release); - self.write_pos.store(0, Ordering::Release); + self.read_frame.store(0, Ordering::Release); + self.write_frame.store(0, Ordering::Release); } } diff --git a/crates/molotov-core/src/audio/builder.rs b/crates/molotov-core/src/audio/builder.rs index 64e0675..404ecf2 100644 --- a/crates/molotov-core/src/audio/builder.rs +++ b/crates/molotov-core/src/audio/builder.rs @@ -18,6 +18,9 @@ pub struct AAudioStreamBuilder { user_data: *mut (), } +unsafe impl Send for AAudioStreamBuilder {} +unsafe impl Sync for AAudioStreamBuilder {} + impl Default for AAudioStreamBuilder { fn default() -> Self { Self::new() diff --git a/crates/molotov-core/src/audio/c_api.rs b/crates/molotov-core/src/audio/c_api.rs index aee0b9f..69d4cf5 100644 --- a/crates/molotov-core/src/audio/c_api.rs +++ b/crates/molotov-core/src/audio/c_api.rs @@ -1,113 +1,352 @@ // SPDX-License-Identifier: GPL-3.0-or-later -use super::builder::AAudioStreamBuilder; +use super::builder::{AAudioDataCallback, AAudioStreamBuilder}; use super::pipewire::PipeWireStreamController; -use super::types::{AAudioDirection, AAudioResult}; +use super::types::{AAudioDirection, AAudioFormat, AAudioResult, AAudioStreamState}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::RwLock; pub struct AAudioStreamHandle { pub controller: PipeWireStreamController, } -#[no_mangle] -pub unsafe extern "C" fn AAudio_createStreamBuilder( - builder: *mut *mut AAudioStreamBuilder, -) -> i32 { - if builder.is_null() { - return AAudioResult::ErrorNull as i32; - } - let b = Box::new(AAudioStreamBuilder::new()); - *builder = Box::into_raw(b); - AAudioResult::Ok as i32 +static NEXT_HANDLE_ID: AtomicUsize = AtomicUsize::new(1001); + +lazy_static::lazy_static! { + static ref BUILDER_TABLE: RwLock> = RwLock::new(HashMap::new()); + static ref STREAM_TABLE: RwLock> = RwLock::new(HashMap::new()); } #[no_mangle] -pub unsafe extern "C" fn AAudioStreamBuilder_setDirection( - builder: *mut AAudioStreamBuilder, - direction: i32, +pub unsafe extern "C" fn AAudio_createStreamBuilder(builder: *mut *mut ()) -> i32 { + if builder.is_null() { + return AAudioResult::ErrorNull as i32; + } + let id = NEXT_HANDLE_ID.fetch_add(1, Ordering::Relaxed); + if let Ok(mut table) = BUILDER_TABLE.write() { + table.insert(id, AAudioStreamBuilder::new()); + *builder = id as *mut (); + AAudioResult::Ok as i32 + } else { + AAudioResult::ErrorInternal as i32 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStreamBuilder_setDirection(builder: *mut (), direction: i32) { + let id = builder as usize; + if let Ok(mut table) = BUILDER_TABLE.write() { + if let Some(b) = table.get_mut(&id) { + let dir = if direction == 1 { + AAudioDirection::Input + } else { + AAudioDirection::Output + }; + b.set_direction(dir); + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStreamBuilder_setSampleRate(builder: *mut (), sample_rate: i32) { + let id = builder as usize; + if let Ok(mut table) = BUILDER_TABLE.write() { + if let Some(b) = table.get_mut(&id) { + b.set_sample_rate(sample_rate); + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStreamBuilder_setChannelCount(builder: *mut (), channel_count: i32) { + let id = builder as usize; + if let Ok(mut table) = BUILDER_TABLE.write() { + if let Some(b) = table.get_mut(&id) { + b.set_channel_count(channel_count); + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStreamBuilder_setFormat(builder: *mut (), format: i32) { + let id = builder as usize; + if let Ok(mut table) = BUILDER_TABLE.write() { + if let Some(b) = table.get_mut(&id) { + let fmt = match format { + 1 => AAudioFormat::PcmI16, + 2 => AAudioFormat::PcmFloat, + _ => AAudioFormat::Unspecified, + }; + b.set_format(fmt); + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStreamBuilder_setBufferCapacityInFrames( + builder: *mut (), + num_frames: i32, ) { - if let Some(b) = builder.as_mut() { - let dir = if direction == 1 { - AAudioDirection::Input - } else { - AAudioDirection::Output - }; - b.set_direction(dir); + let id = builder as usize; + if let Ok(mut table) = BUILDER_TABLE.write() { + if let Some(b) = table.get_mut(&id) { + b.buffer_capacity_in_frames = num_frames; + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStreamBuilder_setDataCallback( + builder: *mut (), + callback: Option, + user_data: *mut (), +) { + let id = builder as usize; + if let Ok(mut table) = BUILDER_TABLE.write() { + if let Some(b) = table.get_mut(&id) { + b.set_data_callback(callback, user_data); + } } } #[no_mangle] pub unsafe extern "C" fn AAudioStreamBuilder_openStream( - builder: *mut AAudioStreamBuilder, - stream: *mut *mut AAudioStreamHandle, + builder: *mut (), + stream: *mut *mut (), ) -> i32 { if builder.is_null() || stream.is_null() { return AAudioResult::ErrorNull as i32; } - let b = match builder.as_ref() { - Some(val) => val, - None => return AAudioResult::ErrorNull as i32, + let b_id = builder as usize; + let b_table = match BUILDER_TABLE.read() { + Ok(t) => t, + Err(_) => return AAudioResult::ErrorInternal as i32, }; - let controller = PipeWireStreamController::new(b.direction(), b.sample_rate(), b.channel_count()); - let handle = Box::new(AAudioStreamHandle { controller }); - *stream = Box::into_raw(handle); + let b = match b_table.get(&b_id) { + Some(val) => val, + None => return AAudioResult::ErrorInvalidHandle as i32, + }; - AAudioResult::Ok as i32 -} + let controller = PipeWireStreamController::new( + b.direction(), + b.sample_rate(), + b.channel_count(), + b.format(), + b.buffer_capacity_in_frames, + ); -#[no_mangle] -pub unsafe extern "C" fn AAudioStream_requestStart(stream: *mut AAudioStreamHandle) -> i32 { - if let Some(s) = stream.as_mut() { - match s.controller.request_start() { - Ok(()) => AAudioResult::Ok as i32, - Err(e) => e as i32, - } - } else { - AAudioResult::ErrorNull as i32 - } -} + drop(b_table); -#[no_mangle] -pub unsafe extern "C" fn AAudioStream_requestPause(stream: *mut AAudioStreamHandle) -> i32 { - if let Some(s) = stream.as_mut() { - match s.controller.request_pause() { - Ok(()) => AAudioResult::Ok as i32, - Err(e) => e as i32, - } - } else { - AAudioResult::ErrorNull as i32 - } -} - -#[no_mangle] -pub unsafe extern "C" fn AAudioStream_requestStop(stream: *mut AAudioStreamHandle) -> i32 { - if let Some(s) = stream.as_mut() { - match s.controller.request_stop() { - Ok(()) => AAudioResult::Ok as i32, - Err(e) => e as i32, - } - } else { - AAudioResult::ErrorNull as i32 - } -} - -#[no_mangle] -pub unsafe extern "C" fn AAudioStream_close(stream: *mut AAudioStreamHandle) -> i32 { - if !stream.is_null() { - let mut handle = Box::from_raw(stream); - handle.controller.close(); + let s_id = NEXT_HANDLE_ID.fetch_add(1, Ordering::Relaxed); + if let Ok(mut s_table) = STREAM_TABLE.write() { + s_table.insert(s_id, AAudioStreamHandle { controller }); + *stream = s_id as *mut (); AAudioResult::Ok as i32 } else { - AAudioResult::ErrorNull as i32 + AAudioResult::ErrorInternal as i32 } } #[no_mangle] -pub unsafe extern "C" fn AAudioStreamBuilder_delete(builder: *mut AAudioStreamBuilder) -> i32 { - if !builder.is_null() { - drop(Box::from_raw(builder)); - AAudioResult::Ok as i32 +pub unsafe extern "C" fn AAudioStream_requestStart(stream: *mut ()) -> i32 { + let id = stream as usize; + if let Ok(mut table) = STREAM_TABLE.write() { + if let Some(s) = table.get_mut(&id) { + match s.controller.request_start() { + Ok(()) => AAudioResult::Ok as i32, + Err(e) => e as i32, + } + } else { + AAudioResult::ErrorInvalidHandle as i32 + } } else { - AAudioResult::ErrorNull as i32 + AAudioResult::ErrorInternal as i32 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_requestPause(stream: *mut ()) -> i32 { + let id = stream as usize; + if let Ok(mut table) = STREAM_TABLE.write() { + if let Some(s) = table.get_mut(&id) { + match s.controller.request_pause() { + Ok(()) => AAudioResult::Ok as i32, + Err(e) => e as i32, + } + } else { + AAudioResult::ErrorInvalidHandle as i32 + } + } else { + AAudioResult::ErrorInternal as i32 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_requestStop(stream: *mut ()) -> i32 { + let id = stream as usize; + if let Ok(mut table) = STREAM_TABLE.write() { + if let Some(s) = table.get_mut(&id) { + match s.controller.request_stop() { + Ok(()) => AAudioResult::Ok as i32, + Err(e) => e as i32, + } + } else { + AAudioResult::ErrorInvalidHandle as i32 + } + } else { + AAudioResult::ErrorInternal as i32 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_close(stream: *mut ()) -> i32 { + let id = stream as usize; + if let Ok(mut table) = STREAM_TABLE.write() { + if let Some(mut s) = table.remove(&id) { + s.controller.close(); + AAudioResult::Ok as i32 + } else { + AAudioResult::ErrorInvalidHandle as i32 + } + } else { + AAudioResult::ErrorInternal as i32 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStreamBuilder_delete(builder: *mut ()) -> i32 { + let id = builder as usize; + if let Ok(mut table) = BUILDER_TABLE.write() { + if table.remove(&id).is_some() { + AAudioResult::Ok as i32 + } else { + AAudioResult::ErrorInvalidHandle as i32 + } + } else { + AAudioResult::ErrorInternal as i32 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_getSampleRate(stream: *mut ()) -> i32 { + let id = stream as usize; + if let Ok(table) = STREAM_TABLE.read() { + table.get(&id).map(|s| s.controller.sample_rate).unwrap_or(48000) + } else { + 48000 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_getChannelCount(stream: *mut ()) -> i32 { + let id = stream as usize; + if let Ok(table) = STREAM_TABLE.read() { + table.get(&id).map(|s| s.controller.channel_count).unwrap_or(2) + } else { + 2 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_getFormat(stream: *mut ()) -> i32 { + let id = stream as usize; + if let Ok(table) = STREAM_TABLE.read() { + table.get(&id).map(|s| s.controller.format as i32).unwrap_or(AAudioFormat::PcmFloat as i32) + } else { + AAudioFormat::PcmFloat as i32 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_getState(stream: *mut ()) -> i32 { + let id = stream as usize; + if let Ok(table) = STREAM_TABLE.read() { + table.get(&id).map(|s| s.controller.state() as i32).unwrap_or(AAudioStreamState::Closed as i32) + } else { + AAudioStreamState::Closed as i32 + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_read( + stream: *mut (), + buffer: *mut libc::c_void, + num_frames: i32, + _timeout_nanoseconds: i64, +) -> i32 { + if stream.is_null() || buffer.is_null() || num_frames <= 0 { + return AAudioResult::ErrorNull as i32; + } + + let id = stream as usize; + let mut table = match STREAM_TABLE.write() { + Ok(t) => t, + Err(_) => return AAudioResult::ErrorInternal as i32, + }; + + let s = match table.get_mut(&id) { + Some(handle) => handle, + None => return AAudioResult::ErrorInvalidHandle as i32, + }; + + let ch = s.controller.channel_count.max(1) as usize; + let total_samples = (num_frames as usize) * ch; + + match s.controller.format { + AAudioFormat::PcmFloat => { + let slice = std::slice::from_raw_parts_mut(buffer as *mut f32, total_samples); + s.controller.buffer.read_frames(slice) as i32 + } + AAudioFormat::PcmI16 => { + let mut temp_float = vec![0.0f32; total_samples]; + let read_frames = s.controller.buffer.read_frames(&mut temp_float); + let out_slice = std::slice::from_raw_parts_mut(buffer as *mut i16, total_samples); + for (i, &f) in temp_float.iter().enumerate() { + out_slice[i] = (f * 32767.0).clamp(-32768.0, 32767.0) as i16; + } + read_frames as i32 + } + _ => AAudioResult::ErrorUnimplemented as i32, + } +} + +#[no_mangle] +pub unsafe extern "C" fn AAudioStream_write( + stream: *mut (), + buffer: *const libc::c_void, + num_frames: i32, + _timeout_nanoseconds: i64, +) -> i32 { + if stream.is_null() || buffer.is_null() || num_frames <= 0 { + return AAudioResult::ErrorNull as i32; + } + + let id = stream as usize; + let mut table = match STREAM_TABLE.write() { + Ok(t) => t, + Err(_) => return AAudioResult::ErrorInternal as i32, + }; + + let s = match table.get_mut(&id) { + Some(handle) => handle, + None => return AAudioResult::ErrorInvalidHandle as i32, + }; + + let ch = s.controller.channel_count.max(1) as usize; + let total_samples = (num_frames as usize) * ch; + + match s.controller.format { + AAudioFormat::PcmFloat => { + let slice = std::slice::from_raw_parts(buffer as *const f32, total_samples); + s.controller.buffer.write_frames(slice) as i32 + } + AAudioFormat::PcmI16 => { + let in_slice = std::slice::from_raw_parts(buffer as *const i16, total_samples); + let float_samples: Vec = in_slice.iter().map(|&i| i as f32 / 32768.0).collect(); + s.controller.buffer.write_frames(&float_samples) as i32 + } + _ => AAudioResult::ErrorUnimplemented as i32, } } diff --git a/crates/molotov-core/src/audio/pipewire.rs b/crates/molotov-core/src/audio/pipewire.rs index b6196db..0a73f4b 100644 --- a/crates/molotov-core/src/audio/pipewire.rs +++ b/crates/molotov-core/src/audio/pipewire.rs @@ -1,32 +1,42 @@ // SPDX-License-Identifier: GPL-3.0-or-later use super::buffer::AudioRingBuffer; -use super::types::{AAudioDirection, AAudioResult, AAudioStreamState}; +use super::types::{AAudioDirection, AAudioFormat, AAudioResult, AAudioStreamState}; pub struct PipeWireStreamController { - direction: AAudioDirection, - sample_rate: i32, - channel_count: i32, + pub direction: AAudioDirection, + pub sample_rate: i32, + pub channel_count: i32, + pub format: AAudioFormat, state: AAudioStreamState, node_active: bool, pub buffer: AudioRingBuffer, } impl PipeWireStreamController { - pub fn new(direction: AAudioDirection, sample_rate: i32, channel_count: i32) -> Self { + pub fn new( + direction: AAudioDirection, + sample_rate: i32, + channel_count: i32, + format: AAudioFormat, + buffer_capacity_frames: i32, + ) -> Self { + let frames = if buffer_capacity_frames > 0 { + buffer_capacity_frames as usize + } else { + 2048 + }; + Self { direction, sample_rate, channel_count, + format, state: AAudioStreamState::Open, node_active: false, - buffer: AudioRingBuffer::new(48000), + buffer: AudioRingBuffer::new_framed(frames, channel_count as usize), } } - pub fn direction(&self) -> AAudioDirection { - self.direction - } - pub fn state(&self) -> AAudioStreamState { self.state } @@ -36,41 +46,50 @@ impl PipeWireStreamController { } pub fn request_start(&mut self) -> Result<(), AAudioResult> { - if self.state == AAudioStreamState::Started { - return Ok(()); + match self.state { + AAudioStreamState::Closed | AAudioStreamState::Closing => { + return Err(AAudioResult::ErrorInvalidState); + } + AAudioStreamState::Started => { + return Ok(()); + } + _ => {} } tracing::info!( - "PipeWireStreamController: activating {} stream at {} Hz (channels: {})", + "PipeWireStreamController: activating {} stream at {} Hz (channels: {}, format: {:?})", if self.direction == AAudioDirection::Input { "CAPTURE" } else { "PLAYBACK" }, self.sample_rate, - self.channel_count + self.channel_count, + self.format ); - // Dynamically instantiate the PipeWire stream node self.node_active = true; self.state = AAudioStreamState::Started; Ok(()) } pub fn request_pause(&mut self) -> Result<(), AAudioResult> { - if self.state == AAudioStreamState::Paused { - return Ok(()); - } - - if self.direction == AAudioDirection::Input { - // Non-negotiable constraint: completely destroy capture stream to extinguish mic icon - tracing::info!("PipeWireStreamController: destroying microphone capture node on pause"); - self.node_active = false; + match self.state { + AAudioStreamState::Started => {} + AAudioStreamState::Paused => return Ok(()), + _ => return Err(AAudioResult::ErrorInvalidState), } + tracing::info!("PipeWireStreamController: destroying stream node on pause"); + self.node_active = false; + self.buffer.clear(); self.state = AAudioStreamState::Paused; Ok(()) } pub fn request_stop(&mut self) -> Result<(), AAudioResult> { - if self.state == AAudioStreamState::Stopped { - return Ok(()); + match self.state { + AAudioStreamState::Closed | AAudioStreamState::Closing => { + return Err(AAudioResult::ErrorInvalidState); + } + AAudioStreamState::Stopped => return Ok(()), + _ => {} } tracing::info!("PipeWireStreamController: destroying stream node on stop"); diff --git a/crates/molotov-core/tests/audio_fixes_test.rs b/crates/molotov-core/tests/audio_fixes_test.rs new file mode 100644 index 0000000..ce897f1 --- /dev/null +++ b/crates/molotov-core/tests/audio_fixes_test.rs @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +use molotov_core::audio::buffer::AudioRingBuffer; +use molotov_core::audio::c_api::*; +use molotov_core::audio::types::AAudioResult; + +#[test] +fn test_ring_buffer_overrun_no_panic_and_preserves_frames() { + let mut buffer = AudioRingBuffer::new_framed(4, 2); // 4 frames capacity, 2 channels per frame + // Write 6 stereo frames (12 samples): [L0, R0, L1, R1, ...] + let samples = [ + 1.0, 1.5, + 2.0, 2.5, + 3.0, 3.5, + 4.0, 4.5, + 5.0, 5.5, + 6.0, 6.5, + ]; + + // Must not panic on overrun + buffer.write_frames(&samples); + + let mut out = [0.0f32; 8]; // Read 4 frames (8 samples) + let read_frames = buffer.read_frames(&mut out); + assert_eq!(read_frames, 4); + + // Channels must remain aligned: left has .fract() == 0.0, right has .fract() == 0.5 + for i in 0..4 { + let left = out[i * 2]; + let right = out[i * 2 + 1]; + assert_eq!(left.fract(), 0.0, "Left channel phase shifted: {}", left); + assert_eq!(right.fract(), 0.5, "Right channel phase shifted: {}", right); + } +} + +#[test] +fn test_ring_buffer_underrun_fills_silence() { + let mut buffer = AudioRingBuffer::new_framed(8, 1); + buffer.write_frames(&[0.5, 0.6]); // 2 frames + + let mut out = [9.9f32; 6]; // Request 6 frames + let read_frames = buffer.read_frames(&mut out); + assert_eq!(read_frames, 2); + assert_eq!(out[0], 0.5); + assert_eq!(out[1], 0.6); + // Unread frames must be filled with silence (0.0) + for &sample in &out[2..6] { + assert_eq!(sample, 0.0, "Underrun must fill silence"); + } +} + +#[test] +fn test_c_api_double_close_safety_handle_registry() { + unsafe { + let mut builder = std::ptr::null_mut(); + AAudio_createStreamBuilder(&mut builder); + + let mut stream = std::ptr::null_mut(); + AAudioStreamBuilder_openStream(builder, &mut stream); + assert!(!stream.is_null()); + + // First close succeeds + let res1 = AAudioStream_close(stream); + assert_eq!(res1, AAudioResult::Ok as i32); + + // Second close must NOT double-free; it safely returns ErrorInvalidHandle + let res2 = AAudioStream_close(stream); + assert_eq!(res2, AAudioResult::ErrorInvalidHandle as i32); + + // Delete builder + assert_eq!(AAudioStreamBuilder_delete(builder), AAudioResult::Ok as i32); + // Double delete builder also safe + assert_eq!(AAudioStreamBuilder_delete(builder), AAudioResult::ErrorInvalidHandle as i32); + } +} diff --git a/crates/molotov-core/tests/pipewire_lifecycle_test.rs b/crates/molotov-core/tests/pipewire_lifecycle_test.rs index 7b1c25e..ce8e1a6 100644 --- a/crates/molotov-core/tests/pipewire_lifecycle_test.rs +++ b/crates/molotov-core/tests/pipewire_lifecycle_test.rs @@ -1,10 +1,16 @@ // SPDX-License-Identifier: GPL-3.0-or-later use molotov_core::audio::pipewire::PipeWireStreamController; -use molotov_core::audio::types::{AAudioDirection, AAudioStreamState}; +use molotov_core::audio::types::{AAudioDirection, AAudioFormat, AAudioStreamState}; #[test] fn test_capture_stream_only_active_during_start() { - let mut controller = PipeWireStreamController::new(AAudioDirection::Input, 48000, 2); + let mut controller = PipeWireStreamController::new( + AAudioDirection::Input, + 48000, + 2, + AAudioFormat::PcmFloat, + 2048, + ); // Initial state: open, but capture node NOT yet active assert_eq!(controller.state(), AAudioStreamState::Open); diff --git a/tests/ci_audio_loopback.rs b/tests/ci_audio_loopback.rs index 873b7d8..2e34e1e 100644 --- a/tests/ci_audio_loopback.rs +++ b/tests/ci_audio_loopback.rs @@ -1,53 +1,117 @@ // SPDX-License-Identifier: GPL-3.0-or-later -use molotov_core::audio::pipewire::PipeWireStreamController; -use molotov_core::audio::types::AAudioDirection; +use molotov_core::audio::c_api::*; +use molotov_core::audio::types::{AAudioDirection, AAudioFormat, AAudioResult}; +use std::ffi::c_void; #[test] fn test_440hz_loopback_frequency_and_amplitude() { let sample_rate = 48000; - let freq = 440.0f32; - let duration_samples = 48000 / 2; // 0.5s + let target_freq = 440.0f32; + let duration_frames = 24000; // 0.5s at 48kHz - let mut controller = PipeWireStreamController::new(AAudioDirection::Input, sample_rate, 1); - controller.request_start().unwrap(); + // 1. Setup Stream via AAudio C-ABI + unsafe { + let mut builder_ptr = std::ptr::null_mut(); + assert_eq!(AAudio_createStreamBuilder(&mut builder_ptr), AAudioResult::Ok as i32); - // Generate 440 Hz sine wave - let mut generated = Vec::with_capacity(duration_samples); - for i in 0..duration_samples { - let t = i as f32 / sample_rate as f32; - let sample = (2.0 * std::f32::consts::PI * freq * t).sin(); - generated.push(sample); + AAudioStreamBuilder_setDirection(builder_ptr, AAudioDirection::Input as i32); + AAudioStreamBuilder_setSampleRate(builder_ptr, sample_rate); + AAudioStreamBuilder_setChannelCount(builder_ptr, 1); + AAudioStreamBuilder_setFormat(builder_ptr, AAudioFormat::PcmFloat as i32); + AAudioStreamBuilder_setBufferCapacityInFrames(builder_ptr, duration_frames); + + let mut stream_ptr = std::ptr::null_mut(); + assert_eq!(AAudioStreamBuilder_openStream(builder_ptr, &mut stream_ptr), AAudioResult::Ok as i32); + assert!(!stream_ptr.is_null()); + + assert_eq!(AAudioStream_requestStart(stream_ptr), AAudioResult::Ok as i32); + + // 2. Synthesize 440 Hz pure tone into playback + let mut generated = Vec::with_capacity(duration_frames as usize); + for i in 0..duration_frames { + let t = i as f32 / sample_rate as f32; + let sample = (2.0 * std::f32::consts::PI * target_freq * t).sin(); + generated.push(sample); + } + + let written = AAudioStream_write( + stream_ptr, + generated.as_ptr() as *const c_void, + duration_frames, + 0, + ); + assert_eq!(written, duration_frames); + + // 3. Capture audio from stream + let mut captured = vec![0.0f32; duration_frames as usize]; + let read_frames = AAudioStream_read( + stream_ptr, + captured.as_mut_ptr() as *mut c_void, + duration_frames, + 0, + ); + assert_eq!(read_frames, duration_frames); + + // 4. Teardown stream via C-ABI + assert_eq!(AAudioStream_requestPause(stream_ptr), AAudioResult::Ok as i32); + assert_eq!(AAudioStream_close(stream_ptr), AAudioResult::Ok as i32); + assert_eq!(AAudioStreamBuilder_delete(builder_ptr), AAudioResult::Ok as i32); + + // 5. Rigorous Spectral Scan across 430 Hz to 450 Hz in 0.5 Hz steps + let mut max_magnitude = 0.0f32; + let mut peak_freq = 0.0f32; + + let start_freq = 430.0f32; + let end_freq = 450.0f32; + let step = 0.5f32; + let steps = ((end_freq - start_freq) / step) as usize; + + for s in 0..=steps { + let f = start_freq + (s as f32) * step; + let mut real = 0.0f32; + let mut imag = 0.0f32; + for (i, &sample) in captured.iter().enumerate() { + let angle = 2.0 * std::f32::consts::PI * f * (i as f32) / (sample_rate as f32); + real += sample * angle.cos(); + imag -= sample * angle.sin(); + } + let mag = (real * real + imag * imag).sqrt(); + if mag > max_magnitude { + max_magnitude = mag; + peak_freq = f; + } + } + + // Assert peak frequency is detected within ±1.0 Hz tolerance + let freq_delta = (peak_freq - target_freq).abs(); + assert!( + freq_delta <= 1.0, + "Peak frequency {} Hz exceeds ±1.0 Hz tolerance around {} Hz", + peak_freq, + target_freq + ); + + // 6. Compute Normalized Cross-Correlation (rho) between generated and captured waveforms + let mean_gen = generated.iter().sum::() / (duration_frames as f32); + let mean_cap = captured.iter().sum::() / (duration_frames as f32); + + let mut numerator = 0.0f32; + let mut denom_gen = 0.0f32; + let mut denom_cap = 0.0f32; + + for i in 0..duration_frames as usize { + let dg = generated[i] - mean_gen; + let dc = captured[i] - mean_cap; + numerator += dg * dc; + denom_gen += dg * dg; + denom_cap += dc * dc; + } + + let rho = numerator / (denom_gen.sqrt() * denom_cap.sqrt()); + assert!( + rho >= 0.95, + "Waveform fidelity rho = {:.4} is below required 95% threshold", + rho + ); } - - controller.buffer.write(&generated); - - let mut captured = vec![0.0f32; duration_samples]; - let read_count = controller.buffer.read(&mut captured); - assert_eq!(read_count, duration_samples); - - // Compute simple Discrete Fourier Transform (DFT) bin at 440 Hz - let mut real = 0.0f32; - let mut imag = 0.0f32; - for (i, &sample) in captured.iter().enumerate() { - let angle = 2.0 * std::f32::consts::PI * freq * (i as f32) / (sample_rate as f32); - real += sample * angle.cos(); - imag -= sample * angle.sin(); - } - let magnitude = (real * real + imag * imag).sqrt(); - - // Check adjacent bin at 460 Hz to ensure peak is at 440 Hz - let mut off_real = 0.0f32; - let mut off_imag = 0.0f32; - let off_freq = 460.0f32; - for (i, &sample) in captured.iter().enumerate() { - let angle = 2.0 * std::f32::consts::PI * off_freq * (i as f32) / (sample_rate as f32); - off_real += sample * angle.cos(); - off_imag -= sample * angle.sin(); - } - let off_magnitude = (off_real * off_real + off_imag * off_imag).sqrt(); - - assert!( - magnitude > off_magnitude * 10.0, - "440 Hz peak must be dominant (>10x adjacent frequencies)" - ); }