pipewire: handle commands in the main loop using pw channel

Attempt to fix #105

v2: don't use a `Mutex`, do some Arc/Weak/Drop/channel/unsave tricks instead.

Author: strohel

src/backends/pipewire/stream.rs

@@ -3,6 +3,7 @@
 use crate::audio_buffer::{AudioMut, AudioRef};
 use crate::backends::pipewire::error::PipewireError;
 use crate::channel_map::Bitset;
+use crate::prelude::pipewire::utils::{BlackHole, CallbackHolder};
 use crate::timestamp::Timestamp;
 use crate::{
     AudioCallbackContext, AudioInput, AudioInputCallback, AudioOutput, AudioOutputCallback,
@@ -15,12 +16,14 @@ use libspa::utils::Direction;
 use libspa_sys::{SPA_PARAM_EnumFormat, SPA_TYPE_OBJECT_Format};
 use pipewire::context::Context;
 use pipewire::keys;
-use pipewire::main_loop::{MainLoop, WeakMainLoop};
+use pipewire::main_loop::MainLoop;
 use pipewire::properties::Properties;
 use pipewire::stream::{Stream, StreamFlags};
+use std::cell::Cell;
 use std::collections::HashMap;
 use std::fmt;
 use std::fmt::Formatter;
+use std::sync::{Arc, Weak};
 use std::thread::JoinHandle;
 
 enum StreamCommands<Callback> {
@@ -36,38 +39,10 @@ impl<Callback> fmt::Debug for StreamCommands<Callback> {
 }
 
 struct StreamInner<Callback> {
-    commands: rtrb::Consumer<StreamCommands<Callback>>,
     scratch_buffer: Box<[f32]>,
-    callback: Option<Callback>,
+    callback: Weak<CallbackHolder<Callback>>,
     config: StreamConfig,
     timestamp: Timestamp,
-    loop_ref: WeakMainLoop,
-}
-
-impl<Callback> StreamInner<Callback> {
-    fn handle_command(&mut self, command: StreamCommands<Callback>) {
-        log::debug!("Handling command: {command:?}");
-        match command {
-            StreamCommands::Eject(reply) => {
-                if let Some(callback) = self.callback.take() {
-                    reply.send(callback).unwrap();
-                    if let Some(loop_ref) = self.loop_ref.upgrade() {
-                        loop_ref.quit();
-                    }
-                }
-            }
-        }
-    }
-
-    fn handle_commands(&mut self) {
-        while let Ok(command) = self.commands.pop() {
-            self.handle_command(command);
-        }
-    }
-
-    fn ejected(&self) -> bool {
-        self.callback.is_none()
-    }
 }
 
 impl<Callback: AudioOutputCallback> StreamInner<Callback> {
@@ -77,7 +52,7 @@ impl<Callback: AudioOutputCallback> StreamInner<Callback> {
             channels,
         )
         .unwrap();
-        if let Some(callback) = self.callback.as_mut() {
+        if let Some(mut callback) = self.callback.upgrade() {
             let context = AudioCallbackContext {
                 stream_config: self.config,
                 timestamp: self.timestamp,
@@ -87,7 +62,12 @@ impl<Callback: AudioOutputCallback> StreamInner<Callback> {
                 buffer,
                 timestamp: self.timestamp,
             };
+
+            // SAFETY: there is max one other owner of the callback Arc, and it never dereferences
+            // it thanks to `BlackHole`, fulfilling safety requirements of `arc_get_mut_unchecked()`.
+            let callback = unsafe { arc_get_mut_unchecked(&mut callback) };
             callback.on_output_data(context, output);
+
             self.timestamp += num_frames as u64;
             num_frames
         } else {
@@ -101,7 +81,7 @@ impl<Callback: AudioInputCallback> StreamInner<Callback> {
         let buffer =
             AudioRef::from_interleaved(&self.scratch_buffer[..channels * frames], channels)
                 .unwrap();
-        if let Some(callback) = self.callback.as_mut() {
+        if let Some(mut callback) = self.callback.upgrade() {
             let context = AudioCallbackContext {
                 stream_config: self.config,
                 timestamp: self.timestamp,
@@ -111,7 +91,12 @@ impl<Callback: AudioInputCallback> StreamInner<Callback> {
                 buffer,
                 timestamp: self.timestamp,
             };
+
+            // SAFETY: there is max one other owner of the callback Arc, and it never dereferences
+            // it thanks to `BlackHole`, fulfilling safety requirements of `arc_get_mut_unchecked()`.
+            let callback = unsafe { arc_get_mut_unchecked(&mut callback) };
             callback.on_input_data(context, input);
+
             self.timestamp += num_frames as u64;
             num_frames
         } else {
@@ -122,19 +107,19 @@ impl<Callback: AudioInputCallback> StreamInner<Callback> {
 
 /// PipeWire stream handle.
 pub struct StreamHandle<Callback> {
-    commands: rtrb::Producer<StreamCommands<Callback>>,
+    commands: pipewire::channel::Sender<StreamCommands<Callback>>,
     handle: JoinHandle<Result<(), PipewireError>>,
 }
 
 impl<Callback> AudioStreamHandle<Callback> for StreamHandle<Callback> {
     type Error = PipewireError;
 
-    fn eject(mut self) -> Result<Callback, Self::Error> {
+    fn eject(self) -> Result<Callback, Self::Error> {
         log::info!("Ejecting stream");
         let (tx, rx) = oneshot::channel();
         self.commands
-            .push(StreamCommands::Eject(tx))
-            .expect("Command buffer overflow");
+            .send(StreamCommands::Eject(tx))
+            .expect("Should be able to send a message through PipeWire channel");
         self.handle.join().unwrap()?;
         Ok(rx.recv().unwrap())
     }
@@ -152,7 +137,10 @@ impl<Callback: 'static + Send> StreamHandle<Callback> {
             + Send
             + 'static,
     ) -> Result<Self, PipewireError> {
-        let (tx, rx) = rtrb::RingBuffer::new(16);
+        // Create a channel for sending command into PipeWire main loop.
+        let (pipewire_sender, pipewire_receiver) =
+            pipewire::channel::channel::<StreamCommands<Callback>>();
+
         let handle = std::thread::spawn(move || {
             let main_loop = MainLoop::new(None)?;
             let context = Context::new(&main_loop)?;
@@ -177,23 +165,27 @@ impl<Callback: 'static + Send> StreamHandle<Callback> {
                 properties.insert(*keys::TARGET_OBJECT, device_object_serial);
             }
 
+            let (callback_holder, callback_rx) = CallbackHolder::new(callback);
+            let callback_holder = Arc::new(callback_holder);
+
+            let stream_inner = StreamInner {
+                callback: Arc::downgrade(&callback_holder),
+                scratch_buffer: vec![0.0; MAX_FRAMES * channels].into_boxed_slice(),
+                config,
+                timestamp: Timestamp::new(config.samplerate),
+            };
+
+            // SAFETY of StreamInner::process_input(), StreamInner::process_output() depends on us
+            // never _dereferencing_ `callback_holder` outside of `StreamInner`. Achieve that at
+            // type level by wrapping it in a black hole.
+            let callback_holder = BlackHole::new(callback_holder);
+
             let stream = Stream::new(&core, &name, properties)?;
             config.samplerate = config.samplerate.round();
             let _listener = stream
-                .add_local_listener_with_user_data(StreamInner {
-                    callback: Some(callback),
-                    commands: rx,
-                    scratch_buffer: vec![0.0; MAX_FRAMES * channels].into_boxed_slice(),
-                    loop_ref: main_loop.downgrade(),
-                    config,
-                    timestamp: Timestamp::new(config.samplerate),
-                })
+                .add_local_listener_with_user_data(stream_inner)
                 .process(move |stream, inner| {
                     log::debug!("Processing stream");
-                    inner.handle_commands();
-                    if inner.ejected() {
-                        return;
-                    }
                     if let Some(mut buffer) = stream.dequeue_buffer() {
                         let datas = buffer.datas_mut();
                         log::debug!("Datas: len={}", datas.len());
@@ -232,12 +224,48 @@ impl<Callback: 'static + Send> StreamHandle<Callback> {
                 StreamFlags::AUTOCONNECT | StreamFlags::MAP_BUFFERS | StreamFlags::RT_PROCESS,
                 &mut params,
             )?;
+
+            // Handle commands (stream ejection). Runs in the PipeWire main loop.
+            let loop_ref = main_loop.downgrade();
+            // pipewire::channel::receiver::attach() only accepts `Fn()` (instead of expected
+            // `FnMut()`), so we need interior mutability. Cell is sufficient.
+            let callback_holder = Cell::new(Some(callback_holder));
+            let _attached_receiver = pipewire_receiver.attach(main_loop.loop_(), move |command| {
+                log::debug!("Handling command: {command:?}");
+                match command {
+                    StreamCommands::Eject(reply) => {
+                        // Take the callback holder our of its `Cell`, leaving `None` in place.
+                        let callback_holder = callback_holder.take();
+
+                        if callback_holder.is_none() {
+                            // We've already ejected the callback, nothing to do.
+                            return;
+                        }
+
+                        // Drop our reference to the Arc, which is its only persistent strong
+                        // reference. The `CallbackHolder` will go out of scope (usually right away,
+                        // but if the callback is running right now in the rt thread, then after it
+                        // releases it), and its Drop impl will send it through `callback_tx`.
+                        drop(callback_holder);
+
+                        let callback = callback_rx.recv().expect(
+                            "channel from StreamInner to receiver in pipewire main thread should \
+                             not be closed",
+                        );
+                        reply.send(callback).unwrap();
+                        if let Some(loop_ref) = loop_ref.upgrade() {
+                            loop_ref.quit();
+                        }
+                    }
+                }
+            });
+
             log::debug!("Starting Pipewire main loop");
             main_loop.run();
             Ok::<_, PipewireError>(())
         });
         Ok(Self {
-            commands: tx,
+            commands: pipewire_sender,
             handle,
         })
     }
@@ -342,3 +370,23 @@ const DEFAULT_EXPECTED_FRAMES: usize = 512;
 fn stream_buffer_size(range: (Option<usize>, Option<usize>)) -> usize {
     range.0.or(range.1).unwrap_or(DEFAULT_EXPECTED_FRAMES)
 }
+
+/// Returns a mutable reference into the given `Arc`, without any check.
+///
+/// This does the same thing as unstable [`Arc::get_mut_unchecked()`], but on stable Rust.
+/// The documentation including Safety prerequisites are copied from Rust stdlib.
+/// This helper can be removed once `get_mut_unchecked()` is stabilized and hits our MSRV.
+///
+/// Unsafe variant of [`Arc::get_mut()`], which is safe and does appropriate checks.
+///
+/// # Safety
+///
+/// If any other `Arc` or [`Weak`] pointers to the same allocation exist, then
+/// they must not be dereferenced or have active borrows for the duration
+/// of the returned borrow, and their inner type must be exactly the same as the
+/// inner type of this Rc (including lifetimes). This is trivially the case if no
+/// such pointers exist, for example immediately after `Arc::new`.
+unsafe fn arc_get_mut_unchecked<T>(arc: &mut Arc<T>) -> &mut T {
+    let raw_pointer = Arc::as_ptr(arc) as *mut T;
+    unsafe { &mut *raw_pointer }
+}

src/backends/pipewire/utils.rs

@@ -5,7 +5,9 @@ use pipewire::context::Context;
 use pipewire::main_loop::MainLoop;
 use pipewire::registry::GlobalObject;
 use std::cell::{Cell, RefCell};
+use std::ops::{Deref, DerefMut};
 use std::rc::Rc;
+use std::sync::mpsc;
 
 fn get_device_type(object: &GlobalObject<&DictRef>) -> Option<DeviceType> {
     fn is_input(media_class: &str) -> bool {
@@ -89,3 +91,70 @@ pub fn get_devices() -> Result<Vec<(u32, DeviceType, String)>, PipewireError> {
     drop(_listener_reg);
     Ok(Rc::into_inner(data).unwrap().into_inner())
 }
+
+/// A little helper that holds user's callback and sends it out using a channel when it goes out of
+/// scope. Dereferences to `Callback`, including mutably.
+pub(super) struct CallbackHolder<Callback> {
+    /// Invariant: `callback` is always `Some`, except in the second half of the [`Drop`] impl.
+    callback: Option<Callback>,
+    tx: mpsc::SyncSender<Callback>,
+}
+
+impl<Callback> CallbackHolder<Callback> {
+    /// Returns a pair (self, rx), where `rx` should be used to fetch the callback when the holder
+    /// goes out of scope.
+    pub(super) fn new(callback: Callback) -> (Self, mpsc::Receiver<Callback>) {
+        // Our first choice would be and `rtrb` channel, but that doesn't allow receiver to wait
+        // for a message, which we need. It doesn't matter, we use a channel of capacity 1 and
+        // we only use it exactly once, it never blocks in this case.
+        let (tx, rx) = mpsc::sync_channel(1);
+        let myself = Self {
+            callback: Some(callback),
+            tx,
+        };
+        (myself, rx)
+    }
+}
+
+impl<Callback> Deref for CallbackHolder<Callback> {
+    type Target = Callback;
+
+    fn deref(&self) -> &Callback {
+        self.callback
+            .as_ref()
+            .expect("never None outside destructor")
+    }
+}
+
+impl<Callback> DerefMut for CallbackHolder<Callback> {
+    fn deref_mut(&mut self) -> &mut Callback {
+        self.callback
+            .as_mut()
+            .expect("never None outside destructor")
+    }
+}
+
+impl<Callback> Drop for CallbackHolder<Callback> {
+    fn drop(&mut self) {
+        let callback = self.callback.take().expect("never None outside destructor");
+        match self.tx.try_send(callback) {
+            Ok(()) => (),
+            Err(mpsc::TrySendError::Full(_)) => {
+                panic!("The channel in CallbackHolder should be never full")
+            }
+            Err(mpsc::TrySendError::Disconnected(_)) => log::warn!(
+                "Channel in CallbackHolder is disconnected, did PipeWire main loop already exit?"
+            ),
+        }
+    }
+}
+
+/// Allows you to send to value to a black hole. It keeps at alive as long as it is in scope, but
+/// you cannot get the value back in any way.
+pub struct BlackHole<T>(T);
+
+impl<T> BlackHole<T> {
+    pub fn new(wrapped: T) -> Self {
+        Self(wrapped)
+    }
+}