diff --git a/src/audio/pipewire_impl.rs b/src/audio/pipewire_impl.rs index 40a7cbd..5c4a2fc 100644 --- a/src/audio/pipewire_impl.rs +++ b/src/audio/pipewire_impl.rs @@ -1,5 +1,5 @@ use crate::audio::{AudioBackend, AudioError}; -use std::sync::mpsc::{Sender, Receiver}; +use std::sync::mpsc::{Sender, Receiver, RecvTimeoutError}; use std::sync::{Arc, Mutex}; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::thread::{self, JoinHandle}; @@ -238,6 +238,31 @@ fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender>, target_n /// never the whole slice, because over-pulling past the ring depth is exactly /// the sustained crackle this replaces. Under-filling a cycle is a harmless /// brief glitch (PipeWire pads). +// How often the playback worker wakes to re-check its `running` flag while idle, +// bounding how long `stop` can block joining the worker (bug A7). A plain +// blocking receive would never re-check the flag, waking only on a new frame or +// the sender dropping, so shutdown would hang if the sender outlives `stop`. +const WORKER_POLL: Duration = Duration::from_millis(100); + +/// Pump frames from `rx` to `on_frame` until `running` goes false or the sender +/// disconnects. Uses a timed receive so the loop re-checks `running` at least +/// every `WORKER_POLL` even when no frames arrive — this is what lets `stop()` +/// join the worker promptly instead of hanging on a parked blocking `recv()` +/// (bug A7). Pure w.r.t. its inputs (no PipeWire), so it's unit-testable. +fn drain_loop( + rx: &Receiver>, + running: &AtomicBool, + mut on_frame: impl FnMut(Vec), +) { + while running.load(Ordering::Relaxed) { + match rx.recv_timeout(WORKER_POLL) { + Ok(frame) => on_frame(frame), + Err(RecvTimeoutError::Timeout) => continue, // re-check `running` + Err(RecvTimeoutError::Disconnected) => return, // sender gone for good + } + } +} + fn frames_to_produce(requested: usize, mapped_frames: usize) -> usize { /// Safe per-cycle fallback when the graph doesn't report a quantum. const FALLBACK_FRAMES: usize = 1024; @@ -483,24 +508,20 @@ fn run_playback( let worker_dropped = dropped_frames.clone(); let worker_fill = fill_gauge.clone(); let worker_handle = thread::spawn(move || { - while running_clone.load(Ordering::Relaxed) { - if let Ok(frame) = rx.recv() { - // Drop the whole frame (rather than tearing it) only if it truly - // won't fit — checked against the exact occupancy counter, not - // the ring's stale length observer. With clock-paced production - // this should never fire. - if worker_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY { - worker_dropped.fetch_add(1, Ordering::Relaxed); - continue; - } - for &sample in &frame { - let _ = producer.try_push(sample); - } - worker_fill.fetch_add(frame.len(), Ordering::Relaxed); - } else { + drain_loop(&rx, &running_clone, |frame| { + // Drop the whole frame (rather than tearing it) only if it truly + // won't fit — checked against the exact occupancy counter, not + // the ring's stale length observer. With clock-paced production + // this should never fire. + if worker_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY { + worker_dropped.fetch_add(1, Ordering::Relaxed); return; } - } + for &sample in &frame { + let _ = producer.try_push(sample); + } + worker_fill.fetch_add(frame.len(), Ordering::Relaxed); + }); }); // Diagnostic logger: once per second, report the ring fill and the delta in @@ -551,7 +572,11 @@ fn run_playback( #[cfg(test)] mod tests { - use super::frames_to_produce; + use super::{drain_loop, frames_to_produce}; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + use std::{sync::mpsc, thread}; #[test] fn requested_in_range_is_honored() { @@ -583,4 +608,56 @@ mod tests { assert_eq!(frames_to_produce(0, 0), 0); assert_eq!(frames_to_produce(1024, 0), 0); } + + // --- drain_loop (A7: worker must not hang shutdown) --- + + #[test] + fn drain_loop_exits_when_running_flips_even_with_sender_alive() { + // The exact A7 hang scenario: the frame sender is STILL alive (never + // dropped) when `running` goes false. A blocking `recv()` would park + // forever; `drain_loop` must wake within a poll interval and return. + let (tx, rx) = mpsc::channel::>(); + let running = Arc::new(AtomicBool::new(true)); + let r2 = running.clone(); + let h = thread::spawn(move || drain_loop(&rx, &r2, |_| {})); + // Let it park in recv_timeout, then signal stop. + thread::sleep(Duration::from_millis(50)); + running.store(false, Ordering::Relaxed); + // Wait comfortably longer than one poll interval; it must have exited. + thread::sleep(super::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); // keep tx alive until here so the test really covers the case + h.join().unwrap(); + } + + #[test] + fn drain_loop_returns_on_disconnect() { + // Sender dropped before we start → recv errors Disconnected → return. + // If this hangs, the test runner times out (i.e. it would fail loudly). + let (tx, rx) = mpsc::channel::>(); + let running = Arc::new(AtomicBool::new(true)); + drop(tx); + drain_loop(&rx, &running, |_| panic!("no frame should arrive")); + } + + #[test] + fn drain_loop_delivers_frames_to_callback() { + let (tx, rx) = mpsc::channel::>(); + 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(); + let got = got.lock().unwrap(); + assert_eq!(*got, vec![vec![1, 2, 3], vec![4, 5]]); + } }