Store client reference in lifetime Arc (#235)

Author: wmedrano

CHANGELOG.md

@@ -2,8 +2,13 @@
 
 ## Unreleased
 
-- Failed client activation now closes the client and frees both callback handlers instead of
-  leaking them ([#232](https://github.com/RustAudio/rust-jack/issues/232)).
+- Ports and transport objects now hold strong references to their JACK client and remain
+  usable after `Client` is dropped.
+- Added `ClientHandle` with an `as_ptr` method. `Port::from_raw` now takes
+  `Arc<ClientHandle>` instead of `Weak<()>` and no longer takes a separate client pointer.
+- Callback cleanup now explicitly stops processing and detaches callbacks before freeing
+  their handlers, including after failed activation, even if ports or transport keep the
+  client open ([#232](https://github.com/RustAudio/rust-jack/issues/232)).
 - `Error` is now `#[non_exhaustive]`. `match`es on it outside of this crate need a wildcard arm.
 - Added `Error::PortUnregistrationError`.
 - `Client::unregister_port` now returns `Err(Error::PortUnregistrationError)` on failure instead

src/client/async_client.rs

@@ -31,7 +31,7 @@ use crate::Error;
 ///     eprintln!("Error deactivating client: {err}");
 /// };
 /// ```
-#[must_use = "The jack client is shut down when the AsyncClient is dropped. You most likely want to keep this alive and manually tear down with `AsyncClient::deactivate`."]
+#[must_use = "Processing stops when the AsyncClient is dropped. You most likely want to keep this alive and manually tear down with `AsyncClient::deactivate`."]
 pub struct AsyncClient<N, P> {
     callback: Option<Box<CallbackContext<N, P>>>,
 }
@@ -48,8 +48,9 @@ where
     /// methods specified by the `NotificationHandler` and `ProcessHandler` objects.
     ///
     /// On failure, either `Err(Error::CallbackRegistrationError)` or
-    /// `Err(Error::ClientActivationError)` is returned. The client is closed and both handlers
-    /// are dropped.
+    /// `Err(Error::ClientActivationError)` is returned. Callbacks are detached and both handlers
+    /// are dropped. Ports and transport objects may keep the JACK client open.
+    /// If callback cleanup also fails, the callback context is retained to keep JACK's pointers valid.
     ///
     /// `notification_handler` and `process_handler` are consumed, but they are returned when
     /// `Client::deactivate` is called.
@@ -78,13 +79,26 @@ where
             callback_lock: std::sync::Mutex::new(()),
         });
         // Declare the guard after the context so it is released before error cleanup:
-        // Client::drop also acquires this mutex.
+        // ClientHandle::drop also acquires this mutex.
         let _m = CREATE_OR_DESTROY_CLIENT_MUTEX.lock().ok();
-        unsafe {
-            CallbackContext::register_callbacks(&mut callback_context)?;
-        }
-        if activate(&callback_context.client) != 0 {
-            return Err(Error::ClientActivationError);
+        let registration = unsafe { CallbackContext::register_callbacks(&mut callback_context) };
+        let result = registration.and_then(|()| {
+            if activate(&callback_context.client) == 0 {
+                Ok(())
+            } else {
+                Err(Error::ClientActivationError)
+            }
+        });
+        if let Err(error) = result {
+            if Self::detach_callbacks(&callback_context).is_err() {
+                crate::logging::log_error(
+                    b"Failed to detach callbacks after activation error. Leaking callback context.\0",
+                );
+                // JACK may still hold pointers into the context. Retain it rather
+                // than free memory that a callback could still access.
+                Box::leak(callback_context);
+            }
+            return Err(error);
         }
         Ok(AsyncClient {
             callback: Some(callback_context),
@@ -106,8 +120,9 @@ impl<N, P> AsyncClient<N, P> {
     /// The `handler` that was used for `Client::activate` is returned on success. Its state may
     /// have changed due to JACK calling its methods.
     ///
-    /// In the case of error, the `Client` is destroyed because its state is unknown, and it is
-    /// therefore unsafe to continue using.
+    /// In the case of error, the `Client` is dropped. Ports and transport objects may keep the
+    /// underlying JACK client open.
+    /// If callback cleanup fails, the callback context is retained to keep JACK's pointers valid.
     pub fn deactivate(self) -> Result<(Client, N, P), Error> {
         let mut c = self;
         c.maybe_deactivate()
@@ -123,29 +138,40 @@ impl<N, P> AsyncClient<N, P> {
             return Err(Error::ClientIsNoLongerAlive);
         }
         let cb = self.callback.take().ok_or(Error::ClientIsNoLongerAlive)?;
-        drop(cb.callback_lock.lock());
-        cb.is_valid_for_callback
-            .store(false, std::sync::atomic::Ordering::Relaxed);
-        // deactivate
-        if unsafe { j::jack_deactivate(cb.client.raw()) } != 0 {
+        if let Err(error) = Self::detach_callbacks(&cb) {
+            crate::logging::log_error(
+                b"Failed to detach callbacks on deactivation. Leaking callback context.\0",
+            );
+            // Keep the callback data alive if JACK could still use it.
+            Box::leak(cb);
             drop(m);
-            return Err(Error::ClientDeactivationError);
+            return Err(error);
         }
 
-        // clear the callbacks
-        unsafe { clear_callbacks(cb.client.raw()) }?;
         // done, take ownership of callback
         if cb.has_panic.load(std::sync::atomic::Ordering::Relaxed) {
             drop(m);
             return Err(Error::ClientPanicked);
         }
         Ok(cb)
     }
+
+    fn detach_callbacks(cb: &CallbackContext<N, P>) -> Result<(), Error> {
+        {
+            let _guard = cb.callback_lock.lock();
+            cb.is_valid_for_callback
+                .store(false, std::sync::atomic::Ordering::Relaxed);
+        }
+        if unsafe { j::jack_deactivate(cb.client.raw()) } != 0 {
+            return Err(Error::ClientDeactivationError);
+        }
+        unsafe { clear_callbacks(cb.client.raw()) }
+    }
 }
 
-/// Closes the client.
+/// Stops processing and releases this reference to the client.
 impl<N, P> Drop for AsyncClient<N, P> {
-    // Deactivate and close the client.
+    // Deactivate and release the client.
     fn drop(&mut self) {
         let _ = self.maybe_deactivate();
     }
@@ -195,15 +221,17 @@ mod tests {
 
     impl Drop for Handler {
         fn drop(&mut self) {
-            // The client must be closed before either handler is destroyed.
-            assert_eq!(self.port.name(), Err(Error::ClientIsNoLongerAlive));
+            // Each handler's port keeps the client alive during destruction.
+            assert!(self.port.name().is_ok());
             self.drops.fetch_add(1, Ordering::SeqCst);
         }
     }
 
     fn check_activation_failure(start_processing: bool) {
         let (client, _) = Client::new("", ClientOptions::default()).unwrap();
         let port = client.register_port("input", AudioIn::default()).unwrap();
+        let transport = client.transport();
+        let life = Arc::downgrade(&transport.client);
         let notification_drops = Arc::new(AtomicUsize::new(0));
         let process_drops = Arc::new(AtomicUsize::new(0));
         let (send, recv) = sync_channel(1);
@@ -229,16 +257,49 @@ mod tests {
         assert_eq!(result.err(), Some(Error::ClientActivationError));
         assert_eq!(notification_drops.load(Ordering::SeqCst), 1);
         assert_eq!(process_drops.load(Ordering::SeqCst), 1);
-        assert_eq!(port.name(), Err(Error::ClientIsNoLongerAlive));
+        assert!(port.name().is_ok());
+        // The retained client must be safe to activate after its original
+        // callback context and handlers have been freed.
+        assert_eq!(unsafe { j::jack_activate(port.client_ptr()) }, 0);
+        assert_eq!(unsafe { j::jack_deactivate(port.client_ptr()) }, 0);
+        drop(port);
+        assert!(life.upgrade().is_some());
+        drop(transport);
+        assert!(life.upgrade().is_none());
     }
 
     #[test]
-    fn activation_failure_closes_client_and_drops_handlers() {
+    fn activation_failure_detaches_callbacks_and_drops_handlers() {
         check_activation_failure(false);
     }
 
     #[test]
-    fn activation_failure_after_start_closes_client_and_drops_handlers() {
+    fn activation_failure_after_start_detaches_callbacks_and_drops_handlers() {
         check_activation_failure(true);
     }
+
+    #[test]
+    fn dropping_async_client_detaches_callbacks_while_port_keeps_client_alive() {
+        let (client, _) = Client::new("", ClientOptions::default()).unwrap();
+        let port = client.register_port("input", AudioIn::default()).unwrap();
+        let transport = client.transport();
+        let life = Arc::downgrade(&transport.client);
+        let drops = Arc::new(AtomicUsize::new(0));
+        let handler = || Handler {
+            drops: drops.clone(),
+            port: port.clone_unowned(),
+            processed: None,
+        };
+        let active = client.activate_async(handler(), handler()).unwrap();
+
+        drop(active);
+        assert_eq!(drops.load(Ordering::SeqCst), 2);
+        assert!(port.name().is_ok());
+        assert_eq!(unsafe { j::jack_activate(port.client_ptr()) }, 0);
+        assert_eq!(unsafe { j::jack_deactivate(port.client_ptr()) }, 0);
+
+        drop(port);
+        drop(transport);
+        assert!(life.upgrade().is_none());
+    }
 }

src/client/callbacks.rs

@@ -496,30 +496,45 @@ where
     }
 }
 
+unsafe extern "C" fn noop_thread_init_callback(_: *mut libc::c_void) {}
+
 /// Unsafe ffi wrapper that clears the callbacks registered to `client`.
 ///
 /// This is mostly for use within the jack crate itself.
 ///
 /// Returns `Err(Error::CallbackDeregistrationError)` on failure.
 ///
-/// # Unsafe
-///
-/// * Uses ffi calls, be careful.
-///
-/// # TODO
-///
-/// * Implement correctly. Freezes on my system. Maybe this makes sense now? it doesn't disturb my program
+/// # Safety
+/// The pointer must refer to a valid, inactive JACK client.
 pub unsafe fn clear_callbacks(client: *mut j::jack_client_t) -> Result<(), Error> {
-    j::jack_set_thread_init_callback(client, None, std::ptr::null_mut());
-    j::jack_set_process_callback(client, None, std::ptr::null_mut());
-    Ok(())
+    let data = std::ptr::null_mut();
+    j::jack_on_info_shutdown(client, None, data);
+    let results = [
+        // jackd2 rejects a null thread init callback, so install a no-op instead.
+        j::jack_set_thread_init_callback(client, Some(noop_thread_init_callback), data),
+        j::jack_set_process_callback(client, None, data),
+        j::jack_set_sync_callback(client, None, data),
+        j::jack_set_freewheel_callback(client, None, data),
+        j::jack_set_buffer_size_callback(client, None, data),
+        j::jack_set_sample_rate_callback(client, None, data),
+        j::jack_set_client_registration_callback(client, None, data),
+        j::jack_set_port_registration_callback(client, None, data),
+        j::jack_set_port_connect_callback(client, None, data),
+        j::jack_set_graph_order_callback(client, None, data),
+        j::jack_set_xrun_callback(client, None, data),
+    ];
+    if results.iter().any(|&result| result != 0) {
+        Err(Error::CallbackDeregistrationError)
+    } else {
+        Ok(())
+    }
 }
 
 /// The information used by JACK to process data.
 pub struct CallbackContext<N, P> {
     /// The underlying JACK client.
-    /// Keep this field first so the client is closed, stopping callbacks, before the handlers
-    /// and the rest of the context are dropped.
+    /// Callbacks must be stopped and detached before this context is dropped:
+    /// ports and transport may keep the JACK client open afterwards.
     pub client: Client,
     /// The handler for notifications.
     pub notification: N,