From 20bfcffe6d9f940672a5c4f4a1d426055440463b Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 19 Jun 2026 04:56:01 -0400 Subject: [PATCH] Complete Windows audio remap path --- docs/WINDOWS.md | 4 +- src/audio/cpal_impl.rs | 244 +++++++++++++++++++++++++++++++++-------- src/bin/audio_probe.rs | 134 ++++++++++++++++++++-- 3 files changed, 324 insertions(+), 58 deletions(-) diff --git a/docs/WINDOWS.md b/docs/WINDOWS.md index 0b0d711..37ca748 100644 --- a/docs/WINDOWS.md +++ b/docs/WINDOWS.md @@ -68,9 +68,9 @@ connections are expected and valid. | Echo cancellation | Linux-only PipeWire feature. The Windows UI shows it disabled as unavailable. | | Screen share | Requires a Windows `pixelpass.exe` on `PATH` or a configured override. | | Chimes | Now routed through Windows `SoundPlayer`; needs a real Windows host to audibly verify. | -| Resampling/device format | Open. Devices must support 48 kHz, and output must support stereo; a 44.1 kHz-only/default device currently errors instead of playing. | +| Resampling/device format | Cross-compiled. cpal/WASAPI now chooses native 48 kHz when available and otherwise resamples/remaps at the device boundary; needs real Windows hardware audio verification. | | Device persistence | Open. WASAPI friendly names may duplicate or change across driver/profile changes. | -| Playback pacing | Open. The fixed playback target under WASAPI shared mode still needs real-hardware verification. | +| Playback pacing | Cross-compiled. The fixed playback target under WASAPI shared mode still needs real-hardware verification with `audio_probe`. | Before calling Windows support done, verify a real Windows machine can create/join a room, capture mic audio, hear remote audio, select devices, restart with selections preserved, and diff --git a/src/audio/cpal_impl.rs b/src/audio/cpal_impl.rs index dec29b4..4a51156 100644 --- a/src/audio/cpal_impl.rs +++ b/src/audio/cpal_impl.rs @@ -52,16 +52,18 @@ //! seam ready for a higher-quality resampler later. use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; -use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender}; +use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel}; use std::sync::{Arc, Mutex}; use std::thread::{self, JoinHandle}; use std::time::Duration; use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; -use cpal::{Device, FromSample, Sample, SampleFormat, SampleRate, SizedSample, Stream, StreamConfig}; +use cpal::{ + Device, FromSample, Sample, SampleFormat, SampleRate, SizedSample, Stream, StreamConfig, +}; use ringbuf::{ - traits::{Consumer, Producer, Split}, HeapRb, + traits::{Consumer, Producer, Split}, }; use super::resample::{PushResampler, StereoPullResampler}; @@ -155,7 +157,12 @@ impl AudioBackend for CpalBackend { run_playback(rx, target_node, ring_fill, running_thread, ready_tx); }) .map_err(|e| AudioError::Init(e.to_string()))?; - finish_start(guard, StreamWorker { running, thread }, ready_rx, "playback") + finish_start( + guard, + StreamWorker { running, thread }, + ready_rx, + "playback", + ) } fn stop(&self) -> Result<(), AudioError> { @@ -296,7 +303,9 @@ fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option Option Result { +fn choose_config(device: &Device, output: bool) -> Result { let ranges: Vec = if output { device .supported_output_configs() @@ -398,7 +404,9 @@ fn run_capture( "unsupported capture sample format: {other:?}" ))), }?; - stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + 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, device_rate)) }; @@ -420,7 +428,8 @@ fn run_capture( // If the device isn't at 48 kHz, resample its mono stream up/down to 48 kHz on // this (non-RT) thread before framing (W4). At 48 kHz this stays None and the // samples pass straight through, bit-exact. - let mut resampler = (device_rate != SAMPLE_RATE).then(|| PushResampler::new(device_rate, SAMPLE_RATE)); + let mut resampler = + (device_rate != SAMPLE_RATE).then(|| PushResampler::new(device_rate, SAMPLE_RATE)); // Reused scratch for a sample's resampled output (off-RT alloc; tiny — at most // a couple of samples per input). Avoids a nested-closure borrow over `acc`/`tx`. let mut resampled: Vec = Vec::new(); @@ -590,28 +599,47 @@ fn run_playback( // 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 setup = || -> Result<(Stream, String, SampleFormat, usize, u32), AudioError> { let (device, config, sample_format) = resolve(true, target)?; + let channels = config.channels as usize; + let device_rate = config.sample_rate.0; let stream = match sample_format { - SampleFormat::F32 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone(), max_cb.clone()) - } - SampleFormat::I16 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone(), max_cb.clone()) - } - SampleFormat::U16 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone(), max_cb.clone()) - } + SampleFormat::F32 => build_output::( + &device, + &config, + consumer, + ring_fill.clone(), + underrun.clone(), + max_cb.clone(), + ), + SampleFormat::I16 => build_output::( + &device, + &config, + consumer, + ring_fill.clone(), + underrun.clone(), + max_cb.clone(), + ), + SampleFormat::U16 => build_output::( + &device, + &config, + consumer, + ring_fill.clone(), + underrun.clone(), + max_cb.clone(), + ), other => Err(AudioError::Stream(format!( "unsupported playback sample format: {other:?}" ))), }?; - stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + stream + .play() + .map_err(|e| AudioError::Stream(e.to_string()))?; let name = device.name().unwrap_or_else(|_| "".to_string()); - Ok((stream, name, sample_format)) + Ok((stream, name, sample_format, channels, device_rate)) }; - let (stream, dev_name, sample_format) = match setup() { + let (stream, dev_name, sample_format, channels, device_rate) = match setup() { Ok(v) => { let _ = ready.send(Ok(())); v @@ -622,7 +650,7 @@ fn run_playback( } }; crate::log_msg(&format!( - "cpal playback started: device='{dev_name}' format={sample_format:?} channels={PLAYBACK_CHANNELS} rate={SAMPLE_RATE} Hz" + "cpal playback started: device='{dev_name}' format={sample_format:?} channels={channels} device_rate={device_rate} Hz <- {SAMPLE_RATE} Hz" )); let logger = spawn_health_logger( @@ -667,27 +695,55 @@ where C: Consumer + Send + 'static, { let err_fn = |e| crate::log_msg(&format!("cpal playback stream error: {e}")); - device - .build_output_stream::( - config, - move |data: &mut [T], _| { - // Wait-free; the logger thread reads this off the RT path. - max_cb.fetch_max(data.len(), Ordering::Relaxed); - let (popped, starved) = fill_output(&mut consumer, data); - if starved > 0 { - underrun.fetch_add(starved, Ordering::Relaxed); - } - if popped > 0 { - // Decrement the exact occupancy by what we actually pulled - // (underruns removed nothing) so the mixer paces against the - // true ring depth. - ring_fill.fetch_sub(popped, Ordering::Relaxed); - } - }, - err_fn, - None, - ) - .map_err(|e| AudioError::Stream(e.to_string())) + let device_rate = config.sample_rate.0; + let device_channels = config.channels as usize; + if device_rate == SAMPLE_RATE && device_channels == PLAYBACK_CHANNELS { + device + .build_output_stream::( + config, + move |data: &mut [T], _| { + // Wait-free; the logger thread reads this off the RT path. + max_cb.fetch_max(data.len(), Ordering::Relaxed); + let (popped, starved) = fill_output(&mut consumer, data); + if starved > 0 { + underrun.fetch_add(starved, Ordering::Relaxed); + } + if popped > 0 { + // Decrement the exact occupancy by what we actually pulled + // (underruns removed nothing) so the mixer paces against the + // true ring depth. + ring_fill.fetch_sub(popped, Ordering::Relaxed); + } + }, + err_fn, + None, + ) + .map_err(|e| AudioError::Stream(e.to_string())) + } else { + let mut resampler = StereoPullResampler::new(SAMPLE_RATE, device_rate); + device + .build_output_stream::( + config, + move |data: &mut [T], _| { + // Wait-free; the logger thread reads this off the RT path. + max_cb.fetch_max(data.len(), Ordering::Relaxed); + let (popped, starved) = + fill_output_remap(&mut consumer, data, device_channels, &mut resampler); + if starved > 0 { + underrun.fetch_add(starved, Ordering::Relaxed); + } + if popped > 0 { + // Decrement the exact occupancy by what we actually pulled + // (underruns removed nothing) so the mixer paces against the + // true ring depth. + ring_fill.fetch_sub(popped, Ordering::Relaxed); + } + }, + err_fn, + None, + ) + .map_err(|e| AudioError::Stream(e.to_string())) + } } /// Drain the ring into the device buffer, substituting silence on underrun. @@ -714,6 +770,60 @@ where (popped, starved) } +/// Resample/remap internal 48 kHz stereo ring samples into the device buffer. +/// Returns `(internal_samples_popped, device_samples_starved)`. RT-safe. +fn fill_output_remap( + consumer: &mut C, + out: &mut [T], + device_channels: usize, + resampler: &mut StereoPullResampler, +) -> (usize, u64) +where + T: Sample + FromSample, + C: Consumer, +{ + let mut popped = 0usize; + let mut starved = 0u64; + for frame in out.chunks_mut(device_channels) { + match resampler.next(|| { + let l = match consumer.try_pop() { + Some(v) => { + popped += 1; + v + } + None => return None, + }; + let r = match consumer.try_pop() { + Some(v) => { + popped += 1; + v + } + None => return None, + }; + Some((i16_to_f32(l), i16_to_f32(r))) + }) { + Some((l, r)) => { + if device_channels == 1 { + frame[0] = T::from_sample(f32_to_i16((l + r) * 0.5)); + } else { + frame[0] = T::from_sample(f32_to_i16(l)); + frame[1] = T::from_sample(f32_to_i16(r)); + for slot in &mut frame[2..] { + *slot = T::from_sample(0i16); + } + } + } + None => { + for slot in frame { + *slot = T::from_sample(0i16); + } + starved += device_channels as u64; + } + } + } + (popped, starved) +} + /// Once-per-second playout-health line (mirrors the PipeWire backend). Quiet /// unless a second actually glitched, or `PEERSPEAK_AUDIO_VERBOSE` is set. fn spawn_health_logger( @@ -819,6 +929,48 @@ mod tests { assert_eq!(out, [1, 2, 3, 0, 0]); } + #[test] + fn fill_output_remap_downmixes_to_mono() { + let rb = HeapRb::::new(8); + let (mut prod, mut cons) = rb.split(); + for v in [100, 300, 500, -100, 7, 9] { + prod.try_push(v).unwrap(); + } + let mut out = [0i16; 2]; + let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE); + let (popped, starved) = fill_output_remap(&mut cons, &mut out, 1, &mut resampler); + assert_eq!(popped, 6); + assert_eq!(starved, 0); + assert_eq!(out, [200, 200]); + } + + #[test] + fn fill_output_remap_silences_underrun() { + let rb = HeapRb::::new(8); + let (_prod, mut cons) = rb.split(); + let mut out = [11i16; 4]; + let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE); + let (popped, starved) = fill_output_remap(&mut cons, &mut out, 2, &mut resampler); + assert_eq!(popped, 0); + assert_eq!(starved, out.len() as u64); + assert_eq!(out, [0, 0, 0, 0]); + } + + #[test] + fn fill_output_remap_copies_stereo_at_matching_rate() { + let rb = HeapRb::::new(8); + let (mut prod, mut cons) = rb.split(); + for v in [1, -1, 2, -2, 3, -3] { + prod.try_push(v).unwrap(); + } + let mut out = [0i16; 4]; + let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE); + let (popped, starved) = fill_output_remap(&mut cons, &mut out, 2, &mut resampler); + assert_eq!(popped, 6); + assert_eq!(starved, 0); + assert_eq!(out, [1, -1, 2, -2]); + } + #[test] fn drain_loop_exits_when_running_flips_even_with_sender_alive() { let (tx, rx) = mpsc::channel::>(); diff --git a/src/bin/audio_probe.rs b/src/bin/audio_probe.rs index b08712d..f7429a3 100644 --- a/src/bin/audio_probe.rs +++ b/src/bin/audio_probe.rs @@ -1,11 +1,11 @@ //! Audio playout diagnostic probe. //! -//! Drives a phase-continuous sine tone through the *real* PipeWire playback path -//! (`PipeWireBackend::start_playback`), using the *same* fill-paced production -//! the production mixer uses (`core/mod.rs`): generate a frame only while the +//! Drives a phase-continuous sine tone through the *real* playback path +//! (PipeWire on Linux, cpal/WASAPI on Windows), using the *same* fill-paced +//! production the production mixer uses (`core/mod.rs`): generate a frame only while the //! playback ring is below `PLAYBACK_TARGET_SAMPLES`, so production tracks the -//! PipeWire hardware clock. No network, no microphone — this isolates the local -//! output path so we can confirm the clock-paced playout is glitch-free. +//! hardware clock. No network, no microphone — this isolates the local output +//! path so we can confirm the clock-paced playout is glitch-free. //! //! Use your ears on the tone (any click/pop is a glitch) together with the //! `playout-health:` lines tailed to stdout: @@ -18,17 +18,24 @@ //! Run: cargo run --bin audio_probe -- [freq_hz] [seconds] [target_node] //! e.g. cargo run --release --bin audio_probe -- 440 30 //! -//! This probe exercises the PipeWire backend directly, so it is a Linux-only tool. -//! On non-Linux targets `main` is a stub that explains the limitation. +//! This probe exercises the platform playback backend directly: PipeWire on Linux +//! and cpal/WASAPI on Windows. Other targets use a stub that explains the limitation. #[cfg(target_os = "linux")] fn main() { unix_probe::run(); } -#[cfg(not(target_os = "linux"))] +#[cfg(windows)] fn main() { - eprintln!("audio_probe is only supported on Linux builds (it drives the PipeWire backend directly)."); + win_probe::run(); +} + +#[cfg(not(any(target_os = "linux", windows)))] +fn main() { + eprintln!( + "audio_probe is only supported on Linux and Windows builds (it drives the platform playback backend directly)." + ); } #[cfg(target_os = "linux")] @@ -88,7 +95,114 @@ mod unix_probe { for _ in 0..FRAME_SAMPLES { let t = n as f32 / SAMPLE_RATE; // 0.25 amplitude: clearly audible but not harsh. - let sample = (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16; + let sample = + (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16; + // Stereo playback bus: duplicate the probe tone to L/R. + frame.push(sample); + frame.push(sample); + n += 1; + } + if tx.send(frame).is_err() { + eprintln!("playback channel closed early"); + break; + } + } + + // Let the ring drain, then stop. + tokio::time::sleep(Duration::from_millis(300)).await; + let _ = backend.stop(); + println!("\naudio_probe: done."); + } + + /// Open the app log, seek to the end, and echo new lines (the `playout-health:` + /// reports) to stdout once they appear. + fn spawn_log_tailer() { + let path = peerspeak::log_file_path(); + std::thread::spawn(move || { + // Wait for the file to exist (first log_msg creates it). + let file = loop { + if let Ok(f) = std::fs::File::open(&path) { + break f; + } + std::thread::sleep(Duration::from_millis(100)); + }; + let mut reader = BufReader::new(file); + let _ = reader.seek(SeekFrom::End(0)); + loop { + let mut line = String::new(); + match reader.read_line(&mut line) { + Ok(0) => std::thread::sleep(Duration::from_millis(150)), + Ok(_) => { + if line.contains("playout-health:") { + print!("{line}"); + } + } + Err(_) => std::thread::sleep(Duration::from_millis(150)), + } + } + }); + } +} + +#[cfg(windows)] +mod win_probe { + use std::io::{BufRead, BufReader, Seek, SeekFrom}; + use std::sync::Arc; + use std::sync::atomic::AtomicUsize; + use std::sync::mpsc; + use std::time::Duration; + + use peerspeak::audio::AudioBackend; + use peerspeak::audio::cpal_impl::CpalBackend; + use peerspeak::core::jitter::FRAME_SAMPLES; // 960 mono frames = 20ms @ 48kHz + + const SAMPLE_RATE: f32 = 48_000.0; + + #[tokio::main] + pub async fn run() { + let mut args = std::env::args().skip(1); + let freq: f32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(440.0); + let secs: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(30); + let target_node: Option = args.next(); + + // The playout-health logger is quiet in normal operation (it only logs + // glitches); ask it for the full once-per-second heartbeat so the probe can + // show the steady-state numbers. + // SAFETY: set before any playback thread starts, so no concurrent env read. + unsafe { std::env::set_var("PEERSPEAK_AUDIO_VERBOSE", "1") }; + + println!("audio_probe: {freq} Hz tone for {secs}s through the real playback path."); + println!("Listen for clicks/pops; watch the playout-health lines below.\n"); + + // Tail the app log (where playout-health lines land) to stdout in the + // background so it's all in one terminal. + spawn_log_tailer(); + + let backend = CpalBackend::new(); + let (tx, rx) = mpsc::channel::>(); + let ring_fill = Arc::new(AtomicUsize::new(0)); + if let Err(e) = backend.start_playback(rx, target_node, ring_fill.clone()) { + eprintln!("failed to start playback: {e}"); + return; + } + + // Phase-continuous sine, generated one 20ms frame at a time, fill-paced + // exactly like the production mixer: only produce while the ring is below + // target, so production tracks the cpal/WASAPI hardware clock. + use std::sync::atomic::Ordering; + let deadline = tokio::time::Instant::now() + Duration::from_secs(secs); + let mut n: u64 = 0; // running sample index keeps phase continuous across frames + while tokio::time::Instant::now() < deadline { + if ring_fill.load(Ordering::Relaxed) >= peerspeak::audio::PLAYBACK_TARGET_SAMPLES { + tokio::time::sleep(Duration::from_millis(2)).await; + continue; + } + let mut frame = Vec::with_capacity(FRAME_SAMPLES * peerspeak::audio::PLAYBACK_CHANNELS); + for _ in 0..FRAME_SAMPLES { + let t = n as f32 / SAMPLE_RATE; + // 0.25 amplitude: clearly audible but not harsh. + let sample = + (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16; // Stereo playback bus: duplicate the probe tone to L/R. frame.push(sample); frame.push(sample);