mollusk/peerspeak
Public Access
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases 10 Wiki Activity

audio(win): fix RT-safety + start-handshake bugs in the cpal backend

Browse Source 
Addresses Codex's xhigh RT-audio audit of the new Windows cpal path (review
2026-06-19; all Windows-only, no Linux-path change):

- W1 (P1): start_capture/start_playback reported Ok as soon as cpal's play()
  returned, but cpal's WASAPI play() only QUEUES IAudioClient::Start(); a later
  Start failure left the UI joined-but-silent. Readiness is now driven by the
  stream actually proving itself: the first RT data callback sets a started
  flag (or the error callback sets an error code), and the owner thread waits
  (bounded by STREAM_START_TIMEOUT) before reporting Ok.
- W2: both RT error callbacks ran format!+log_msg on the time-critical stream
  thread. They now store a category in an AtomicU8 only; the owner / health
  logger translate + log off the RT path.
- W3: the playback ring was published one interleaved sample at a time, letting
  the RT consumer read a half-written L/R pair and letting a raced fetch_sub
  wrap ring_fill to usize::MAX (wedging mixer pacing). Now reserves occupancy
  before publishing and writes the whole frame with a single push_slice.
- W6: finish_start did an unbounded recv() while holding the slot mutex, so a
  wedged driver hung start_* and any concurrent stop. Now recv_timeout with a
  FINISH_START_TIMEOUT backstop; on timeout it signals + detaches (never joins).
- W7: OS-reported device geometry is validated in resolve() (channels>0, rate in
  8k-384k) so 0 channels can't panic chunks_exact(0) and a 0/absurd rate can't
  make an infinite/huge resample ratio. resample.rs constructors also clamp
  rates >=1 (release-safe; +2 tests) instead of a debug-only assert.
- W4 (diagnostic half): the playout-health logger compared raw device samples
  against the internal-stereo prefill target. The callback now records demand in
  internal 48 kHz-stereo units (internal_demand) so the comparison is correct
  for remapped/non-48k devices. The dynamic-target restructure stays deferred.

Deferred (logged in review-2026-06-19-cpal-rt-audit.md): W5 (bounded mixer->
worker channel) touches the shared Linux audio path and wants its own design +
regression pass; the W2 dynamic-target sizing needs a real WASAPI callback.

Verified: Linux cargo test --lib 326/0, clippy --all-targets clean; windows-gnu
cargo check --lib --tests --bins clean; windows-gnu release peerspeak.exe builds.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-06-19 16:15:49 -04:00
co-authored by Claude Opus 4.8
parent 4d07e03395
commit f52b5ea64e
2 changed files with 279 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/audio/cpal_impl.rs
+240 -61
View File
@@ -51,11 +51,11 @@
//! 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, AtomicU64, AtomicUsize, Ordering};
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;
use std::time::{Duration, Instant};
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{
@@ -88,6 +88,80 @@ 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 deliver its first RT callback (proof
/// WASAPI actually `Start()`ed it) 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);
/// Backstop for [`finish_start`]: covers the whole owner setup (resolve + build +
/// play + first-callback wait). Larger than [`STREAM_START_TIMEOUT`] so the owner's
/// own timeout normally fires first with a precise error; this only trips if the
/// owner itself wedges in a driver call before it can report readiness (review W6).
const FINISH_START_TIMEOUT: Duration = Duration::from_secs(6);
/// 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 first RT
/// callback sets `started`, or an error callback sets `err_code`. Returns `Ok` on
/// the first callback, `Err` on an error-callback code or [`STREAM_START_TIMEOUT`],
/// or a clean abort if `stop()` cleared `running` mid-start. Polls a couple of
/// cheap atomics on the owner thread — never the RT thread (review W1).
fn wait_for_stream_start(
started: &AtomicBool,
err_code: &AtomicU8,
running: &AtomicBool,
) -> Result<(), AudioError> {
let deadline = Instant::now() + STREAM_START_TIMEOUT;
loop {
if started.load(Ordering::Relaxed) {
return Ok(());
}
let code = err_code.load(Ordering::Relaxed);
if code != STREAM_ERR_NONE {
return Err(AudioError::Stream(stream_err_text(code).to_string()));
}
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 {
@@ -186,25 +260,39 @@ fn finish_start(
ready_rx: Receiver<Result<(), AudioError>>,
what: &str,
) -> Result<(), AudioError> {
match ready_rx.recv() {
// 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 = 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.
// 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(_) => {
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. The thread unwinds on its own if/when the
// driver call ever returns.
worker.running.store(false, Ordering::Relaxed);
drop(worker.thread);
Err(AudioError::Init(format!(
"cpal {what} did not start within {FINISH_START_TIMEOUT:?}"
)))
}
}
}
@@ -294,6 +382,22 @@ fn resolve(
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))
}
@@ -383,23 +487,30 @@ fn run_capture(
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`: the first RT data
// callback sets `started`; the RT error callback sets `err_code` (it does NOT
// log — that would allocate/syscall on the time-critical thread). See W1/W2.
let started = Arc::new(AtomicBool::new(false));
let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE));
// 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.
// 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())
}
SampleFormat::I16 => {
build_input::<i16, _>(&device, &config, producer, channels, overrun.clone())
}
SampleFormat::U16 => {
build_input::<u16, _>(&device, &config, producer, channels, overrun.clone())
}
SampleFormat::F32 => build_input::<f32, _>(
&device, &config, producer, channels, overrun.clone(), started.clone(),
err_code.clone(),
),
SampleFormat::I16 => build_input::<i16, _>(
&device, &config, producer, channels, overrun.clone(), started.clone(),
err_code.clone(),
),
SampleFormat::U16 => build_input::<u16, _>(
&device, &config, producer, channels, overrun.clone(), started.clone(),
err_code.clone(),
),
other => Err(AudioError::Stream(format!(
"unsupported capture sample format: {other:?}"
))),
@@ -411,11 +522,21 @@ fn run_capture(
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) => {
let _ = ready.send(Ok(()));
v
}
Ok(v) => match wait_for_stream_start(&started, &err_code, &running) {
Ok(()) => {
let _ = ready.send(Ok(()));
v
}
Err(e) => {
let _ = ready.send(Err(e));
drop(v.0); // the stream
return;
}
},
Err(e) => {
let _ = ready.send(Err(e));
return;
@@ -439,6 +560,7 @@ fn run_capture(
// 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() {
@@ -469,6 +591,12 @@ fn run_capture(
));
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);
}
@@ -476,23 +604,31 @@ fn run_capture(
drop(stream);
}
#[allow(clippy::too_many_arguments)]
fn build_input<T, P>(
device: &Device,
config: &StreamConfig,
mut producer: P,
channels: usize,
overrun: Arc<AtomicU64>,
started: Arc<AtomicBool>,
err_code: Arc<AtomicU8>,
) -> Result<Stream, AudioError>
where
T: SizedSample + Send + 'static,
i16: FromSample<T>,
P: Producer<Item = i16> + Send + 'static,
{
let err_fn = |e| crate::log_msg(&format!("cpal capture stream error: {e}"));
// 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], _| {
// First callback proves WASAPI actually started the stream (W1).
started.store(true, 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) {
@@ -595,38 +731,28 @@ fn run_playback(
// 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): first RT callback
// sets `started`; the RT error callback sets `err_code` without logging.
let started = Arc::new(AtomicBool::new(false));
let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE));
// 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.
// 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(),
&device, &config, consumer, ring_fill.clone(), underrun.clone(),
max_cb.clone(), started.clone(), err_code.clone(),
),
SampleFormat::I16 => build_output::<i16, _>(
&device,
&config,
consumer,
ring_fill.clone(),
underrun.clone(),
max_cb.clone(),
&device, &config, consumer, ring_fill.clone(), underrun.clone(),
max_cb.clone(), started.clone(), err_code.clone(),
),
SampleFormat::U16 => build_output::<u16, _>(
&device,
&config,
consumer,
ring_fill.clone(),
underrun.clone(),
max_cb.clone(),
&device, &config, consumer, ring_fill.clone(), underrun.clone(),
max_cb.clone(), started.clone(), err_code.clone(),
),
other => Err(AudioError::Stream(format!(
"unsupported playback sample format: {other:?}"
@@ -639,11 +765,19 @@ fn run_playback(
Ok((stream, name, sample_format, channels, device_rate))
};
// Build + queue, then wait for a real callback before reporting readiness (W1).
let (stream, dev_name, sample_format, channels, device_rate) = match setup() {
Ok(v) => {
let _ = ready.send(Ok(()));
v
}
Ok(v) => match wait_for_stream_start(&started, &err_code, &running) {
Ok(()) => {
let _ = ready.send(Ok(()));
v
}
Err(e) => {
let _ = ready.send(Err(e));
drop(v.0); // the stream
return;
}
},
Err(e) => {
let _ = ready.send(Err(e));
return;
@@ -659,6 +793,7 @@ fn run_playback(
underrun.clone(),
dropped.clone(),
max_cb.clone(),
err_code.clone(),
);
// Feed the ring from the network mixer until `stop()` flips `running` or the
@@ -669,10 +804,20 @@ fn run_playback(
dropped.fetch_add(1, Ordering::Relaxed);
return;
}
for &sample in &frame {
let _ = producer.try_push(sample);
}
// 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
@@ -682,6 +827,7 @@ fn run_playback(
drop(stream);
}
#[allow(clippy::too_many_arguments)]
fn build_output<T, C>(
device: &Device,
config: &StreamConfig,
@@ -689,12 +835,15 @@ fn build_output<T, C>(
ring_fill: Arc<AtomicUsize>,
underrun: Arc<AtomicU64>,
max_cb: Arc<AtomicUsize>,
started: Arc<AtomicBool>,
err_code: Arc<AtomicU8>,
) -> Result<Stream, AudioError>
where
T: SizedSample + FromSample<i16> + Send + 'static,
C: Consumer<Item = i16> + Send + 'static,
{
let err_fn = |e| crate::log_msg(&format!("cpal playback stream error: {e}"));
// 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 {
@@ -702,8 +851,14 @@ where
.build_output_stream::<T, _, _>(
config,
move |data: &mut [T], _| {
// Wait-free; the logger thread reads this off the RT path.
max_cb.fetch_max(data.len(), Ordering::Relaxed);
started.store(true, 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);
@@ -725,8 +880,11 @@ where
.build_output_stream::<T, _, _>(
config,
move |data: &mut [T], _| {
// Wait-free; the logger thread reads this off the RT path.
max_cb.fetch_max(data.len(), Ordering::Relaxed);
started.store(true, 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 {
@@ -746,6 +904,18 @@ where
}
}
/// 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)
@@ -832,11 +1002,13 @@ fn spawn_health_logger(
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);
@@ -851,21 +1023,28 @@ fn spawn_health_logger(
fill / (48 * PLAYBACK_CHANNELS),
));
}
// Report the device's callback size the first time it's seen (and on
// any new high). If a callback asks for more than the prefill target,
// the ring can't satisfy it and underruns every cycle — the W2 bug
// signature; warn so a real-host log shows whether it's biting.
// 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 up to {cb} samples/cycle (~{ms}ms) EXCEEDS prefill target {PLAYBACK_TARGET_SAMPLES} — expect periodic underruns; needs a larger target or a fixed buffer size (review W2)",
"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 up to {cb} samples/cycle (~{ms}ms), target {PLAYBACK_TARGET_SAMPLES}",
"cpal output callback demands up to {cb} internal samples/cycle (~{ms}ms), target {PLAYBACK_TARGET_SAMPLES}",
));
}
}
src/audio/resample.rs
+39 -6
View File
@@ -45,11 +45,14 @@ pub struct PushResampler {
}
impl PushResampler {
/// Build a resampler from `in_rate` to `out_rate` (both in Hz, must be > 0).
/// Build a resampler from `in_rate` to `out_rate` (both in Hz). Rates are
/// clamped to `>= 1` so `step` is always finite and non-zero: a zero `step`
/// would make [`push`](Self::push)'s `while self.next < 1.0` loop forever. The
/// cpal backend's `resolve()` also rejects such rates up front, so this is
/// belt-and-suspenders against a future caller (review W7).
pub fn new(in_rate: u32, out_rate: u32) -> Self {
debug_assert!(in_rate > 0 && out_rate > 0);
Self {
step: in_rate as f64 / out_rate as f64,
step: in_rate.max(1) as f64 / out_rate.max(1) as f64,
next: 0.0,
prev: 0.0,
started: false,
@@ -108,11 +111,12 @@ pub struct StereoPullResampler {
}
impl StereoPullResampler {
/// Build a resampler from `in_rate` to `out_rate` (both in Hz, must be > 0).
/// Build a resampler from `in_rate` to `out_rate` (both in Hz). Rates are
/// clamped to `>= 1` so `step` is finite and non-zero — otherwise
/// [`next`](Self::next)'s `while self.frac >= 1.0` could spin (review W7).
pub fn new(in_rate: u32, out_rate: u32) -> Self {
debug_assert!(in_rate > 0 && out_rate > 0);
Self {
step: in_rate as f64 / out_rate as f64,
step: in_rate.max(1) as f64 / out_rate.max(1) as f64,
frac: 0.0,
prev: (0.0, 0.0),
cur: (0.0, 0.0),
@@ -271,4 +275,33 @@ mod tests {
// At step 2.0 we consume ~2 input frames per output frame.
assert!(idx > emitted, "consumed {idx} input, emitted {emitted} output");
}
/// A zero rate must not produce a zero `step` (which would spin `push`'s inner
/// `while self.next < 1.0` forever). Clamping makes the call terminate (W7).
#[test]
fn push_zero_rate_does_not_spin() {
let mut r = PushResampler::new(0, 48_000);
let mut count = 0usize;
// Feed two samples; with a clamped non-zero step this returns promptly.
r.push(0.0, |_| count += 1);
r.push(1.0, |_| count += 1);
// Reaching here at all is the assertion (no hang); some output is produced.
assert!(count >= 1);
}
/// A zero output rate must not make the pull resampler's segment-advance loop
/// spin. Clamping keeps `step` finite so `next` terminates (W7).
#[test]
fn pull_zero_out_rate_does_not_spin() {
let mut r = StereoPullResampler::new(48_000, 0);
let frames = [(0.0, 0.0), (1.0, 1.0), (2.0, 2.0)];
let mut idx = 0;
let got = r.next(|| {
let v = frames.get(idx).copied();
idx += 1;
v
});
// Terminates and yields the primed frame instead of hanging.
assert!(got.is_some());
}
}