diff --git a/src/audio/cpal_impl.rs b/src/audio/cpal_impl.rs index 31676f7..81e200e 100644 --- a/src/audio/cpal_impl.rs +++ b/src/audio/cpal_impl.rs @@ -88,16 +88,23 @@ 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). +/// 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); -/// 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); +/// 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 @@ -132,24 +139,36 @@ fn stream_err_text(code: u8) -> &'static str { } } -/// 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). +/// 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( - started: &AtomicBool, + 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 { - 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())); + 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())); @@ -487,10 +506,10 @@ fn run_capture( let rb = HeapRb::::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)); + // 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. @@ -500,15 +519,15 @@ fn run_capture( let device_rate = config.sample_rate.0; let stream = match sample_format { SampleFormat::F32 => build_input::( - &device, &config, producer, channels, overrun.clone(), started.clone(), + &device, &config, producer, channels, overrun.clone(), callbacks.clone(), err_code.clone(), ), SampleFormat::I16 => build_input::( - &device, &config, producer, channels, overrun.clone(), started.clone(), + &device, &config, producer, channels, overrun.clone(), callbacks.clone(), err_code.clone(), ), SampleFormat::U16 => build_input::( - &device, &config, producer, channels, overrun.clone(), started.clone(), + &device, &config, producer, channels, overrun.clone(), callbacks.clone(), err_code.clone(), ), other => Err(AudioError::Stream(format!( @@ -526,14 +545,18 @@ fn run_capture( // 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(&started, &err_code, &running) { + 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)); - drop(v.0); // the stream return; } }, @@ -611,7 +634,7 @@ fn build_input( mut producer: P, channels: usize, overrun: Arc, - started: Arc, + callbacks: Arc, err_code: Arc, ) -> Result where @@ -627,8 +650,9 @@ where .build_input_stream::( config, move |data: &[T], _| { - // First callback proves WASAPI actually started the stream (W1). - started.store(true, Ordering::Relaxed); + // 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) { @@ -731,9 +755,9 @@ 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)); + // 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. @@ -744,15 +768,15 @@ fn run_playback( let stream = match sample_format { SampleFormat::F32 => build_output::( &device, &config, consumer, ring_fill.clone(), underrun.clone(), - max_cb.clone(), started.clone(), err_code.clone(), + max_cb.clone(), callbacks.clone(), err_code.clone(), ), SampleFormat::I16 => build_output::( &device, &config, consumer, ring_fill.clone(), underrun.clone(), - max_cb.clone(), started.clone(), err_code.clone(), + max_cb.clone(), callbacks.clone(), err_code.clone(), ), SampleFormat::U16 => build_output::( &device, &config, consumer, ring_fill.clone(), underrun.clone(), - max_cb.clone(), started.clone(), err_code.clone(), + max_cb.clone(), callbacks.clone(), err_code.clone(), ), other => Err(AudioError::Stream(format!( "unsupported playback sample format: {other:?}" @@ -765,16 +789,18 @@ fn run_playback( Ok((stream, name, sample_format, channels, device_rate)) }; - // Build + queue, then wait for a real callback before reporting readiness (W1). + // 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(&started, &err_code, &running) { + 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)); - drop(v.0); // the stream return; } }, @@ -835,7 +861,7 @@ fn build_output( ring_fill: Arc, underrun: Arc, max_cb: Arc, - started: Arc, + callbacks: Arc, err_code: Arc, ) -> Result where @@ -851,7 +877,7 @@ where .build_output_stream::( config, move |data: &mut [T], _| { - started.store(true, Ordering::Relaxed); + 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). @@ -880,7 +906,7 @@ where .build_output_stream::( config, move |data: &mut [T], _| { - started.store(true, Ordering::Relaxed); + callbacks.fetch_add(1, Ordering::Relaxed); max_cb.fetch_max( internal_demand(data.len(), device_channels, device_rate), Ordering::Relaxed,