Skip to content

fix(pipewire): stream timeout, disconnect errors and fallback socket #1371

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

Merged
roderickvd merged 6 commits into from
Sep 19, 2026
Merged

fix(pipewire): stream timeout, disconnect errors and fallback socket #1371

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e75c492
fix(pipewire): honor the build timeout
roderickvd Sep 16, 2026
a6d5ae4
fix(pipewire): don't report a disconnect when a stream is dropped
roderickvd Sep 16, 2026
7df00e9
fix(pipewire): report a disconnect when a device is removed
roderickvd Sep 17, 2026
a2415e9
fix(pipewire): detect host via /run/pipewire fallback socket
roderickvd Sep 17, 2026
8360343
fix(ci): drop dbus/jack/pipewire from mipsel install
roderickvd Sep 17, 2026
0cae9c0
fix(pipewire): demote rt-promoted thread before teardown to avoid sig…
roderickvd Sep 17, 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
9 changes: 5 additions & 4 deletions .github/workflows/platforms.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,8 +199,9 @@ jobs:
- name: Install mipsel cross-toolchain and audio libs
run: |
dpkg --add-architecture mipsel
apt-get update -qq
apt-get install -y --no-install-recommends libclang-dev crossbuild-essential-mipsel $(for p in ${{ env.PACKAGES_LINUX }}; do echo "$p:mipsel"; done)
apt-get update
# ALSA only: dbus/jack/pipewire need libsystemd0:mipsel, which apt 2.6's resolver can't add without breaking Essential packages
apt-get install -y --no-install-recommends libclang-dev crossbuild-essential-mipsel libasound2-dev:mipsel

- name: Install Rust nightly (for build-std)
uses: dtolnay/rust-toolchain@nightly
Expand All @@ -215,8 +216,8 @@ jobs:
- name: Check examples (default features)
run: cargo +nightly check --examples --workspace --verbose --target ${{ env.TARGET }} -Z build-std

- name: Check examples (all features)
run: cargo +nightly check --examples --all-features --workspace --verbose --target ${{ env.TARGET }} -Z build-std
- name: Check examples (realtime, pulseaudio, custom)
run: cargo +nightly check --examples --features realtime,pulseaudio,custom --workspace --verbose --target ${{ env.TARGET }} -Z build-std

# Windows (x86_64 and i686)
windows:
Expand Down
6 changes: 6 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- **JACK**: Streams no longer panic when the server delivers a larger period than the negotiated buffer size.
- **PipeWire**: Fix an empty chunk being emitted when a cycle requests no frames.
- **PipeWire**: Fix capture reading from the wrong offset in the buffer on some devices.
- **PipeWire**: Building a stream without a timeout now waits indefinitely instead of failing after two seconds.
- **PipeWire**: A worker thread that stops before the stream is ready is now reported as a backend error instead of a timeout.
- **PipeWire**: Dropping a stream for a specific device no longer reports it as disconnected.
- **PipeWire**: Removing a device now reports `DeviceNotAvailable` on streams using it.
- **PipeWire**: The host is now detected when `PIPEWIRE_RUNTIME_DIR`/`XDG_RUNTIME_DIR` are unset but the socket exists at `/run/pipewire`.
- **PipeWire**: Dropping a stream no longer risks crashing the process with `SIGXCPU` if it was real-time promoted.
- **WASAPI**: Device enumeration no longer panics if the COM enumerator fails to initialize.
- **WASAPI**: Revert "Default device changes no longer report `DeviceChanged`" as it was misinformed.
- **WASAPI**: Output streams now start with real audio immediately instead of undefined content in the render buffer.
Expand Down
146 changes: 90 additions & 56 deletions src/host/pipewire/device.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@ use crate::{
latch::Latch,
pipewire::{
stream::{
DefaultDeviceMonitor, PwInitGuard, SUPPORTED_FORMATS, StreamCommand, StreamData,
PwInitGuard, SUPPORTED_FORMATS, StreamCommand, StreamData, watch_default_device,
watch_target_node,
},
utils::{DEVICE_ICON_NAME, METADATA_NAME, audio, clock, default},
},
Expand All @@ -47,8 +48,6 @@ use crate::{

pub type Devices = std::vec::IntoIter<Device>;

const INIT_TIMEOUT: Duration = Duration::from_secs(2);

// This enum record whether it is created by human or just default device
#[derive(Clone, Debug, Default, Copy)]
pub(crate) enum Class {
Expand Down Expand Up @@ -373,7 +372,6 @@ impl DeviceTrait for Device {
let mut latch = Latch::new();
let waiter = latch.waiter();
let device = self.clone();
let wait_timeout = timeout.unwrap_or(Duration::from_secs(2));
let initial_quantum = match config.buffer_size {
BufferSize::Fixed(n) => n,
BufferSize::Default => self.quantum,
Expand Down Expand Up @@ -432,25 +430,31 @@ impl DeviceTrait for Device {
invalidated,
} = stream_data;

let default_monitor = if let Some(key) = device.default_metadata_key() {
match core.get_registry_rc() {
Ok(registry) => Some(DefaultDeviceMonitor::new(
registry,
key,
error_callback.clone(),
invalidated,
pending_device_changed,
)),
Err(e) => {
let _ = init_tx.send(Err(Error::with_message(
ErrorKind::BackendError,
format!("Could not acquire registry: {e}"),
)));
return;
}
let invalidated_cmd = invalidated.clone();
let registry = match core.get_registry_rc() {
Ok(registry) => registry,
Err(e) => {
let _ = init_tx.send(Err(Error::with_message(
ErrorKind::BackendError,
format!("Could not acquire registry: {e}"),
)));
return;
}
} else {
None
};
let device_listener = match device.default_metadata_key() {
Some(key) => watch_default_device(
&registry,
key,
error_callback.clone(),
invalidated,
pending_device_changed,
),
None => watch_target_node(
&registry,
device.object_serial,
error_callback.clone(),
invalidated,
),
};
let stream_clone = stream.clone();
let mainloop_rc1 = mainloop.clone();
Expand All @@ -469,6 +473,8 @@ impl DeviceTrait for Device {
}
StreamCommand::Drain => {}
StreamCommand::Stop => {
// Flag an intentional stop to prevent the stream reporting a disconnected event.
invalidated_cmd.store(true, Ordering::Relaxed);
if let Err(e) = stream_clone.disconnect() {
emit_error(
&error_callback_cmd,
Expand All @@ -494,7 +500,8 @@ impl DeviceTrait for Device {
mainloop.run();

drop(listener);
drop(default_monitor);
drop(device_listener);
drop(registry);
drop(core_monitor);
drop(core);
drop(context);
Expand All @@ -506,12 +513,7 @@ impl DeviceTrait for Device {
)
})?;

let init_result = init_rx.recv_timeout(wait_timeout).unwrap_or_else(|_| {
Err(Error::with_message(
ErrorKind::DeviceNotAvailable,
"PipeWire timed out",
))
});
let init_result = recv_init(&init_rx, timeout);

if let Err(e) = init_result {
drop(latch);
Expand Down Expand Up @@ -565,7 +567,6 @@ impl DeviceTrait for Device {
let mut latch = Latch::new();
let waiter = latch.waiter();
let device = self.clone();
let wait_timeout = timeout.unwrap_or(Duration::from_secs(2));
let initial_quantum = match config.buffer_size {
BufferSize::Fixed(n) => n,
BufferSize::Default => self.quantum,
Expand Down Expand Up @@ -627,25 +628,31 @@ impl DeviceTrait for Device {
invalidated,
} = stream_data;

let default_monitor = if let Some(key) = device.default_metadata_key() {
match core.get_registry_rc() {
Ok(registry) => Some(DefaultDeviceMonitor::new(
registry,
key,
error_callback.clone(),
invalidated,
pending_device_changed,
)),
Err(e) => {
let _ = init_tx.send(Err(Error::with_message(
ErrorKind::BackendError,
format!("Could not acquire registry: {e}"),
)));
return;
}
let invalidated_cmd = invalidated.clone();
let registry = match core.get_registry_rc() {
Ok(registry) => registry,
Err(e) => {
let _ = init_tx.send(Err(Error::with_message(
ErrorKind::BackendError,
format!("Could not acquire registry: {e}"),
)));
return;
}
} else {
None
};
let device_listener = match device.default_metadata_key() {
Some(key) => watch_default_device(
&registry,
key,
error_callback.clone(),
invalidated,
pending_device_changed,
),
None => watch_target_node(
&registry,
device.object_serial,
error_callback.clone(),
invalidated,
),
};
let stream_clone = stream.clone();
let mainloop_rc1 = mainloop.clone();
Expand Down Expand Up @@ -677,6 +684,8 @@ impl DeviceTrait for Device {
}
}
StreamCommand::Stop => {
// Flag an intentional stop to prevent the stream reporting a disconnected event.
invalidated_cmd.store(true, Ordering::Relaxed);
if let Err(e) = stream_clone.disconnect() {
emit_error(
&error_callback_cmd,
Expand All @@ -701,7 +710,8 @@ impl DeviceTrait for Device {

mainloop.run();
drop(listener);
drop(default_monitor);
drop(device_listener);
drop(registry);
drop(core_monitor);
drop(core);
drop(context);
Expand All @@ -713,12 +723,7 @@ impl DeviceTrait for Device {
)
})?;

let init_result = init_rx.recv_timeout(wait_timeout).unwrap_or_else(|_| {
Err(Error::with_message(
ErrorKind::DeviceNotAvailable,
"PipeWire timed out",
))
});
let init_result = recv_init(&init_rx, timeout);

if let Err(e) = init_result {
drop(latch);
Expand All @@ -741,6 +746,32 @@ impl DeviceTrait for Device {
}
}

/// Waits for the worker thread to report whether the stream came up.
///
/// Honors `build_*_stream`'s timeout contract: `None` waits indefinitely.
fn recv_init(
init_rx: &mpsc::Receiver<Result<(), Error>>,
timeout: Option<Duration>,
) -> Result<(), Error> {
let received = match timeout {
None => init_rx
.recv()
.map_err(|_| mpsc::RecvTimeoutError::Disconnected),
Some(timeout) => init_rx.recv_timeout(timeout),
};
received.unwrap_or_else(|err| {
Err(match err {
mpsc::RecvTimeoutError::Timeout => {
Error::with_message(ErrorKind::DeviceNotAvailable, "PipeWire timed out")
}
mpsc::RecvTimeoutError::Disconnected => Error::with_message(
ErrorKind::BackendError,
"PipeWire worker stopped before reporting the stream state",
),
})
})
}

#[derive(Clone, Default)]
struct Settings {
rate: SampleRate,
Expand Down Expand Up @@ -790,6 +821,9 @@ fn remote_props() -> Option<PropertiesBox> {
Some(props)
}

// Enumeration does not take a configurable timeout.
const ENUMERATION_TIMEOUT: Duration = Duration::from_secs(2);

pub fn init_devices(connect_automatically: Arc<AtomicBool>) -> Option<Vec<Device>> {
let _pw = PwInitGuard::new();
let mainloop = MainLoopRc::new(None).ok()?;
Expand Down Expand Up @@ -1077,7 +1111,7 @@ pub fn init_devices(connect_automatically: Arc<AtomicBool>) -> Option<Vec<Device
loop_quit.quit();
});
thread::spawn(move || {
if cancel_rx.recv_timeout(INIT_TIMEOUT).is_err() {
if cancel_rx.recv_timeout(ENUMERATION_TIMEOUT).is_err() {
let _ = timeout_tx.send(());
}
});
Expand Down
14 changes: 12 additions & 2 deletions src/host/pipewire/rt_promote.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,18 +41,28 @@ impl RtPromoter {
let join = thread::Builder::new()
.name("cpal_rt_promote".to_owned())
.spawn(move || {
let mut promoted = false;
loop {
if shutdown_bg.load(Ordering::Relaxed) {
// Demote before pw::deinit() to prevent spurious SIGXCPU signals.
if promoted {
if let Err(e) =
audio_thread_priority::demote_thread_from_real_time(thread_info)
{
emit_error(&error_callback_bg, Error::from(e));
}
}
return;
}
let frames = pending_frames_bg.swap(0, Ordering::Relaxed);
if frames != 0 {
if let Err(e) = audio_thread_priority::promote_thread_to_real_time(
match audio_thread_priority::promote_thread_to_real_time(
thread_info,
frames,
rate,
) {
emit_error(&error_callback_bg, Error::from(e));
Ok(_) => promoted = true,
Err(e) => emit_error(&error_callback_bg, Error::from(e)),
}
continue;
}
Expand Down
Loading
Loading