molluskandClaude Opus 4.8
d0a16cb8b9
The repo never enforced rustfmt, so formatting had drifted broadly. This is a single mechanical `cargo fmt` pass over the whole crate (no behavioral change; lib suite green, 493 passed). Going forward fmt should be enforced (planned CI fmt --check step). Part of the 0.6.1 hygiene pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1487 lines
60 KiB
Rust
1487 lines
60 KiB
Rust
//! Windows audio backend — cpal / WASAPI (Phase 1).
|
||
//!
|
||
//! Implements [`AudioBackend`] on top of [`cpal`], which wraps WASAPI on Windows.
|
||
//! It is the Windows counterpart to `pipewire_impl.rs` and deliberately preserves
|
||
//! the exact same contract so the rest of the app (mixer, encoder, jitter buffer)
|
||
//! is unchanged:
|
||
//!
|
||
//! - **Capture**: mono, 48 kHz, S16 PCM, emitted as `Vec<i16>` 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.
|
||
//!
|
||
//! ## Threading and the `!Send` stream
|
||
//!
|
||
//! `cpal::Stream` is `!Send` (some backends require it to be created and dropped
|
||
//! on the same thread), but [`AudioBackend`] is `Send + Sync` and the backend is
|
||
//! shared through an `Arc`. So the stream never lives in the struct: each of
|
||
//! `start_capture`/`start_playback` spawns one owning thread that builds the
|
||
//! 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 (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 and channel layout (W4)
|
||
//!
|
||
//! The whole pipeline runs internally at 48 kHz (Opus + the 960-sample frame) and
|
||
//! mono capture / stereo playback. We prefer a native-48 kHz device config so the
|
||
//! common case is conversion-free and bit-exact. When the device can't do 48 kHz
|
||
//! (commonly a 44.1 kHz-only endpoint) or can't do stereo output, we fall back to
|
||
//! the device's default config and convert at the boundary with the dep-free
|
||
//! [`super::resample`] linear resamplers instead of hard-erroring:
|
||
//!
|
||
//! - **Capture**: the device-rate mono stream is resampled to 48 kHz on the
|
||
//! capture drain thread (off the RT callback) before framing.
|
||
//! - **Playback**: the internal 48 kHz stereo bus is resampled to the device rate
|
||
//! and remapped to the device channel count inside the output RT callback, which
|
||
//! pulls internal frames from the ring on demand (allocation-free, so RT-safe).
|
||
//! The ring, prefill, and `ring_fill` pacing stay in internal 48 kHz-stereo
|
||
//! units, so the mixer is unchanged.
|
||
//!
|
||
//! Linear interpolation has no anti-aliasing filter (see [`super::resample`] docs);
|
||
//! it is adequate for speech and keeps the matching-rate path bit-exact, with the
|
||
//! seam ready for a higher-quality resampler later.
|
||
|
||
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, AtomicUsize, Ordering};
|
||
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
|
||
use std::sync::{Arc, Mutex};
|
||
use std::thread::{self, JoinHandle};
|
||
use std::time::{Duration, Instant};
|
||
|
||
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
|
||
use cpal::{
|
||
Device, FromSample, Sample, SampleFormat, SampleRate, SizedSample, Stream, StreamConfig,
|
||
};
|
||
use ringbuf::{
|
||
HeapRb,
|
||
traits::{Consumer, Producer, Split},
|
||
};
|
||
|
||
use super::resample::{PushResampler, StereoPullResampler};
|
||
use super::{AudioBackend, AudioDevice, AudioError, PLAYBACK_CHANNELS, PLAYBACK_TARGET_SAMPLES};
|
||
|
||
/// The one sample rate the pipeline supports (Opus + the 20 ms frame).
|
||
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.
|
||
const RING_CAPACITY: usize = 9600 * PLAYBACK_CHANNELS;
|
||
/// How often a blocked playback worker re-checks its `running` flag, bounding how
|
||
/// long `stop()` can take to join it (mirrors the PipeWire backend's `WORKER_POLL`).
|
||
const WORKER_POLL: Duration = Duration::from_millis(100);
|
||
/// How long a freshly-played stream has to prove itself (deliver its first RT
|
||
/// callbacks) before the start is treated as failed. cpal's `play()` only *queues*
|
||
/// the WASAPI `Start()`, so a queued-but-failed start would otherwise masquerade as
|
||
/// success and join the UI into a silent room (review W1).
|
||
const STREAM_START_TIMEOUT: Duration = Duration::from_secs(3);
|
||
/// Number of completed RT callbacks the owner waits for before declaring the stream
|
||
/// live. One callback isn't proof: a stream can fire once and immediately fail in
|
||
/// the same processing cycle, so requiring a couple of cycles (plus the terminal
|
||
/// error check) keeps a one-shot-then-dead stream from being reported Ok (B1).
|
||
const MIN_START_CALLBACKS: usize = 2;
|
||
/// Backstop for [`finish_start`]: bounds the WHOLE owner path (resolve + build +
|
||
/// play + the [`STREAM_START_TIMEOUT`] callback wait + any wedged-stream cleanup).
|
||
/// Sized as a generous setup budget plus the callback wait plus slack so a slow but
|
||
/// valid device (e.g. a Bluetooth endpoint that takes seconds to spin up) is not
|
||
/// falsely failed, while a driver that wedges before the owner can report is still
|
||
/// released eventually (review W6, B4).
|
||
const FINISH_START_TIMEOUT: Duration = Duration::from_secs(10);
|
||
/// Lowest / highest device sample rate the backend will drive. The floor bounds
|
||
/// the playback pull-resampler's input-pulls-per-output-frame (≈48000/rate) so a
|
||
/// pathological low rate can't blow the RT callback's deadline; the ceiling and a
|
||
/// nonzero floor also reject the 0 Hz / absurd values a misbehaving driver could
|
||
/// report, which would otherwise panic or spin (review W7).
|
||
const MIN_DEVICE_RATE: u32 = 8_000;
|
||
const MAX_DEVICE_RATE: u32 = 384_000;
|
||
|
||
// Stream-error categories carried from the RT error callback to the owner thread
|
||
// through an `AtomicU8`, so the callback itself never allocates or logs — both of
|
||
// which it previously did via `format!`/`log_msg` on the time-critical stream
|
||
// thread (review W2). The owner/logger translates the code off the RT path.
|
||
const STREAM_ERR_NONE: u8 = 0;
|
||
const STREAM_ERR_DEVICE_UNAVAILABLE: u8 = 1;
|
||
const STREAM_ERR_BACKEND: u8 = 2;
|
||
|
||
/// Map a cpal stream error to its [`STREAM_ERR_*`](STREAM_ERR_NONE) code. Pure +
|
||
/// allocation-free, so it is safe to call from the RT error callback.
|
||
fn stream_err_code(e: &cpal::StreamError) -> u8 {
|
||
match e {
|
||
cpal::StreamError::DeviceNotAvailable => STREAM_ERR_DEVICE_UNAVAILABLE,
|
||
_ => STREAM_ERR_BACKEND,
|
||
}
|
||
}
|
||
|
||
/// Human-readable text for a [`STREAM_ERR_*`](STREAM_ERR_NONE) code, logged off the
|
||
/// RT path by the owner/health-logger thread.
|
||
fn stream_err_text(code: u8) -> &'static str {
|
||
match code {
|
||
STREAM_ERR_DEVICE_UNAVAILABLE => "audio device became unavailable",
|
||
_ => "audio backend stream error",
|
||
}
|
||
}
|
||
|
||
/// Wait for a just-played stream to prove it actually started: its RT data
|
||
/// callback bumps `callbacks`, or an error callback sets `err_code`. Returns `Ok`
|
||
/// once [`MIN_START_CALLBACKS`] cycles have run with no error, `Err` on an
|
||
/// error-callback code or [`STREAM_START_TIMEOUT`], or a clean abort if `stop()`
|
||
/// cleared `running` mid-start. Polls a few cheap atomics on the owner thread —
|
||
/// never the RT thread (review W1).
|
||
///
|
||
/// The error is **terminal and wins any race** with `callbacks`: a stream can run a
|
||
/// callback and then fail in the same processing cycle, so `err_code` is checked
|
||
/// first each loop AND re-checked before declaring success (review B1).
|
||
fn wait_for_stream_start(
|
||
callbacks: &AtomicUsize,
|
||
err_code: &AtomicU8,
|
||
running: &AtomicBool,
|
||
) -> Result<(), AudioError> {
|
||
let deadline = Instant::now() + STREAM_START_TIMEOUT;
|
||
let as_err = |code: u8| Err(AudioError::Stream(stream_err_text(code).to_string()));
|
||
loop {
|
||
let code = err_code.load(Ordering::Relaxed);
|
||
if code != STREAM_ERR_NONE {
|
||
return as_err(code);
|
||
}
|
||
if callbacks.load(Ordering::Relaxed) >= MIN_START_CALLBACKS {
|
||
// Re-check: a callback that pushed us to the threshold may have been the
|
||
// last before a same-cycle failure. Let a terminal error win.
|
||
let code = err_code.load(Ordering::Relaxed);
|
||
if code != STREAM_ERR_NONE {
|
||
return as_err(code);
|
||
}
|
||
return Ok(());
|
||
}
|
||
if !running.load(Ordering::Relaxed) {
|
||
return Err(AudioError::Stream("stream start aborted".to_string()));
|
||
}
|
||
if Instant::now() >= deadline {
|
||
return Err(AudioError::Stream(
|
||
"stream did not start within timeout (no WASAPI callback)".to_string(),
|
||
));
|
||
}
|
||
thread::sleep(Duration::from_millis(5));
|
||
}
|
||
}
|
||
|
||
/// Windows audio backend. See module docs.
|
||
pub struct CpalBackend {
|
||
capture: Mutex<SlotState>,
|
||
playback: Mutex<SlotState>,
|
||
}
|
||
|
||
/// A spawned owning thread plus the flags that coordinate its lifetime: `running`
|
||
/// tells it to drop its stream and exit; `exited` is flipped true (by [`ExitGuard`]
|
||
/// in the thread body) when it actually returns, so a *detached* wedged start can be
|
||
/// detected as finished later (review B3).
|
||
struct StreamWorker {
|
||
running: Arc<AtomicBool>,
|
||
exited: Arc<AtomicBool>,
|
||
thread: JoinHandle<()>,
|
||
}
|
||
|
||
/// Flips its flag true when dropped, marking a worker thread as exited. Lives at the
|
||
/// top of the worker closure so it fires on normal return, panic unwind, or whenever
|
||
/// a wedged driver call finally releases the thread — which is what lets a [`SlotState::Wedged`]
|
||
/// tombstone (B3) know its orphan is gone.
|
||
struct ExitGuard(Arc<AtomicBool>);
|
||
impl Drop for ExitGuard {
|
||
fn drop(&mut self) {
|
||
self.0.store(true, Ordering::Relaxed);
|
||
}
|
||
}
|
||
|
||
/// The lifecycle state of a capture or playback slot.
|
||
enum SlotState {
|
||
/// No stream — a new start may proceed.
|
||
Idle,
|
||
/// A live, started stream owned by its worker thread.
|
||
Live(StreamWorker),
|
||
/// A start that timed out wedged in a driver call (review B3). Its worker thread
|
||
/// was *detached* rather than joined — joining would re-introduce the unbounded
|
||
/// hang [`FINISH_START_TIMEOUT`] exists to prevent — so it may still be alive,
|
||
/// holding the COM/device handle. `exited` flips true when that orphan finally
|
||
/// returns. New starts are rejected until then, so retries against a permanently
|
||
/// wedged device don't pile up more orphan threads.
|
||
Wedged { exited: Arc<AtomicBool> },
|
||
}
|
||
|
||
/// Inspect a slot before starting a stream into it. Clears a [`SlotState::Wedged`]
|
||
/// tombstone whose orphan has since exited (the slot becomes reusable), but rejects
|
||
/// a start while a wedged orphan is still alive or a live stream already owns the
|
||
/// slot. Pure w.r.t. the passed state, so the tombstone logic is unit-testable (B3).
|
||
fn ensure_idle(state: &mut SlotState, what: &str) -> Result<(), AudioError> {
|
||
match state {
|
||
SlotState::Idle => Ok(()),
|
||
SlotState::Live(_) => Err(AudioError::Stream(format!("{what} already started"))),
|
||
SlotState::Wedged { exited } => {
|
||
if exited.load(Ordering::Relaxed) {
|
||
*state = SlotState::Idle;
|
||
Ok(())
|
||
} else {
|
||
Err(AudioError::Stream(format!(
|
||
"{what} is recovering from an unresponsive audio device; retry shortly"
|
||
)))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
impl CpalBackend {
|
||
pub fn new() -> Self {
|
||
Self {
|
||
capture: Mutex::new(SlotState::Idle),
|
||
playback: Mutex::new(SlotState::Idle),
|
||
}
|
||
}
|
||
}
|
||
|
||
impl Default for CpalBackend {
|
||
fn default() -> Self {
|
||
Self::new()
|
||
}
|
||
}
|
||
|
||
impl AudioBackend for CpalBackend {
|
||
fn start_capture(
|
||
&self,
|
||
tx: Sender<Vec<i16>>,
|
||
target_node: Option<String>,
|
||
) -> Result<(), AudioError> {
|
||
let mut guard = self.capture.lock().unwrap();
|
||
ensure_idle(&mut guard, "capture")?;
|
||
let running = Arc::new(AtomicBool::new(true));
|
||
let exited = Arc::new(AtomicBool::new(false));
|
||
let running_thread = running.clone();
|
||
let exited_thread = exited.clone();
|
||
let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>();
|
||
let thread = thread::Builder::new()
|
||
.name("peerspeak-cpal-capture".to_string())
|
||
.spawn(move || {
|
||
let _exit = ExitGuard(exited_thread);
|
||
run_capture(tx, target_node, running_thread, ready_tx);
|
||
})
|
||
.map_err(|e| AudioError::Init(e.to_string()))?;
|
||
finish_start(
|
||
guard,
|
||
StreamWorker {
|
||
running,
|
||
exited,
|
||
thread,
|
||
},
|
||
ready_rx,
|
||
"capture",
|
||
)
|
||
}
|
||
|
||
fn start_playback(
|
||
&self,
|
||
rx: Receiver<Vec<i16>>,
|
||
target_node: Option<String>,
|
||
ring_fill: Arc<AtomicUsize>,
|
||
) -> Result<(), AudioError> {
|
||
let mut guard = self.playback.lock().unwrap();
|
||
ensure_idle(&mut guard, "playback")?;
|
||
let running = Arc::new(AtomicBool::new(true));
|
||
let exited = Arc::new(AtomicBool::new(false));
|
||
let running_thread = running.clone();
|
||
let exited_thread = exited.clone();
|
||
let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>();
|
||
let thread = thread::Builder::new()
|
||
.name("peerspeak-cpal-playback".to_string())
|
||
.spawn(move || {
|
||
let _exit = ExitGuard(exited_thread);
|
||
run_playback(rx, target_node, ring_fill, running_thread, ready_tx);
|
||
})
|
||
.map_err(|e| AudioError::Init(e.to_string()))?;
|
||
finish_start(
|
||
guard,
|
||
StreamWorker {
|
||
running,
|
||
exited,
|
||
thread,
|
||
},
|
||
ready_rx,
|
||
"playback",
|
||
)
|
||
}
|
||
|
||
fn stop(&self) -> Result<(), AudioError> {
|
||
for slot in [&self.capture, &self.playback] {
|
||
let mut guard = slot.lock().unwrap();
|
||
match std::mem::replace(&mut *guard, SlotState::Idle) {
|
||
SlotState::Live(worker) => {
|
||
worker.running.store(false, Ordering::Relaxed);
|
||
let _ = worker.thread.join();
|
||
}
|
||
// A wedged orphan was detached and can't be joined. If it has since
|
||
// exited the slot is now clear; otherwise restore the tombstone so a
|
||
// later start still sees the device is recovering (B3).
|
||
SlotState::Wedged { exited } => {
|
||
if !exited.load(Ordering::Relaxed) {
|
||
*guard = SlotState::Wedged { exited };
|
||
}
|
||
}
|
||
SlotState::Idle => {}
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
/// 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<'_, SlotState>,
|
||
worker: StreamWorker,
|
||
ready_rx: Receiver<Result<(), AudioError>>,
|
||
what: &str,
|
||
) -> Result<(), AudioError> {
|
||
// Bounded wait. An unbounded `recv()` here would hang `start_*` forever — and
|
||
// any concurrent `stop()` behind the same slot mutex — if a WASAPI/driver call
|
||
// wedged the worker before it could report (review W6).
|
||
match ready_rx.recv_timeout(FINISH_START_TIMEOUT) {
|
||
Ok(Ok(())) => {
|
||
*guard = SlotState::Live(worker);
|
||
Ok(())
|
||
}
|
||
// Setup failed (Err) or the worker disconnected before reporting: 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(RecvTimeoutError::Disconnected) => {
|
||
worker.running.store(false, Ordering::Relaxed);
|
||
let _ = worker.thread.join();
|
||
Err(AudioError::Init(format!(
|
||
"cpal {what} worker exited before reporting readiness"
|
||
)))
|
||
}
|
||
Err(RecvTimeoutError::Timeout) => {
|
||
// The worker is wedged in a driver call. Signal it to exit, but DETACH
|
||
// rather than join — joining would re-introduce the unbounded hang this
|
||
// timeout exists to prevent. Leave a Wedged tombstone so subsequent
|
||
// starts are rejected until the orphan's ExitGuard flips `exited`, rather
|
||
// than spawning more orphan threads against the same dead device (B3).
|
||
let StreamWorker {
|
||
running,
|
||
exited,
|
||
thread,
|
||
} = worker;
|
||
running.store(false, Ordering::Relaxed);
|
||
drop(thread);
|
||
*guard = SlotState::Wedged { exited };
|
||
Err(AudioError::Init(format!(
|
||
"cpal {what} did not start within {FINISH_START_TIMEOUT:?}"
|
||
)))
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Device enumeration (for the settings device pickers)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Enumerate WASAPI input/output devices via cpal, sorted by description to match
|
||
/// the PipeWire backend's stable UI ordering.
|
||
///
|
||
/// cpal exposes a single friendly name per device, which is also what [`resolve`]
|
||
/// matches `target_node` against — so `name` and `description` are the same string
|
||
/// and a saved selection round-trips. Note: WASAPI device names are less stable
|
||
/// across driver/endpoint changes than PipeWire node names, so a saved device may
|
||
/// not always be found again; selection then falls back to the system default.
|
||
pub fn enumerate_audio_devices() -> Vec<AudioDevice> {
|
||
let host = cpal::default_host();
|
||
let mut devices = Vec::new();
|
||
|
||
if let Ok(inputs) = host.input_devices() {
|
||
for device in inputs {
|
||
if let Ok(name) = device.name() {
|
||
devices.push(AudioDevice {
|
||
description: name.clone(),
|
||
name,
|
||
is_input: true,
|
||
});
|
||
}
|
||
}
|
||
}
|
||
if let Ok(outputs) = host.output_devices() {
|
||
for device in outputs {
|
||
if let Ok(name) = device.name() {
|
||
devices.push(AudioDevice {
|
||
description: name.clone(),
|
||
name,
|
||
is_input: false,
|
||
});
|
||
}
|
||
}
|
||
}
|
||
|
||
devices.sort_by(|a, b| a.description.cmp(&b.description));
|
||
devices
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Device / config selection
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Resolve a device (by `target` name, else the system default) and a stream
|
||
/// config. We prefer a config running natively at [`SAMPLE_RATE`] (conversion-free);
|
||
/// if the device has none, we fall back to its default config and resample/remap at
|
||
/// the boundary (W4 — see module docs and [`choose_config`]).
|
||
fn resolve(
|
||
output: bool,
|
||
target: Option<String>,
|
||
) -> Result<(Device, StreamConfig, SampleFormat), AudioError> {
|
||
let host = cpal::default_host();
|
||
|
||
let default = || {
|
||
if output {
|
||
host.default_output_device()
|
||
} else {
|
||
host.default_input_device()
|
||
}
|
||
};
|
||
let device = match target {
|
||
// A saved device name that no longer resolves falls back to the system
|
||
// default — but log it, because WASAPI friendly names can change across
|
||
// driver/endpoint changes, so a silent fallback otherwise looks like
|
||
// "audio went to the wrong device for no reason" (review W7).
|
||
Some(ref name) => match find_device_by_name(&host, output, name) {
|
||
Some(dev) => Some(dev),
|
||
None => {
|
||
crate::log_msg(&format!(
|
||
"cpal: saved {} device '{name}' not found; using system default",
|
||
if output { "output" } else { "input" },
|
||
));
|
||
default()
|
||
}
|
||
},
|
||
None => default(),
|
||
}
|
||
.ok_or_else(|| AudioError::Device("no audio device available".to_string()))?;
|
||
|
||
let supported = choose_config(&device, output)?;
|
||
let sample_format = supported.sample_format();
|
||
let config = supported.config();
|
||
|
||
// Validate the OS-reported geometry before any code divides by it or sizes a
|
||
// loop from it (review W7). Zero channels would panic `chunks_exact(0)` /
|
||
// `chunks_mut(0)`; a zero or absurd rate would yield an infinite/huge resample
|
||
// ratio. Reject up front with a real error instead of panicking or spinning.
|
||
if config.channels == 0 {
|
||
return Err(AudioError::Device(
|
||
"audio device reports zero channels".to_string(),
|
||
));
|
||
}
|
||
if !(MIN_DEVICE_RATE..=MAX_DEVICE_RATE).contains(&config.sample_rate.0) {
|
||
return Err(AudioError::Device(format!(
|
||
"audio device sample rate {} Hz is outside the supported {MIN_DEVICE_RATE}–{MAX_DEVICE_RATE} Hz range",
|
||
config.sample_rate.0,
|
||
)));
|
||
}
|
||
Ok((device, config, sample_format))
|
||
}
|
||
|
||
fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option<Device> {
|
||
let devices = if output {
|
||
host.output_devices().ok()?
|
||
} else {
|
||
host.input_devices().ok()?
|
||
};
|
||
devices
|
||
.into_iter()
|
||
.find(|d| d.name().is_ok_and(|n| n == name))
|
||
}
|
||
|
||
/// Whether the backend can actually open this config range. The workers only build
|
||
/// `F32`/`I16`/`U16` streams ([`build_input`]/[`build_output`] — every other sample
|
||
/// format hits the `other => Err(...)` arm), and a zero-channel range would later be
|
||
/// rejected by [`resolve`]'s geometry check. Filtering both here keeps
|
||
/// [`choose_config`] from *ranking* a range it can't drive ahead of a usable one and
|
||
/// then hard-failing the start instead of trying the next candidate (Codex B5
|
||
/// re-review, P3).
|
||
fn usable_range(r: &cpal::SupportedStreamConfigRange) -> bool {
|
||
r.channels() > 0 && format_supported(r.sample_format())
|
||
}
|
||
|
||
/// Sample formats the capture/playback stream builders accept. Pure, so it's
|
||
/// unit-testable independently of the cpal range types.
|
||
fn format_supported(fmt: SampleFormat) -> bool {
|
||
matches!(
|
||
fmt,
|
||
SampleFormat::F32 | SampleFormat::I16 | SampleFormat::U16
|
||
)
|
||
}
|
||
|
||
/// Pick a sample rate inside both a device's supported `[r_min, r_max]` span and the
|
||
/// backend's drivable `[MIN_DEVICE_RATE, MAX_DEVICE_RATE]` window, preferring
|
||
/// [`SAMPLE_RATE`] when it's reachable and otherwise the nearest in-window bound.
|
||
/// Returns `None` when the device span doesn't overlap the window at all. Pure and
|
||
/// integer-only, so the selection policy is unit-testable (review B5).
|
||
fn bounded_rate(r_min: u32, r_max: u32) -> Option<u32> {
|
||
let lo = r_min.max(MIN_DEVICE_RATE);
|
||
let hi = r_max.min(MAX_DEVICE_RATE);
|
||
(lo <= hi).then(|| SAMPLE_RATE.clamp(lo, hi))
|
||
}
|
||
|
||
/// Pick a stream config. Preference order, best (no conversion) first:
|
||
/// 1. exactly [`SAMPLE_RATE`] at the preferred layout (stereo out / mono in),
|
||
/// 2. exactly [`SAMPLE_RATE`] at any channel count (rate-exact, backend remaps),
|
||
/// 3. a supported config at a [`bounded_rate`] near 48 kHz (backend resamples + remaps),
|
||
/// 4. the device's default config (only if nothing above is drivable).
|
||
///
|
||
/// Cases 3–4 incur resampling; the backend reads the returned config's rate and
|
||
/// channel count and converts at the boundary (W4). Case 3 (review B5) is what keeps
|
||
/// an oddball endpoint whose default rate is outside the drivable window — but which
|
||
/// also exposes a usable in-window config — from being rejected by [`resolve`]. A
|
||
/// device that exposes no config at all is still a hard error.
|
||
fn choose_config(device: &Device, output: bool) -> Result<cpal::SupportedStreamConfig, AudioError> {
|
||
let ranges: Vec<cpal::SupportedStreamConfigRange> = if output {
|
||
device
|
||
.supported_output_configs()
|
||
.map_err(|e| AudioError::Device(e.to_string()))?
|
||
.collect()
|
||
} else {
|
||
device
|
||
.supported_input_configs()
|
||
.map_err(|e| AudioError::Device(e.to_string()))?
|
||
.collect()
|
||
};
|
||
|
||
// A range covers a sample-rate span and a fixed channel count.
|
||
let supports_48k = |r: &cpal::SupportedStreamConfigRange| {
|
||
r.min_sample_rate().0 <= SAMPLE_RATE && SAMPLE_RATE <= r.max_sample_rate().0
|
||
};
|
||
let pick = |channels: Option<u16>| {
|
||
ranges
|
||
.iter()
|
||
.find(|r| {
|
||
usable_range(r) && supports_48k(r) && channels.is_none_or(|c| r.channels() == c)
|
||
})
|
||
.cloned()
|
||
};
|
||
|
||
// Cases 1 + 2: an exact-48 kHz config, preferring the native layout but
|
||
// accepting any channel count (the backend remaps without resampling).
|
||
let exact = if output {
|
||
pick(Some(PLAYBACK_CHANNELS as u16)).or_else(|| pick(None))
|
||
} else {
|
||
pick(Some(1)).or_else(|| pick(None))
|
||
};
|
||
if let Some(r) = exact {
|
||
return Ok(r.with_sample_rate(SampleRate(SAMPLE_RATE)));
|
||
}
|
||
|
||
// Case 3: no native 48 kHz. Before falling back to the device default — which
|
||
// resolve() rejects outright if its rate is outside the drivable window — look
|
||
// for a supported config whose rate range overlaps that window and drive it at a
|
||
// bounded rate, resampling at the boundary (review B5). Prefer the native layout,
|
||
// then the bounded rate closest to 48 kHz.
|
||
let pick_bounded = |channels: Option<u16>| -> Option<(cpal::SupportedStreamConfigRange, u32)> {
|
||
ranges
|
||
.iter()
|
||
.filter(|r| usable_range(r) && channels.is_none_or(|c| r.channels() == c))
|
||
.filter_map(|r| {
|
||
bounded_rate(r.min_sample_rate().0, r.max_sample_rate().0)
|
||
.map(|rate| (r.clone(), rate))
|
||
})
|
||
.min_by_key(|(_, rate)| rate.abs_diff(SAMPLE_RATE))
|
||
};
|
||
let preferred_channels = if output { PLAYBACK_CHANNELS as u16 } else { 1 };
|
||
if let Some((r, rate)) = pick_bounded(Some(preferred_channels)).or_else(|| pick_bounded(None)) {
|
||
crate::log_msg(&format!(
|
||
"cpal: device '{}' has no native {SAMPLE_RATE} Hz {} config; using bounded {rate} Hz / {} ch with linear resampling (W4/B5)",
|
||
device.name().unwrap_or_else(|_| "<unknown>".to_string()),
|
||
if output { "output" } else { "input" },
|
||
r.channels(),
|
||
));
|
||
return Ok(r.with_sample_rate(SampleRate(rate)));
|
||
}
|
||
|
||
// Case 4: last resort — the device's default config. If its rate is outside the
|
||
// drivable window, resolve() rejects it with a clear device error, which is the
|
||
// honest outcome: the device exposes nothing this backend can drive.
|
||
let def = if output {
|
||
device.default_output_config()
|
||
} else {
|
||
device.default_input_config()
|
||
}
|
||
.map_err(|e| AudioError::Device(e.to_string()))?;
|
||
crate::log_msg(&format!(
|
||
"cpal: device '{}' has no bounded {} config near {SAMPLE_RATE} Hz; falling back to default {} Hz / {} ch (W4)",
|
||
device.name().unwrap_or_else(|_| "<unknown>".to_string()),
|
||
if output { "output" } else { "input" },
|
||
def.sample_rate().0,
|
||
def.channels(),
|
||
));
|
||
Ok(def)
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Capture
|
||
// ---------------------------------------------------------------------------
|
||
|
||
fn run_capture(
|
||
tx: Sender<Vec<i16>>,
|
||
target: Option<String>,
|
||
running: Arc<AtomicBool>,
|
||
ready: Sender<Result<(), AudioError>>,
|
||
) {
|
||
// 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::<i16>::new(CAPTURE_RING_CAPACITY);
|
||
let (producer, mut consumer) = rb.split();
|
||
let overrun = Arc::new(AtomicU64::new(0));
|
||
// Stream-liveness signals read by `wait_for_stream_start`: each RT data callback
|
||
// bumps `callbacks`; the RT error callback sets `err_code` (it does NOT log —
|
||
// that would allocate/syscall on the time-critical thread). See W1/W2/B1.
|
||
let callbacks = Arc::new(AtomicUsize::new(0));
|
||
let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE));
|
||
|
||
// Fallible device/stream setup: resolve, build, and *queue* the stream start.
|
||
let setup = || -> Result<(Stream, String, SampleFormat, usize, u32), AudioError> {
|
||
let (device, config, sample_format) = resolve(false, target)?;
|
||
let channels = config.channels as usize;
|
||
let device_rate = config.sample_rate.0;
|
||
let stream = match sample_format {
|
||
SampleFormat::F32 => build_input::<f32, _>(
|
||
&device,
|
||
&config,
|
||
producer,
|
||
channels,
|
||
overrun.clone(),
|
||
callbacks.clone(),
|
||
err_code.clone(),
|
||
),
|
||
SampleFormat::I16 => build_input::<i16, _>(
|
||
&device,
|
||
&config,
|
||
producer,
|
||
channels,
|
||
overrun.clone(),
|
||
callbacks.clone(),
|
||
err_code.clone(),
|
||
),
|
||
SampleFormat::U16 => build_input::<u16, _>(
|
||
&device,
|
||
&config,
|
||
producer,
|
||
channels,
|
||
overrun.clone(),
|
||
callbacks.clone(),
|
||
err_code.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(|_| "<unknown>".to_string());
|
||
Ok((stream, name, sample_format, channels, device_rate))
|
||
};
|
||
|
||
// Build + queue, then wait for the stream to actually prove it's live before
|
||
// reporting readiness. `play()` returning Ok only means WASAPI's `Start()` was
|
||
// queued; a later Start failure would otherwise leave us joined-but-silent (W1).
|
||
let (stream, dev_name, sample_format, channels, device_rate) = match setup() {
|
||
Ok(v) => match wait_for_stream_start(&callbacks, &err_code, &running) {
|
||
Ok(()) => {
|
||
let _ = ready.send(Ok(()));
|
||
v
|
||
}
|
||
Err(e) => {
|
||
// Drop the (possibly wedged) stream BEFORE reporting: cpal's
|
||
// Stream::drop joins its WASAPI worker, so if that wedges we want
|
||
// the Err withheld and finish_start's backstop to detach, rather
|
||
// than finish_start joining this owner forever (B2).
|
||
drop(v.0);
|
||
let _ = ready.send(Err(e));
|
||
return;
|
||
}
|
||
},
|
||
Err(e) => {
|
||
let _ = ready.send(Err(e));
|
||
return;
|
||
}
|
||
};
|
||
crate::log_msg(&format!(
|
||
"cpal capture started: device='{dev_name}' format={sample_format:?} channels={channels} device_rate={device_rate} Hz -> {SAMPLE_RATE} Hz"
|
||
));
|
||
|
||
// 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));
|
||
// 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<i16> = Vec::new();
|
||
|
||
// Drain the RT ring on this thread: pop mono samples, (resample,) 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;
|
||
let mut last_err = STREAM_ERR_NONE;
|
||
while running.load(Ordering::Relaxed) {
|
||
let mut drained = false;
|
||
while let Some(sample) = consumer.try_pop() {
|
||
drained = true;
|
||
resampled.clear();
|
||
match resampler {
|
||
Some(ref mut rs) => {
|
||
rs.push(i16_to_f32(sample), |out| resampled.push(f32_to_i16(out)));
|
||
}
|
||
None => resampled.push(sample),
|
||
}
|
||
for s in resampled.drain(..) {
|
||
if let Some(frame) = acc.push(s) {
|
||
// 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;
|
||
}
|
||
// Surface a stream error the RT callback flagged (it can't log itself).
|
||
let ec = err_code.load(Ordering::Relaxed);
|
||
if ec != STREAM_ERR_NONE && ec != last_err {
|
||
crate::log_msg(&format!(
|
||
"cpal capture stream error: {}",
|
||
stream_err_text(ec)
|
||
));
|
||
last_err = ec;
|
||
}
|
||
if !drained {
|
||
thread::sleep(CAPTURE_POLL);
|
||
}
|
||
}
|
||
drop(stream);
|
||
}
|
||
|
||
#[allow(clippy::too_many_arguments)]
|
||
fn build_input<T, P>(
|
||
device: &Device,
|
||
config: &StreamConfig,
|
||
mut producer: P,
|
||
channels: usize,
|
||
overrun: Arc<AtomicU64>,
|
||
callbacks: Arc<AtomicUsize>,
|
||
err_code: Arc<AtomicU8>,
|
||
) -> Result<Stream, AudioError>
|
||
where
|
||
T: SizedSample + Send + 'static,
|
||
i16: FromSample<T>,
|
||
P: Producer<Item = i16> + Send + 'static,
|
||
{
|
||
// RT-safe error callback: record a category in an atomic only. Formatting +
|
||
// logging happen on the owner thread (the cpal/WASAPI error callback runs on
|
||
// the time-critical stream thread, where alloc/syscall are forbidden — W2).
|
||
let err_fn = move |e: cpal::StreamError| err_code.store(stream_err_code(&e), Ordering::Relaxed);
|
||
device
|
||
.build_input_stream::<T, _, _>(
|
||
config,
|
||
move |data: &[T], _| {
|
||
// Count callbacks so the owner can confirm the stream is really
|
||
// running before reporting Ok (W1/B1).
|
||
callbacks.fetch_add(1, Ordering::Relaxed);
|
||
// 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 producer.try_push(mono).is_err() {
|
||
overrun.fetch_add(1, Ordering::Relaxed);
|
||
}
|
||
}
|
||
},
|
||
err_fn,
|
||
None,
|
||
)
|
||
.map_err(|e| AudioError::Stream(e.to_string()))
|
||
}
|
||
|
||
/// Average a device frame's channels down to a single mono i16. For a 1-channel
|
||
/// device this is just the converted sample.
|
||
fn downmix_to_mono<T>(frame: &[T]) -> i16
|
||
where
|
||
T: Copy,
|
||
i16: FromSample<T>,
|
||
{
|
||
if frame.is_empty() {
|
||
return 0;
|
||
}
|
||
let sum: i32 = frame.iter().map(|&s| i16::from_sample(s) as i32).sum();
|
||
(sum / frame.len() as i32) as i16
|
||
}
|
||
|
||
/// Scale an i16 PCM sample to f32 in roughly `[-1, 1]` for interpolation.
|
||
#[inline]
|
||
fn i16_to_f32(s: i16) -> f32 {
|
||
s as f32 / 32768.0
|
||
}
|
||
|
||
/// Convert an interpolated f32 sample back to i16, clamping to range.
|
||
#[inline]
|
||
fn f32_to_i16(x: f32) -> i16 {
|
||
(x * 32768.0).clamp(i16::MIN as f32, i16::MAX as f32) as i16
|
||
}
|
||
|
||
/// Accumulates mono samples into fixed-size [`CAPTURE_FRAME`] frames. Pulled out
|
||
/// of the RT callback so the framing is unit-testable.
|
||
struct FrameAccumulator {
|
||
buf: Vec<i16>,
|
||
frame_len: usize,
|
||
}
|
||
|
||
impl FrameAccumulator {
|
||
fn new(frame_len: usize) -> Self {
|
||
Self {
|
||
buf: Vec::with_capacity(frame_len),
|
||
frame_len,
|
||
}
|
||
}
|
||
|
||
/// Push one sample; returns a completed frame when the buffer fills.
|
||
fn push(&mut self, sample: i16) -> Option<Vec<i16>> {
|
||
self.buf.push(sample);
|
||
if self.buf.len() == self.frame_len {
|
||
Some(std::mem::replace(
|
||
&mut self.buf,
|
||
Vec::with_capacity(self.frame_len),
|
||
))
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Playback
|
||
// ---------------------------------------------------------------------------
|
||
|
||
fn run_playback(
|
||
rx: Receiver<Vec<i16>>,
|
||
target: Option<String>,
|
||
ring_fill: Arc<AtomicUsize>,
|
||
running: Arc<AtomicBool>,
|
||
ready: Sender<Result<(), AudioError>>,
|
||
) {
|
||
let rb = HeapRb::<i16>::new(RING_CAPACITY);
|
||
let (mut producer, consumer) = rb.split();
|
||
|
||
// Prefill to the steady-state depth so playout starts at target. `ring_fill`
|
||
// is an EXACT occupancy counter maintained by deltas (worker fetch_add on
|
||
// push, RT callback fetch_sub on pop) — not ringbuf's cached `occupied_len`,
|
||
// which is stale across the split halves and would lie high and starve the
|
||
// ring. See pipewire_impl.rs for the full rationale.
|
||
for _ in 0..PLAYBACK_TARGET_SAMPLES {
|
||
let _ = producer.try_push(0);
|
||
}
|
||
ring_fill.store(PLAYBACK_TARGET_SAMPLES, Ordering::Relaxed);
|
||
|
||
// Diagnostics (mirrors the PipeWire backend's playout-health line).
|
||
let underrun = Arc::new(AtomicU64::new(0));
|
||
let dropped = Arc::new(AtomicU64::new(0));
|
||
// Largest single output-callback length seen (interleaved samples). WASAPI
|
||
// shared-mode picks its own period, so this can exceed the prefill target —
|
||
// which would force an underrun every cycle (review W2). The callback only
|
||
// does a wait-free fetch_max; the health logger reports/warns off the RT path.
|
||
let max_cb = Arc::new(AtomicUsize::new(0));
|
||
// Stream-liveness signals (see the capture path / W1, W2, B1): each RT callback
|
||
// bumps `callbacks`; the RT error callback sets `err_code` without logging.
|
||
let callbacks = Arc::new(AtomicUsize::new(0));
|
||
let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE));
|
||
|
||
// Fallible device/stream setup. `consumer` is moved into the output callback.
|
||
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::<f32, _>(
|
||
&device,
|
||
&config,
|
||
consumer,
|
||
ring_fill.clone(),
|
||
underrun.clone(),
|
||
max_cb.clone(),
|
||
callbacks.clone(),
|
||
err_code.clone(),
|
||
),
|
||
SampleFormat::I16 => build_output::<i16, _>(
|
||
&device,
|
||
&config,
|
||
consumer,
|
||
ring_fill.clone(),
|
||
underrun.clone(),
|
||
max_cb.clone(),
|
||
callbacks.clone(),
|
||
err_code.clone(),
|
||
),
|
||
SampleFormat::U16 => build_output::<u16, _>(
|
||
&device,
|
||
&config,
|
||
consumer,
|
||
ring_fill.clone(),
|
||
underrun.clone(),
|
||
max_cb.clone(),
|
||
callbacks.clone(),
|
||
err_code.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(|_| "<unknown>".to_string());
|
||
Ok((stream, name, sample_format, channels, device_rate))
|
||
};
|
||
|
||
// Build + queue, then wait for real callbacks before reporting readiness (W1).
|
||
let (stream, dev_name, sample_format, channels, device_rate) = match setup() {
|
||
Ok(v) => match wait_for_stream_start(&callbacks, &err_code, &running) {
|
||
Ok(()) => {
|
||
let _ = ready.send(Ok(()));
|
||
v
|
||
}
|
||
Err(e) => {
|
||
// Drop before reporting so a wedged Stream::drop withholds the Err
|
||
// and lets finish_start's backstop detach instead of hanging (B2).
|
||
drop(v.0);
|
||
let _ = ready.send(Err(e));
|
||
return;
|
||
}
|
||
},
|
||
Err(e) => {
|
||
let _ = ready.send(Err(e));
|
||
return;
|
||
}
|
||
};
|
||
crate::log_msg(&format!(
|
||
"cpal playback started: device='{dev_name}' format={sample_format:?} channels={channels} device_rate={device_rate} Hz <- {SAMPLE_RATE} Hz"
|
||
));
|
||
|
||
let logger = spawn_health_logger(
|
||
running.clone(),
|
||
ring_fill.clone(),
|
||
underrun.clone(),
|
||
dropped.clone(),
|
||
max_cb.clone(),
|
||
err_code.clone(),
|
||
);
|
||
|
||
// Feed the ring from the network mixer until `stop()` flips `running` or the
|
||
// sender disconnects (call ended). Clock-paced production keeps the ring near
|
||
// target, so the drop path below should never fire in steady state.
|
||
drain_loop(&rx, &running, |frame| {
|
||
if ring_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY {
|
||
dropped.fetch_add(1, Ordering::Relaxed);
|
||
return;
|
||
}
|
||
// Reserve occupancy BEFORE publishing samples, and publish the whole frame
|
||
// in one `push_slice` (review W3). Per-sample pushes let the RT consumer
|
||
// observe a half-written stereo pair (L without R) → channel tear, and a
|
||
// pop that raced the post-loop `fetch_add` could drive `ring_fill` below
|
||
// zero and wrap it to usize::MAX, wedging the mixer's pacing. Reserving
|
||
// first means the consumer can never pop a sample that isn't yet counted.
|
||
ring_fill.fetch_add(frame.len(), Ordering::Relaxed);
|
||
let pushed = producer.push_slice(&frame);
|
||
if pushed != frame.len() {
|
||
// The capacity check above should make this unreachable (the consumer
|
||
// only drains), but stay exact if it ever isn't.
|
||
ring_fill.fetch_sub(frame.len() - pushed, Ordering::Relaxed);
|
||
dropped.fetch_add(1, Ordering::Relaxed);
|
||
}
|
||
});
|
||
|
||
// We're shutting down (either stop() or disconnect). Ensure the logger sees it
|
||
// even on the disconnect path, then drop the stream.
|
||
running.store(false, Ordering::Relaxed);
|
||
let _ = logger.join();
|
||
drop(stream);
|
||
}
|
||
|
||
#[allow(clippy::too_many_arguments)]
|
||
fn build_output<T, C>(
|
||
device: &Device,
|
||
config: &StreamConfig,
|
||
mut consumer: C,
|
||
ring_fill: Arc<AtomicUsize>,
|
||
underrun: Arc<AtomicU64>,
|
||
max_cb: Arc<AtomicUsize>,
|
||
callbacks: Arc<AtomicUsize>,
|
||
err_code: Arc<AtomicU8>,
|
||
) -> Result<Stream, AudioError>
|
||
where
|
||
T: SizedSample + FromSample<i16> + Send + 'static,
|
||
C: Consumer<Item = i16> + Send + 'static,
|
||
{
|
||
// RT-safe error callback: atomic store only, no alloc/log (review W2).
|
||
let err_fn = move |e: cpal::StreamError| err_code.store(stream_err_code(&e), Ordering::Relaxed);
|
||
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::<T, _, _>(
|
||
config,
|
||
move |data: &mut [T], _| {
|
||
callbacks.fetch_add(1, Ordering::Relaxed);
|
||
// Record demand in INTERNAL 48 kHz-stereo samples (not raw
|
||
// device samples) so the health logger's prefill-target
|
||
// comparison is apples-to-apples for any rate/layout (W4).
|
||
max_cb.fetch_max(
|
||
internal_demand(data.len(), device_channels, device_rate),
|
||
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::<T, _, _>(
|
||
config,
|
||
move |data: &mut [T], _| {
|
||
callbacks.fetch_add(1, Ordering::Relaxed);
|
||
max_cb.fetch_max(
|
||
internal_demand(data.len(), device_channels, device_rate),
|
||
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()))
|
||
}
|
||
}
|
||
|
||
/// Convert an output callback's raw device-sample length into the equivalent
|
||
/// internal 48 kHz-stereo sample demand, so the prefill-target comparison stays
|
||
/// meaningful regardless of the device's rate/channel layout (W4 diagnostic fix).
|
||
/// Integer-only and allocation-free, so it is safe on the RT callback thread.
|
||
#[inline]
|
||
fn internal_demand(device_len: usize, device_channels: usize, device_rate: u32) -> usize {
|
||
let device_frames = device_len / device_channels.max(1);
|
||
let need_frames =
|
||
(device_frames as u64 * SAMPLE_RATE as u64).div_ceil(device_rate.max(1) as u64) as usize;
|
||
need_frames * PLAYBACK_CHANNELS
|
||
}
|
||
|
||
/// Drain the ring into the device buffer, substituting silence on underrun.
|
||
/// Returns `(samples_popped, samples_starved)`. RT-safe (wait-free `try_pop`).
|
||
fn fill_output<T, C>(consumer: &mut C, out: &mut [T]) -> (usize, u64)
|
||
where
|
||
T: Sample + FromSample<i16>,
|
||
C: Consumer<Item = i16>,
|
||
{
|
||
let mut popped = 0usize;
|
||
let mut starved = 0u64;
|
||
for slot in out.iter_mut() {
|
||
match consumer.try_pop() {
|
||
Some(v) => {
|
||
*slot = T::from_sample(v);
|
||
popped += 1;
|
||
}
|
||
None => {
|
||
*slot = T::from_sample(0i16);
|
||
starved += 1;
|
||
}
|
||
}
|
||
}
|
||
(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<T, C>(
|
||
consumer: &mut C,
|
||
out: &mut [T],
|
||
device_channels: usize,
|
||
resampler: &mut StereoPullResampler,
|
||
) -> (usize, u64)
|
||
where
|
||
T: Sample + FromSample<i16>,
|
||
C: Consumer<Item = i16>,
|
||
{
|
||
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(
|
||
running: Arc<AtomicBool>,
|
||
ring_fill: Arc<AtomicUsize>,
|
||
underrun: Arc<AtomicU64>,
|
||
dropped: Arc<AtomicU64>,
|
||
max_cb: Arc<AtomicUsize>,
|
||
err_code: Arc<AtomicU8>,
|
||
) -> JoinHandle<()> {
|
||
let verbose = std::env::var_os("PEERSPEAK_AUDIO_VERBOSE").is_some();
|
||
thread::spawn(move || {
|
||
let (mut last_u, mut last_d) = (0u64, 0u64);
|
||
let mut reported_cb = 0usize;
|
||
let mut last_err = STREAM_ERR_NONE;
|
||
while running.load(Ordering::Relaxed) {
|
||
thread::sleep(Duration::from_secs(1));
|
||
let u = underrun.load(Ordering::Relaxed);
|
||
let d = dropped.load(Ordering::Relaxed);
|
||
let fill = ring_fill.load(Ordering::Relaxed);
|
||
let (du, dd) = (u - last_u, d - last_d);
|
||
last_u = u;
|
||
last_d = d;
|
||
if verbose || du > 0 || dd > 0 {
|
||
crate::log_msg(&format!(
|
||
"playout-health: fill={fill} samples (~{}ms) | underrun +{du} samples/s (total {u}) | dropped +{dd} frames/s (total {d})",
|
||
fill / (48 * PLAYBACK_CHANNELS),
|
||
));
|
||
}
|
||
// Surface a stream error the RT callback flagged (it can't log itself).
|
||
let ec = err_code.load(Ordering::Relaxed);
|
||
if ec != STREAM_ERR_NONE && ec != last_err {
|
||
crate::log_msg(&format!(
|
||
"cpal playback stream error: {}",
|
||
stream_err_text(ec)
|
||
));
|
||
last_err = ec;
|
||
}
|
||
// Report the device's per-cycle demand (in internal 48 kHz-stereo
|
||
// samples) the first time it's seen, and on any new high. If a callback
|
||
// demands more than the prefill target, the ring can't satisfy it and
|
||
// underruns every cycle — the W2 signature; warn so a real-host log
|
||
// shows whether it's biting.
|
||
let cb = max_cb.load(Ordering::Relaxed);
|
||
if cb > reported_cb {
|
||
reported_cb = cb;
|
||
let ms = cb / (48 * PLAYBACK_CHANNELS);
|
||
if cb > PLAYBACK_TARGET_SAMPLES {
|
||
crate::log_msg(&format!(
|
||
"cpal output callback demands up to {cb} internal samples/cycle (~{ms}ms) EXCEEDS prefill target {PLAYBACK_TARGET_SAMPLES} — expect periodic underruns; needs a larger target or a fixed buffer size (review W2)",
|
||
));
|
||
} else if verbose {
|
||
crate::log_msg(&format!(
|
||
"cpal output callback demands up to {cb} internal samples/cycle (~{ms}ms), target {PLAYBACK_TARGET_SAMPLES}",
|
||
));
|
||
}
|
||
}
|
||
}
|
||
})
|
||
}
|
||
|
||
/// Pump frames from `rx` to `on_frame` until `running` goes false or the sender
|
||
/// disconnects. The timed receive re-checks `running` at least every
|
||
/// [`WORKER_POLL`], so `stop()` can join the worker promptly instead of hanging
|
||
/// on a parked blocking `recv()` (same A7 fix as the PipeWire backend). Pure
|
||
/// w.r.t. its inputs, so it's unit-testable.
|
||
fn drain_loop(rx: &Receiver<Vec<i16>>, running: &AtomicBool, mut on_frame: impl FnMut(Vec<i16>)) {
|
||
while running.load(Ordering::Relaxed) {
|
||
match rx.recv_timeout(WORKER_POLL) {
|
||
Ok(frame) => on_frame(frame),
|
||
Err(RecvTimeoutError::Timeout) => continue,
|
||
Err(RecvTimeoutError::Disconnected) => return,
|
||
}
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use std::sync::mpsc;
|
||
|
||
#[test]
|
||
fn downmix_averages_channels() {
|
||
assert_eq!(downmix_to_mono::<i16>(&[100, 100]), 100);
|
||
assert_eq!(downmix_to_mono::<i16>(&[100, -100]), 0);
|
||
assert_eq!(downmix_to_mono::<i16>(&[50]), 50);
|
||
assert_eq!(downmix_to_mono::<i16>(&[]), 0);
|
||
// 4-channel average rounds toward zero (integer division).
|
||
assert_eq!(downmix_to_mono::<i16>(&[10, 20, 30, 41]), 25);
|
||
}
|
||
|
||
#[test]
|
||
fn frame_accumulator_emits_full_frames() {
|
||
let mut acc = FrameAccumulator::new(3);
|
||
assert_eq!(acc.push(1), None);
|
||
assert_eq!(acc.push(2), None);
|
||
assert_eq!(acc.push(3), Some(vec![1, 2, 3]));
|
||
// Resets for the next frame.
|
||
assert_eq!(acc.push(4), None);
|
||
assert_eq!(acc.push(5), None);
|
||
assert_eq!(acc.push(6), Some(vec![4, 5, 6]));
|
||
}
|
||
|
||
#[test]
|
||
fn fill_output_pops_then_substitutes_silence() {
|
||
let rb = HeapRb::<i16>::new(8);
|
||
let (mut prod, mut cons) = rb.split();
|
||
for v in [1, 2, 3] {
|
||
prod.try_push(v).unwrap();
|
||
}
|
||
let mut out = [0i16; 5];
|
||
let (popped, starved) = fill_output(&mut cons, &mut out);
|
||
assert_eq!(popped, 3);
|
||
assert_eq!(starved, 2);
|
||
assert_eq!(out, [1, 2, 3, 0, 0]);
|
||
}
|
||
|
||
#[test]
|
||
fn fill_output_remap_downmixes_to_mono() {
|
||
let rb = HeapRb::<i16>::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::<i16>::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::<i16>::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::<Vec<i16>>();
|
||
let running = Arc::new(AtomicBool::new(true));
|
||
let r2 = running.clone();
|
||
let h = thread::spawn(move || drain_loop(&rx, &r2, |_| {}));
|
||
thread::sleep(Duration::from_millis(50));
|
||
running.store(false, Ordering::Relaxed);
|
||
thread::sleep(WORKER_POLL + Duration::from_millis(150));
|
||
assert!(
|
||
h.is_finished(),
|
||
"drain_loop must exit after running=false even while the sender is alive"
|
||
);
|
||
drop(tx);
|
||
h.join().unwrap();
|
||
}
|
||
|
||
#[test]
|
||
fn drain_loop_returns_on_disconnect() {
|
||
let (tx, rx) = mpsc::channel::<Vec<i16>>();
|
||
let running = Arc::new(AtomicBool::new(true));
|
||
drop(tx);
|
||
drain_loop(&rx, &running, |_| panic!("no frame should arrive"));
|
||
}
|
||
|
||
#[test]
|
||
fn bounded_rate_prefers_48k_when_in_window() {
|
||
// A device span that contains 48 kHz resolves exactly.
|
||
assert_eq!(bounded_rate(44_100, 96_000), Some(SAMPLE_RATE));
|
||
assert_eq!(
|
||
bounded_rate(MIN_DEVICE_RATE, MAX_DEVICE_RATE),
|
||
Some(SAMPLE_RATE)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn bounded_rate_clamps_to_nearest_in_window_bound() {
|
||
// Entirely below 48 kHz → the top bound (closest reachable to 48 kHz).
|
||
assert_eq!(bounded_rate(8_000, 16_000), Some(16_000));
|
||
// Entirely above 48 kHz → the bottom bound.
|
||
assert_eq!(bounded_rate(88_200, 192_000), Some(88_200));
|
||
}
|
||
|
||
#[test]
|
||
fn bounded_rate_rejects_spans_outside_the_window() {
|
||
assert_eq!(bounded_rate(1_000, 4_000), None); // below the floor
|
||
assert_eq!(bounded_rate(400_000, 500_000), None); // above the ceiling
|
||
}
|
||
|
||
#[test]
|
||
fn bounded_rate_intersects_window_edges() {
|
||
// Overlaps only the floor: [4k, 8k] ∩ [8k, 384k] = {8k}.
|
||
assert_eq!(bounded_rate(4_000, MIN_DEVICE_RATE), Some(MIN_DEVICE_RATE));
|
||
// Overlaps only the ceiling.
|
||
assert_eq!(
|
||
bounded_rate(MAX_DEVICE_RATE, 500_000),
|
||
Some(MAX_DEVICE_RATE)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn format_supported_matches_the_stream_builders() {
|
||
// Exactly the three the build_input/build_output match arms accept.
|
||
for f in [SampleFormat::F32, SampleFormat::I16, SampleFormat::U16] {
|
||
assert!(format_supported(f), "{f:?} should be drivable");
|
||
}
|
||
// Everything else cpal can expose must be filtered out before ranking, or a
|
||
// start could pick it and then hit the `unsupported sample format` arm (P3).
|
||
for f in [
|
||
SampleFormat::I8,
|
||
SampleFormat::U8,
|
||
SampleFormat::I32,
|
||
SampleFormat::U32,
|
||
SampleFormat::I64,
|
||
SampleFormat::U64,
|
||
SampleFormat::F64,
|
||
] {
|
||
assert!(!format_supported(f), "{f:?} must not be reported drivable");
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn ensure_idle_allows_an_idle_slot() {
|
||
let mut s = SlotState::Idle;
|
||
assert!(ensure_idle(&mut s, "capture").is_ok());
|
||
assert!(matches!(s, SlotState::Idle));
|
||
}
|
||
|
||
#[test]
|
||
fn ensure_idle_rejects_a_live_wedged_orphan_then_clears_when_it_exits() {
|
||
let exited = Arc::new(AtomicBool::new(false));
|
||
let mut s = SlotState::Wedged {
|
||
exited: exited.clone(),
|
||
};
|
||
// Orphan still alive → reject, tombstone preserved.
|
||
assert!(ensure_idle(&mut s, "playback").is_err());
|
||
assert!(matches!(s, SlotState::Wedged { .. }));
|
||
// Orphan's ExitGuard fired → the next start clears the tombstone and proceeds.
|
||
exited.store(true, Ordering::Relaxed);
|
||
assert!(ensure_idle(&mut s, "playback").is_ok());
|
||
assert!(matches!(s, SlotState::Idle));
|
||
}
|
||
|
||
#[test]
|
||
fn drain_loop_delivers_frames() {
|
||
let (tx, rx) = mpsc::channel::<Vec<i16>>();
|
||
let running = Arc::new(AtomicBool::new(true));
|
||
let r2 = running.clone();
|
||
let got = Arc::new(Mutex::new(Vec::new()));
|
||
let g2 = got.clone();
|
||
let h = thread::spawn(move || drain_loop(&rx, &r2, |f| g2.lock().unwrap().push(f)));
|
||
tx.send(vec![1, 2, 3]).unwrap();
|
||
tx.send(vec![4, 5]).unwrap();
|
||
thread::sleep(Duration::from_millis(50));
|
||
running.store(false, Ordering::Relaxed);
|
||
drop(tx);
|
||
h.join().unwrap();
|
||
assert_eq!(*got.lock().unwrap(), vec![vec![1, 2, 3], vec![4, 5]]);
|
||
}
|
||
}
|