From bbbe2d8f17cfdd869205bae6fb6f4a6156a50581 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 19 Jun 2026 03:57:41 -0400 Subject: [PATCH] Windows audio (cpal): startup handshake + RT-safe capture ring (W1, W3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two correctness fixes for the cpal/WASAPI backend from the Codex Windows-compat review, plus device logging. W1 — start_capture/start_playback no longer return Ok before the stream exists. The owning thread did device resolution, config selection, build_stream, and play() and only *logged* failures, so a missing 48 kHz config / unsupported format / WASAPI error left the UI in a joined-but-silent room. The worker now reports readiness over a channel and start_* blocks on it via finish_start(), returning the real AudioError on failure (and joining the dead worker). W3 — the RT capture callback no longer allocates or sends on a channel. It now only downmixes and wait-free-pushes mono samples into a preallocated lock-free HeapRb; the owning thread drains that ring, frames it (the Vec allocation lives off the RT path), and sends completed frames. A full ring increments an overrun counter instead of blocking. Restores the no-alloc/no-block-in-callback contract the PipeWire backend already honors. Also logs the selected device name / sample format / channels / rate on stream start (a review nice-to-have) and logs capture overruns when they occur. Windows-only file (cfg(windows)); Linux build unaffected. Compile-verified via the windows-gnu cross-build; not yet run on a real WASAPI host. Co-Authored-By: Claude Opus 4.8 --- src/audio/cpal_impl.rs | 233 +++++++++++++++++++++++++++++++---------- 1 file changed, 175 insertions(+), 58 deletions(-) diff --git a/src/audio/cpal_impl.rs b/src/audio/cpal_impl.rs index 1a042d0..e4e2fb8 100644 --- a/src/audio/cpal_impl.rs +++ b/src/audio/cpal_impl.rs @@ -7,6 +7,9 @@ //! //! - **Capture**: mono, 48 kHz, S16 PCM, emitted as `Vec` frames of //! [`CAPTURE_FRAME`] (960 = 20 ms) samples — matching the encoder/jitter frame. +//! The RT capture callback only downmixes and pushes samples into a lock-free +//! ring; the owning thread drains that ring, frames it, and sends — so the +//! callback never allocates, locks, or touches an mpsc channel. //! - **Playback**: stereo interleaved ([`PLAYBACK_CHANNELS`]) S16 PCM at 48 kHz, //! drained from a ring buffer that is paced to the device's hardware clock via //! `ring_fill` exactly as the PipeWire backend does. @@ -20,7 +23,12 @@ //! stream, plays it, and keeps it alive until the per-worker `running` flag flips //! (set by `stop`). The struct holds only `Send` handles (the flag + the join //! handle). The stream's RT callback does the actual audio work; the owning -//! thread additionally feeds the playback ring from the network mixer. +//! thread additionally feeds the playback ring (or drains the capture ring). +//! +//! `start_*` does not return until the owning thread reports back over a readiness +//! channel that the device resolved and the stream is built and playing — so a +//! device/format/WASAPI failure surfaces as a real `Err` to the caller instead of +//! leaving the UI in a joined-but-silent room. //! //! ## Sample rate //! @@ -30,7 +38,7 @@ //! sample-rate support (resampling) is a Phase 1.1 follow-up. use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; -use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender}; +use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender}; use std::sync::{Arc, Mutex}; use std::thread::{self, JoinHandle}; use std::time::Duration; @@ -49,6 +57,13 @@ const SAMPLE_RATE: u32 = 48_000; /// Mono capture frame: 960 samples = 20 ms @ 48 kHz. Matches the PipeWire backend /// and `core::jitter::FRAME_SAMPLES`. const CAPTURE_FRAME: usize = 960; +/// Lock-free capture ring capacity (mono samples) between the RT callback and the +/// owning drain thread: 8 frames = 160 ms of headroom, so a scheduling hiccup on +/// the drain thread doesn't immediately overrun the RT producer. +const CAPTURE_RING_CAPACITY: usize = CAPTURE_FRAME * 8; +/// How long the capture drain thread sleeps when the ring is momentarily empty, +/// before polling again. Small enough to stay well under the 20 ms frame cadence. +const CAPTURE_POLL: Duration = Duration::from_millis(5); /// Playback ring capacity in interleaved samples: 200 ms of stereo @ 48 kHz. /// Comfortably above [`PLAYBACK_TARGET_SAMPLES`] so the clock-paced producer has /// headroom and never has to drop frames in steady state. @@ -90,22 +105,20 @@ impl AudioBackend for CpalBackend { tx: Sender>, target_node: Option, ) -> Result<(), AudioError> { - let mut guard = self.capture.lock().unwrap(); + let guard = self.capture.lock().unwrap(); if guard.is_some() { return Err(AudioError::Stream("Capture already started".to_string())); } let running = Arc::new(AtomicBool::new(true)); let running_thread = running.clone(); + let (ready_tx, ready_rx) = channel::>(); let thread = thread::Builder::new() .name("peerspeak-cpal-capture".to_string()) .spawn(move || { - if let Err(e) = run_capture(tx, target_node, running_thread) { - crate::log_msg(&format!("cpal capture error: {e}")); - } + run_capture(tx, target_node, running_thread, ready_tx); }) .map_err(|e| AudioError::Init(e.to_string()))?; - *guard = Some(StreamWorker { running, thread }); - Ok(()) + finish_start(guard, StreamWorker { running, thread }, ready_rx, "capture") } fn start_playback( @@ -114,22 +127,20 @@ impl AudioBackend for CpalBackend { target_node: Option, ring_fill: Arc, ) -> Result<(), AudioError> { - let mut guard = self.playback.lock().unwrap(); + let guard = self.playback.lock().unwrap(); if guard.is_some() { return Err(AudioError::Stream("Playback already started".to_string())); } let running = Arc::new(AtomicBool::new(true)); let running_thread = running.clone(); + let (ready_tx, ready_rx) = channel::>(); let thread = thread::Builder::new() .name("peerspeak-cpal-playback".to_string()) .spawn(move || { - if let Err(e) = run_playback(rx, target_node, ring_fill, running_thread) { - crate::log_msg(&format!("cpal playback error: {e}")); - } + run_playback(rx, target_node, ring_fill, running_thread, ready_tx); }) .map_err(|e| AudioError::Init(e.to_string()))?; - *guard = Some(StreamWorker { running, thread }); - Ok(()) + finish_start(guard, StreamWorker { running, thread }, ready_rx, "playback") } fn stop(&self) -> Result<(), AudioError> { @@ -143,6 +154,38 @@ impl AudioBackend for CpalBackend { } } +/// Block until the just-spawned worker reports (over `ready_rx`) that its stream +/// is built and playing, then either install it (`Ok`) or join it and surface the +/// real error. This is what makes `start_capture`/`start_playback` fail loudly +/// instead of returning `Ok` into a joined-but-silent room (Codex review W1). +fn finish_start( + mut guard: std::sync::MutexGuard<'_, Option>, + worker: StreamWorker, + ready_rx: Receiver>, + what: &str, +) -> Result<(), AudioError> { + match ready_rx.recv() { + Ok(Ok(())) => { + *guard = Some(worker); + Ok(()) + } + // Setup failed (Err) or the worker exited before reporting (recv Err): + // either way it has stopped, so reap it and surface the error. + Ok(Err(e)) => { + worker.running.store(false, Ordering::Relaxed); + let _ = worker.thread.join(); + Err(e) + } + Err(_) => { + worker.running.store(false, Ordering::Relaxed); + let _ = worker.thread.join(); + Err(AudioError::Init(format!( + "cpal {what} worker exited before reporting readiness" + ))) + } + } +} + // --------------------------------------------------------------------------- // Device enumeration (for the settings device pickers) // --------------------------------------------------------------------------- @@ -285,53 +328,108 @@ fn run_capture( tx: Sender>, target: Option, running: Arc, -) -> Result<(), AudioError> { - let (device, config, sample_format) = resolve(false, target)?; - let channels = config.channels as usize; + ready: Sender>, +) { + // The RT callback pushes mono samples into this lock-free ring; we drain it on + // this (non-RT) thread, so the callback never allocates or sends on a channel. + let rb = HeapRb::::new(CAPTURE_RING_CAPACITY); + let (producer, mut consumer) = rb.split(); + let overrun = Arc::new(AtomicU64::new(0)); - let stream = match sample_format { - SampleFormat::F32 => build_input::(&device, &config, tx, channels), - SampleFormat::I16 => build_input::(&device, &config, tx, channels), - SampleFormat::U16 => build_input::(&device, &config, tx, channels), - other => Err(AudioError::Stream(format!( - "unsupported capture sample format: {other:?}" - ))), - }?; + // Fallible device/stream setup. We report the real error to `start_capture` + // before doing any work, so a join never lands in a silent room. + let setup = || -> Result<(Stream, String, SampleFormat, usize), AudioError> { + let (device, config, sample_format) = resolve(false, target)?; + let channels = config.channels as usize; + let stream = match sample_format { + SampleFormat::F32 => { + build_input::(&device, &config, producer, channels, overrun.clone()) + } + SampleFormat::I16 => { + build_input::(&device, &config, producer, channels, overrun.clone()) + } + SampleFormat::U16 => { + build_input::(&device, &config, producer, channels, overrun.clone()) + } + other => Err(AudioError::Stream(format!( + "unsupported capture sample format: {other:?}" + ))), + }?; + stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + let name = device.name().unwrap_or_else(|_| "".to_string()); + Ok((stream, name, sample_format, channels)) + }; - stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + let (stream, dev_name, sample_format, channels) = match setup() { + Ok(v) => { + let _ = ready.send(Ok(())); + v + } + Err(e) => { + let _ = ready.send(Err(e)); + return; + } + }; + crate::log_msg(&format!( + "cpal capture started: device='{dev_name}' format={sample_format:?} channels={channels} rate={SAMPLE_RATE} Hz" + )); - // The RT callback does the work; this thread just keeps `stream` alive until - // `stop()` flips the flag, at which point the stream is dropped (= stopped). + // Drain the RT ring on this thread: pop mono samples, frame them (the `Vec` + // allocation lives here, off the RT path), and send completed frames. Keep + // `stream` alive until `stop()` flips the flag. + let mut acc = FrameAccumulator::new(CAPTURE_FRAME); + let mut last_overrun = 0u64; while running.load(Ordering::Relaxed) { - thread::sleep(WORKER_POLL); + let mut drained = false; + while let Some(sample) = consumer.try_pop() { + drained = true; + if let Some(frame) = acc.push(sample) { + // Consumer gone (call ended) → stop feeding; the stream is + // dropped below on the way out. + if tx.send(frame).is_err() { + drop(stream); + return; + } + } + } + let o = overrun.load(Ordering::Relaxed); + if o != last_overrun { + crate::log_msg(&format!( + "cpal capture overrun: dropped {} samples (drain thread fell behind)", + o - last_overrun + )); + last_overrun = o; + } + if !drained { + thread::sleep(CAPTURE_POLL); + } } - Ok(()) + drop(stream); } -fn build_input( +fn build_input( device: &Device, config: &StreamConfig, - tx: Sender>, + mut producer: P, channels: usize, + overrun: Arc, ) -> Result where T: SizedSample + Send + 'static, i16: FromSample, + P: Producer + Send + 'static, { - let mut acc = FrameAccumulator::new(CAPTURE_FRAME); let err_fn = |e| crate::log_msg(&format!("cpal capture stream error: {e}")); device .build_input_stream::( config, move |data: &[T], _| { + // RT-safe: downmix + wait-free push only. A full ring means the + // drain thread stalled; count the drop and keep going. for frame in data.chunks_exact(channels) { let mono = downmix_to_mono(frame); - if let Some(full) = acc.push(mono) { - // Consumer gone (call ended) → stop feeding; the owning - // thread will drop the stream on `stop()`. - if tx.send(full).is_err() { - return; - } + if producer.try_push(mono).is_err() { + overrun.fetch_add(1, Ordering::Relaxed); } } }, @@ -393,9 +491,8 @@ fn run_playback( target: Option, ring_fill: Arc, running: Arc, -) -> Result<(), AudioError> { - let (device, config, sample_format) = resolve(true, target)?; - + ready: Sender>, +) { let rb = HeapRb::::new(RING_CAPACITY); let (mut producer, consumer) = rb.split(); @@ -413,22 +510,43 @@ fn run_playback( let underrun = Arc::new(AtomicU64::new(0)); let dropped = Arc::new(AtomicU64::new(0)); - let stream = match sample_format { - SampleFormat::F32 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) - } - SampleFormat::I16 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) - } - SampleFormat::U16 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) - } - other => Err(AudioError::Stream(format!( - "unsupported playback sample format: {other:?}" - ))), - }?; + // Fallible device/stream setup; report the real error to `start_playback` + // before any work so a failure surfaces instead of a silent room. `consumer` + // is moved into the output callback here. + let setup = || -> Result<(Stream, String, SampleFormat), AudioError> { + let (device, config, sample_format) = resolve(true, target)?; + let stream = match sample_format { + SampleFormat::F32 => { + build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) + } + SampleFormat::I16 => { + build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) + } + SampleFormat::U16 => { + build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) + } + other => Err(AudioError::Stream(format!( + "unsupported playback sample format: {other:?}" + ))), + }?; + stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + let name = device.name().unwrap_or_else(|_| "".to_string()); + Ok((stream, name, sample_format)) + }; - stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + let (stream, dev_name, sample_format) = match setup() { + Ok(v) => { + let _ = ready.send(Ok(())); + v + } + Err(e) => { + let _ = ready.send(Err(e)); + return; + } + }; + crate::log_msg(&format!( + "cpal playback started: device='{dev_name}' format={sample_format:?} channels={PLAYBACK_CHANNELS} rate={SAMPLE_RATE} Hz" + )); let logger = spawn_health_logger( running.clone(), @@ -456,7 +574,6 @@ fn run_playback( running.store(false, Ordering::Relaxed); let _ = logger.join(); drop(stream); - Ok(()) } fn build_output(