Skip to content

fix(pulseaudio): avoid mutex blocking the callback #1293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
roderickvd wants to merge 2 commits into
base: master
Choose a base branch
from
+53 −42
Open

fix(pulseaudio): avoid mutex blocking the callback #1293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5ef6637
fix(pulseaudio): avoid mutex blocking the callback
roderickvd Jul 29, 2026
d2ae432
refactor(pulseaudio): make latency calculation RT-safe
roderickvd Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- **PipeWire**: Build on 32-bit targets without native 64-bit atomics.
- **PulseAudio**: `NoData` errors are no longer misreported as buffer xruns.
- **PulseAudio**: Build on 32-bit targets without native 64-bit atomics.
- **PulseAudio**: Latency calculation in the audio callback is now RT-safe.
- **visionOS**: The CoreAudio backend now builds.
- **WASAPI**: Default device changes no longer report `DeviceChanged`, which wrongly implied the stream had rerouted automatically.
- **WASAPI**: Reported buffer sizes are no longer off by one frame.
Expand Down
94 changes: 52 additions & 42 deletions src/host/pulseaudio/stream.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
use std::{
sync::{
atomic::{AtomicBool, Ordering},
Arc, Condvar, Mutex,
Arc, Mutex, OnceLock,
},
thread::Thread,
time::{Duration, Instant},
};

Expand All @@ -28,29 +29,42 @@ const LATENCY_MAX_INTERVAL: Duration = Duration::from_millis(100);
struct LatencyHandle {
// Cancellation on drop
cancel: Arc<AtomicBool>,
// Event-driven early wakeup from callbacks and play/pause
update: Arc<(Mutex<bool>, Condvar)>,
// Whether the stream is currently uncorked; the latency thread polls on
// the buffer period while playing, and parks indefinitely otherwise.
playing: Arc<AtomicBool>,
// Set once the latency thread is spawned, so play/pause/cancel can wake it.
thread: OnceLock<Thread>,
}

impl LatencyHandle {
fn new() -> Self {
Self {
cancel: Arc::new(AtomicBool::new(false)),
update: Arc::new((Mutex::new(false), Condvar::new())),
playing: Arc::new(AtomicBool::new(false)),
thread: OnceLock::new(),
}
}

// Trigger an early poll
fn notify(&self) {
let (lock, cvar) = &*self.update;
*lock.lock().unwrap_or_else(|e| e.into_inner()) = true;
cvar.notify_one();
fn set_thread(&self, thread: Thread) {
let _ = self.thread.set(thread);
}

fn wake(&self) {
if let Some(thread) = self.thread.get() {
thread.unpark();
}
}

// Update play state and wake the thread so it switches wait modes.
fn set_playing(&self, playing: bool) {
self.playing.store(playing, Ordering::Relaxed);
self.wake();
}

// Signal cancellation and wake the thread immediately
fn cancel(&self) {
self.cancel.store(true, Ordering::Relaxed);
self.notify();
self.wake();
}
}

Expand Down Expand Up @@ -100,12 +114,12 @@ impl StreamTrait for Stream {
match &self.inner {
StreamInner::Playback(stream, _, handle) => {
block_on(stream.uncork()).map_err(Error::from)?;
handle.notify();
handle.set_playing(true);
}
StreamInner::Record(stream, _, handle) => {
block_on(stream.uncork()).map_err(Error::from)?;
block_on(stream.started()).map_err(Error::from)?;
handle.notify();
handle.set_playing(true);
}
}
Ok(())
Expand All @@ -119,7 +133,7 @@ impl StreamTrait for Stream {
res.map_err(Error::from)?;
match &self.inner {
StreamInner::Playback(_, _, handle) | StreamInner::Record(_, _, handle) => {
handle.notify()
handle.set_playing(false)
}
}
Ok(())
Expand Down Expand Up @@ -169,6 +183,9 @@ impl Stream {
let poll_clone = last_poll_micros.clone();
let sample_spec = params.sample_spec;
let pa_format = sample_spec.format;
// The latency thread polls at this cadence while playing, instead of on every callback.
let poll_interval =
sample_spec.bytes_to_duration(params.buffer_attr.minimum_request_length as usize);

let format: SampleFormat = pa_format.try_into().map_err(|_| {
Error::with_message(
Expand All @@ -188,7 +205,6 @@ impl Stream {
};

let handle = LatencyHandle::new();
let update_callback = handle.update.clone();

// Wrap the write callback to match the pulseaudio signature.
let callback = move |buf: &mut [u8]| {
Expand Down Expand Up @@ -230,11 +246,6 @@ impl Stream {

data_callback(&mut data, &OutputCallbackInfo { timestamp });

// Notify the latency thread that audio was written, so it updates timing info.
let (lock, cvar) = &*update_callback;
*lock.lock().unwrap_or_else(|e| e.into_inner()) = true;
cvar.notify_one();

// We always consider the full buffer filled, because cpal's
// user-facing API doesn't allow short writes.
buf.len()
Expand Down Expand Up @@ -271,7 +282,7 @@ impl Stream {
});

let cancel_thread = handle.cancel.clone();
let update_thread = handle.update.clone();
let playing_thread = handle.playing.clone();
let stream_clone = stream.clone();
let latency_clone = current_latency_micros.clone();
let poll_clone = last_poll_micros.clone();
Expand Down Expand Up @@ -304,15 +315,16 @@ impl Stream {
timing_info.read_offset,
);

// Wait until woken by a write/play/pause/drop event or until LATENCY_MAX_INTERVAL.
let (lock, cvar) = &*update_thread;
let Ok(guard) = lock.lock() else { break };
let (mut guard, _) = cvar
.wait_timeout_while(guard, LATENCY_MAX_INTERVAL, |notified| !*notified)
.unwrap_or_else(|e| e.into_inner());
*guard = false;
// While playing, poll on the buffer period. Otherwise park indefinitely
// until play/pause/drop wakes this thread.
if playing_thread.load(Ordering::Relaxed) {
std::thread::park_timeout(poll_interval);
} else {
std::thread::park();
}
}
});
handle.set_thread(latency_handle.thread().clone());

latch.add_thread(driver_handle.thread().clone());
latch.add_thread(latency_handle.thread().clone());
Expand Down Expand Up @@ -343,6 +355,9 @@ impl Stream {
let poll_clone = last_poll_micros.clone();
let sample_spec = params.sample_spec;
let pa_format = sample_spec.format;
// The latency thread polls at this cadence while recording, instead of on every callback.
let poll_interval =
sample_spec.bytes_to_duration(params.buffer_attr.fragment_size as usize);

let format: SampleFormat = pa_format.try_into().map_err(|_| {
Error::with_message(
Expand All @@ -352,7 +367,6 @@ impl Stream {
})?;

let handle = LatencyHandle::new();
let update_callback = handle.update.clone();

let callback = move |buf: &[u8]| {
let elapsed = Instant::now().saturating_duration_since(start);
Expand Down Expand Up @@ -391,19 +405,14 @@ impl Stream {
let data = unsafe { Data::from_parts(buf.as_ptr() as *mut _, n_samples, format) };

data_callback(&data, &InputCallbackInfo { timestamp });

// Notify the latency thread that audio was read, so it updates timing info.
let (lock, cvar) = &*update_callback;
*lock.lock().unwrap_or_else(|e| e.into_inner()) = true;
cvar.notify_one();
};

let stream =
block_on(client.create_record_stream(params, callback)).map_err(Error::from)?;

// Spawn a thread to monitor the stream's latency in a loop.
let cancel_thread = handle.cancel.clone();
let update_thread = handle.update.clone();
let playing_thread = handle.playing.clone();
let stream_clone = stream.clone();
let latency_clone = current_latency_micros.clone();
let poll_clone = last_poll_micros.clone();
Expand Down Expand Up @@ -440,17 +449,18 @@ impl Stream {
timing_info.read_offset,
);

// Wait until woken by a read/play/pause/drop event or until LATENCY_MAX_INTERVAL.
let (lock, cvar) = &*update_thread;
let Ok(guard) = lock.lock() else { break };
let (mut guard, _) = cvar
.wait_timeout_while(guard, LATENCY_MAX_INTERVAL, |notified| !*notified)
.unwrap_or_else(|e| e.into_inner());
*guard = false;
// While recording, poll on the buffer period. Otherwise park indefinitely
// until play/pause/drop wakes this thread.
if playing_thread.load(Ordering::Relaxed) {
std::thread::park_timeout(poll_interval);
} else {
std::thread::park();
}
}
});

handle.set_thread(latency_handle.thread().clone());
latch.add_thread(latency_handle.thread().clone());

Ok(Self {
inner: StreamInner::Record(stream, start, handle),
workers: vec![latency_handle],
Expand Down
Loading