diff --git a/crates/rmw-zenoh-rs/Cargo.toml b/crates/rmw-zenoh-rs/Cargo.toml index 9f4b99d3..1e6667ac 100644 --- a/crates/rmw-zenoh-rs/Cargo.toml +++ b/crates/rmw-zenoh-rs/Cargo.toml @@ -57,3 +57,6 @@ test-all = ["test-core", "test-msgs"] # Legacy compatibility test-rmw = ["test-all"] + +[dev-dependencies] +pyo3 = { version = "0.22", features = ["auto-initialize"] } diff --git a/crates/rmw-zenoh-rs/src/pubsub.rs b/crates/rmw-zenoh-rs/src/pubsub.rs index fdd2b1a7..fb522a10 100644 --- a/crates/rmw-zenoh-rs/src/pubsub.rs +++ b/crates/rmw-zenoh-rs/src/pubsub.rs @@ -72,6 +72,94 @@ impl PublisherImpl { } } +/// The real notify-callback logic `rmw_create_subscription` wires into +/// `build_with_notifier`. Extracted so it's directly unit-testable without +/// a live zenoh session or the full `rmw_subscription_t` FFI chain. +/// +/// Copies the callback function pointer out of `callback_holder` and drops +/// the lock before calling it -- calling out while a lock is held risks a +/// self-deadlock if the callback re-enters and takes the same lock (e.g. a +/// GIL-holding executor thread calling back into this crate's own setter). +pub(crate) fn build_subscription_notify_callback( + notifier: std::sync::Arc, + callback_holder: std::sync::Arc< + std::sync::Mutex, + >, + user_data_holder: std::sync::Arc>, + unread_count_holder: std::sync::Arc>, +) -> impl Fn() + Send + Sync + 'static { + move || { + notifier.notify_all(); + // The `.lock()` temporary is dropped at the end of this statement -- + // released before any call-out below, unlike a `match`/`if let` + // scrutinee, which would extend it across the whole arm. + let Ok(callback_fn) = callback_holder.lock().map(|g| *g) else { + return; + }; + match callback_fn { + Some(callback_fn) => { + // Copied out and the lock released before the call-out -- + // the setter locks this same mutex, so holding it here + // would be a second AB-BA pair alongside `callback_holder`. + if let Ok(user_data_usize) = user_data_holder.lock().map(|g| *g) { + let user_data_ptr = user_data_usize as *const std::ffi::c_void; + unsafe { callback_fn(user_data_ptr, 1) }; // 1 new message + } + } + None => { + // No callback set, increment unread count + if let Ok(mut unread) = unread_count_holder.lock() { + *unread += 1; + } + } + } + } +} + +/// The real logic behind `rmw_subscription_set_on_new_message_callback`, +/// extracted for the same reason as [`build_subscription_notify_callback`] +/// above. Computes the retroactive-notification count and resets it, then +/// stores the new callback, then calls out -- all three locks released +/// before the call, none held during it. +pub(crate) fn set_subscription_callback_core( + callback_holder: &std::sync::Mutex, + user_data_holder: &std::sync::Mutex, + unread_count_holder: &std::sync::Mutex, + callback: crate::ros::rmw_subscription_new_message_callback_t, + user_data: *mut crate::c_void, +) { + if let Ok(mut ud) = user_data_holder.lock() { + *ud = user_data as usize; + } + + // Nested inside callback_holder's own lock, matching the pre-fix + // structure exactly: if callback_holder is poisoned, nothing here + // runs -- no unread reset, no call-out, no store. Computing `pending` + // independently of this lock would make a poisoned callback_holder + // silently reset progress and still fire the call, which is a real + // (if narrow) behavior change from before, not just a refactor. + let Ok(mut cb) = callback_holder.lock() else { + return; + }; + let pending = if callback.is_some() { + unread_count_holder.lock().ok().map(|mut unread| { + let n = *unread; + *unread = 0; + n + }) + } else { + None + }; + *cb = callback; + drop(cb); // released before any call-out below + + if let (Some(callback_fn), Some(n)) = (callback, pending) { + if n > 0 { + unsafe { callback_fn(user_data as *const std::ffi::c_void, n) }; + } + } +} + /// Subscription implementation for RMW pub struct SubscriptionImpl { pub inner: hiroz::pubsub::ZSub, @@ -769,3 +857,145 @@ pub extern "C" fn rmw_subscription_get_content_filter( // Content filtering is not supported yet RMW_RET_UNSUPPORTED as _ } + +#[cfg(test)] +mod gil_deadlock_tests { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::{Duration, Instant}; + + use crate::c_void; + + // Raw C function pointers carry no captured state, so the handshake + // between the notify thread and the setter thread has to live in + // statics. This file has exactly one test that touches these. + static GIL_ACQUIRED: AtomicBool = AtomicBool::new(false); + static ENTERED_CALLBACK: AtomicBool = AtomicBool::new(false); + static CALLBACK_RAN: AtomicBool = AtomicBool::new(false); + + /// Stands in for `rclpy`'s registered Python callback: the real path + /// this crate cannot see past `rmw_subscription_new_message_callback_t`, + /// which is a raw C function pointer that (via `rcl`'s + /// `RclEventCallbackTrampoline` and pybind11's auto-generated + /// Python-callable wrapper) wants the GIL to run user code. + unsafe extern "C" fn gil_wanting_callback(_user_data: *const std::ffi::c_void, _n: usize) { + ENTERED_CALLBACK.store(true, Ordering::SeqCst); + pyo3::Python::with_gil(|_py| { + CALLBACK_RAN.store(true, Ordering::SeqCst); + }); + } + + fn wait_flag(flag: &AtomicBool, timeout: Duration) -> bool { + let start = Instant::now(); + while !flag.load(Ordering::SeqCst) { + if start.elapsed() > timeout { + return false; + } + std::thread::sleep(Duration::from_millis(1)); + } + true + } + + /// Regression test for the callback-mutex/GIL AB-BA deadlock: a notify + /// thread calling out to a GIL-wanting callback, contended against a + /// setter thread that already holds the GIL and wants the same + /// `callback_holder` mutex. Before this fix, this pair deadlocked -- + /// this test asserts it no longer does, unconditionally (no feature + /// flag needed: the fix removes the hazard outright). + #[test] + fn subscription_notify_and_setter_do_not_deadlock_under_gil_contention() { + pyo3::prepare_freethreaded_python(); + GIL_ACQUIRED.store(false, Ordering::SeqCst); + ENTERED_CALLBACK.store(false, Ordering::SeqCst); + CALLBACK_RAN.store(false, Ordering::SeqCst); + + let notifier = Arc::new(crate::utils::Notifier::default()); + let callback_holder: Arc> = + Arc::new(Mutex::new(Some( + gil_wanting_callback as unsafe extern "C" fn(*const std::ffi::c_void, usize), + ))); + let user_data_holder = Arc::new(Mutex::new(0usize)); + let unread_count_holder = Arc::new(Mutex::new(0usize)); + + let notify = super::build_subscription_notify_callback( + notifier, + callback_holder.clone(), + user_data_holder.clone(), + unread_count_holder.clone(), + ); + + // Thread S: acquire the GIL first (uncontended, instant), then wait + // for confirmation that the notify thread is inside the callback + // before trying to also lock `callback_holder` -- exactly the real + // setter's shape, since `rmw_subscription_set_on_new_message_ + // callback` runs on a GIL-holding thread in real `rclpy` usage. + let cb2 = callback_holder.clone(); + let ud2 = user_data_holder.clone(); + let un2 = unread_count_holder.clone(); + let setter_thread = std::thread::spawn(move || { + pyo3::Python::with_gil(|_py| { + GIL_ACQUIRED.store(true, Ordering::SeqCst); + wait_flag(&ENTERED_CALLBACK, Duration::from_secs(3)); + super::set_subscription_callback_core( + &cb2, + &ud2, + &un2, + Some(gil_wanting_callback), + std::ptr::null_mut::(), + ); + }); + }); + + // Thread N: wait until S genuinely holds the GIL, then lock + // `callback_holder` and call the registered callback -- which + // wants the GIL, held by S. + let notify_thread = std::thread::spawn(move || { + wait_flag(&GIL_ACQUIRED, Duration::from_secs(3)); + notify(); + }); + + // Before the fix this pair deadlocked and these joins never + // returned. With the fix, both locks are released before either + // call-out, so both threads complete quickly regardless of + // scheduling order. + notify_thread.join().unwrap(); + setter_thread.join().unwrap(); + + assert!( + CALLBACK_RAN.load(Ordering::SeqCst), + "the callback never actually ran -- the repro did not exercise the real call-out" + ); + } + + /// Control: registering a callback with no unread messages pending does + /// not call out at all (see `set_subscription_callback_core`'s + /// `pending` computation), and must complete immediately either way. + #[test] + fn subscription_setter_with_no_pending_messages_does_not_panic() { + // Own reset of the shared statics: this test doesn't run the GIL + // contention scenario, just checks the n == 0 boundary, but reuses + // `gil_wanting_callback` (the only extern "C" fn available) as the + // registered callback, so it must confirm that fn body never runs. + ENTERED_CALLBACK.store(false, Ordering::SeqCst); + CALLBACK_RAN.store(false, Ordering::SeqCst); + + let callback_holder: Arc> = + Arc::new(Mutex::new(None)); + let user_data_holder = Arc::new(Mutex::new(0usize)); + let unread_count_holder = Arc::new(Mutex::new(0usize)); + + super::set_subscription_callback_core( + &callback_holder, + &user_data_holder, + &unread_count_holder, + Some(gil_wanting_callback), + std::ptr::null_mut::(), + ); + + assert!(callback_holder.lock().unwrap().is_some()); + assert!( + !ENTERED_CALLBACK.load(Ordering::SeqCst), + "the callback fired despite zero pending messages" + ); + } +} diff --git a/crates/rmw-zenoh-rs/src/rmw.rs b/crates/rmw-zenoh-rs/src/rmw.rs index 69589e7f..c842b347 100644 --- a/crates/rmw-zenoh-rs/src/rmw.rs +++ b/crates/rmw-zenoh-rs/src/rmw.rs @@ -543,29 +543,12 @@ pub extern "C" fn rmw_create_subscription( let unread_count_holder = std::sync::Arc::new(std::sync::Mutex::new(0usize)); // Track unread messages // Create notification callback that will wake up wait sets and invoke user callback - let notifier_clone = notifier.clone(); - let callback_holder_clone = callback_holder.clone(); - let user_data_holder_clone = user_data_holder.clone(); - let unread_count_clone = unread_count_holder.clone(); - let notify_callback = move || { - notifier_clone.notify_all(); - // Invoke the user callback if set, otherwise increment unread count - if let Ok(cb) = callback_holder_clone.lock() { - if let Some(callback_fn) = *cb { - if let Ok(user_data_usize) = user_data_holder_clone.lock() { - unsafe { - let user_data_ptr = *user_data_usize as *const std::ffi::c_void; - callback_fn(user_data_ptr, 1); // 1 new message - } - } - } else { - // No callback set, increment unread count - if let Ok(mut unread) = unread_count_clone.lock() { - *unread += 1; - } - } - } - }; + let notify_callback = crate::pubsub::build_subscription_notify_callback( + notifier.clone(), + callback_holder.clone(), + user_data_holder.clone(), + unread_count_holder.clone(), + ); let zsub = match zsub_builder.build_with_notifier(notify_callback) { Ok(zsub) => zsub, @@ -1234,29 +1217,12 @@ pub extern "C" fn rmw_create_service( let unread_count_holder = std::sync::Arc::new(std::sync::Mutex::new(0usize)); // Track unread requests // Create notification callback that will wake up wait sets and invoke user callback - let notifier_clone = notifier.clone(); - let callback_holder_clone = callback_holder.clone(); - let user_data_holder_clone = user_data_holder.clone(); - let unread_count_clone = unread_count_holder.clone(); - let notify_callback = move || { - notifier_clone.notify_all(); - // Invoke user callback if set, otherwise increment unread count - if let Ok(cb) = callback_holder_clone.lock() { - if let Some(callback_fn) = *cb { - if let Ok(user_data_usize) = user_data_holder_clone.lock() { - unsafe { - let user_data_ptr = *user_data_usize as *const std::ffi::c_void; - callback_fn(user_data_ptr, 1); // 1 new request - } - } - } else { - // No callback set, increment unread count - if let Ok(mut unread) = unread_count_clone.lock() { - *unread += 1; - } - } - } - }; + let notify_callback = crate::service::build_service_notify_callback( + notifier.clone(), + callback_holder.clone(), + user_data_holder.clone(), + unread_count_holder.clone(), + ); let zserver = match zserver_builder.build_with_notifier(notify_callback) { Ok(server) => server, @@ -2473,29 +2439,13 @@ pub extern "C" fn rmw_subscription_set_on_new_message_callback( Err(_) => return RMW_RET_INVALID_ARGUMENT as _, }; - // Set user_data first - if let Ok(mut ud) = subscription_impl.callback_user_data.lock() { - *ud = user_data as usize; // Store pointer as usize for thread safety - } - - // Then set callback and check for unread messages - if let Ok(mut cb) = subscription_impl.callback.lock() { - if callback.is_some() { - // Check if there are unread messages and invoke callback if needed - if let Ok(mut unread) = subscription_impl.unread_count.lock() { - if *unread > 0 { - // Invoke callback with unread count - unsafe { - if let Some(callback_fn) = callback { - callback_fn(user_data as *const std::ffi::c_void, *unread); - } - } - *unread = 0; // Reset unread count after notifying - } - } - } - *cb = callback; - } + crate::pubsub::set_subscription_callback_core( + &subscription_impl.callback, + &subscription_impl.callback_user_data, + &subscription_impl.unread_count, + callback, + user_data, + ); RMW_RET_OK as _ } @@ -2515,36 +2465,13 @@ pub extern "C" fn rmw_service_set_on_new_request_callback( Err(_) => return RMW_RET_INVALID_ARGUMENT as _, }; - if let Some(callback_fn) = callback { - // Push events arrived before setting the executor callback (retroactive notification) - if let Ok(mut unread) = service_impl.unread_count.lock() { - if *unread > 0 { - tracing::debug!( - "[rmw_service_set_on_new_request_callback] Invoking callback retroactively for {} unread requests", - *unread - ); - unsafe { - callback_fn(user_data as *const std::ffi::c_void, *unread); - } - *unread = 0; // Reset unread count after notification - } - } - // Store the new callback and user_data - if let Ok(mut cb) = service_impl.callback.lock() { - *cb = callback; - } - if let Ok(mut ud) = service_impl.callback_user_data.lock() { - *ud = user_data as usize; - } - } else { - // Callback is being cleared (set to None) - if let Ok(mut cb) = service_impl.callback.lock() { - *cb = None; - } - if let Ok(mut ud) = service_impl.callback_user_data.lock() { - *ud = 0; - } - } + crate::service::set_service_callback_core( + &service_impl.callback, + &service_impl.callback_user_data, + &service_impl.unread_count, + callback, + user_data, + ); RMW_RET_OK as _ } @@ -2564,36 +2491,13 @@ pub extern "C" fn rmw_client_set_on_new_response_callback( Err(_) => return RMW_RET_INVALID_ARGUMENT as _, }; - if let Some(callback_fn) = callback { - // Push events arrived before setting the executor callback (retroactive notification) - if let Ok(mut unread) = client_impl.unread_count.lock() { - if *unread > 0 { - tracing::debug!( - "[rmw_client_set_on_new_response_callback] Invoking callback retroactively for {} unread responses", - *unread - ); - unsafe { - callback_fn(user_data as *const std::ffi::c_void, *unread); - } - *unread = 0; // Reset unread count after notification - } - } - // Store the new callback and user_data - if let Ok(mut cb) = client_impl.callback.lock() { - *cb = callback; - } - if let Ok(mut ud) = client_impl.callback_user_data.lock() { - *ud = user_data as usize; - } - } else { - // Callback is being cleared (set to None) - if let Ok(mut cb) = client_impl.callback.lock() { - *cb = None; - } - if let Ok(mut ud) = client_impl.callback_user_data.lock() { - *ud = 0; - } - } + crate::service::set_client_callback_core( + &client_impl.callback, + &client_impl.callback_user_data, + &client_impl.unread_count, + callback, + user_data, + ); RMW_RET_OK as _ } diff --git a/crates/rmw-zenoh-rs/src/service.rs b/crates/rmw-zenoh-rs/src/service.rs index effb9259..6657698c 100644 --- a/crates/rmw-zenoh-rs/src/service.rs +++ b/crates/rmw-zenoh-rs/src/service.rs @@ -24,29 +24,93 @@ pub struct ClientImpl { pub entity: hiroz::entity::EndpointEntity, } -impl ClientImpl { - pub fn send_request(&self, request: *const c_void, sequence_id: *mut i64) -> Result<()> { - let req = crate::msg::RosMessage::new(request, self.request_ts.request); - - let notifier = self.notifier.clone(); - let callback_holder = self.callback.clone(); - let user_data_holder = self.callback_user_data.clone(); - let unread_count_holder = self.unread_count.clone(); - let notify_callback = move || { - notifier.notify_all(); - if let Ok(cb) = callback_holder.lock() { - if let Some(callback_fn) = *cb { - if let Ok(user_data_usize) = user_data_holder.lock() { - unsafe { - let user_data_ptr = *user_data_usize as *const std::ffi::c_void; - callback_fn(user_data_ptr, 1); - } - } - } else if let Ok(mut unread) = unread_count_holder.lock() { +/// The real notify-callback logic `ClientImpl::send_request` builds fresh +/// per request. See [`crate::pubsub::build_subscription_notify_callback`] +/// for why the lock must be released before the call-out. +pub(crate) fn build_client_notify_callback( + notifier: std::sync::Arc, + callback_holder: std::sync::Arc>, + user_data_holder: std::sync::Arc>, + unread_count_holder: std::sync::Arc>, +) -> impl Fn() + Send + Sync + 'static { + move || { + notifier.notify_all(); + let Ok(callback_fn) = callback_holder.lock().map(|g| *g) else { + return; + }; + match callback_fn { + Some(callback_fn) => { + // Copied out and the lock released before the call-out -- + // the setter locks this same mutex, so holding it here + // would be a second AB-BA pair alongside `callback_holder`. + if let Ok(user_data_usize) = user_data_holder.lock().map(|g| *g) { + let user_data_ptr = user_data_usize as *const std::ffi::c_void; + unsafe { callback_fn(user_data_ptr, 1) }; + } + } + None => { + if let Ok(mut unread) = unread_count_holder.lock() { *unread += 1; } } + } + } +} + +/// The real logic behind `rmw_client_set_on_new_response_callback`. See +/// [`crate::pubsub::set_subscription_callback_core`] for the same +/// collect/reset/store/call-out-last pattern. +pub(crate) fn set_client_callback_core( + callback_holder: &Mutex, + user_data_holder: &Mutex, + unread_count_holder: &Mutex, + callback: rmw_client_new_response_callback_t, + user_data: *mut c_void, +) { + let pending = if callback.is_some() { + unread_count_holder.lock().ok().map(|mut unread| { + let n = *unread; + *unread = 0; + n + }) + } else { + None + }; + + if let Ok(mut cb) = callback_holder.lock() { + *cb = callback; + } + if let Ok(mut ud) = user_data_holder.lock() { + // Matches the pre-fix behavior exactly: clearing the callback also + // zeroes the stored user_data, regardless of what was passed in. + *ud = if callback.is_some() { + user_data as usize + } else { + 0 }; + } + + if let (Some(callback_fn), Some(n)) = (callback, pending) { + if n > 0 { + tracing::debug!( + "[rmw_client_set_on_new_response_callback] Invoking callback retroactively for {} unread responses", + n + ); + unsafe { callback_fn(user_data as *const std::ffi::c_void, n) }; + } + } +} + +impl ClientImpl { + pub fn send_request(&self, request: *const c_void, sequence_id: *mut i64) -> Result<()> { + let req = crate::msg::RosMessage::new(request, self.request_ts.request); + + let notify_callback = build_client_notify_callback( + self.notifier.clone(), + self.callback.clone(), + self.callback_user_data.clone(), + self.unread_count.clone(), + ); // rmw_send_request returns the sequence number stamped into the attachment, // which is the same value the server will echo back. Use it directly as the @@ -169,6 +233,84 @@ pub struct ServiceImpl { pub entity: hiroz::entity::EndpointEntity, } +/// The real notify-callback logic `rmw_create_service` wires into +/// `build_with_notifier`. See +/// [`crate::pubsub::build_subscription_notify_callback`] for why the lock +/// must be released before the call-out. +pub(crate) fn build_service_notify_callback( + notifier: std::sync::Arc, + callback_holder: std::sync::Arc>, + user_data_holder: std::sync::Arc>, + unread_count_holder: std::sync::Arc>, +) -> impl Fn() + Send + Sync + 'static { + move || { + notifier.notify_all(); + let Ok(callback_fn) = callback_holder.lock().map(|g| *g) else { + return; + }; + match callback_fn { + Some(callback_fn) => { + // Copied out and the lock released before the call-out -- + // the setter locks this same mutex, so holding it here + // would be a second AB-BA pair alongside `callback_holder`. + if let Ok(user_data_usize) = user_data_holder.lock().map(|g| *g) { + let user_data_ptr = user_data_usize as *const std::ffi::c_void; + unsafe { callback_fn(user_data_ptr, 1) }; // 1 new request + } + } + None => { + if let Ok(mut unread) = unread_count_holder.lock() { + *unread += 1; + } + } + } + } +} + +/// The real logic behind `rmw_service_set_on_new_request_callback`. See +/// [`crate::pubsub::set_subscription_callback_core`] for the same +/// collect/reset/store/call-out-last pattern. +pub(crate) fn set_service_callback_core( + callback_holder: &Mutex, + user_data_holder: &Mutex, + unread_count_holder: &Mutex, + callback: rmw_service_new_request_callback_t, + user_data: *mut c_void, +) { + let pending = if callback.is_some() { + unread_count_holder.lock().ok().map(|mut unread| { + let n = *unread; + *unread = 0; + n + }) + } else { + None + }; + + if let Ok(mut cb) = callback_holder.lock() { + *cb = callback; + } + if let Ok(mut ud) = user_data_holder.lock() { + // Matches the pre-fix behavior exactly: clearing the callback also + // zeroes the stored user_data, regardless of what was passed in. + *ud = if callback.is_some() { + user_data as usize + } else { + 0 + }; + } + + if let (Some(callback_fn), Some(n)) = (callback, pending) { + if n > 0 { + tracing::debug!( + "[rmw_service_set_on_new_request_callback] Invoking callback retroactively for {} unread requests", + n + ); + unsafe { callback_fn(user_data as *const std::ffi::c_void, n) }; + } + } +} + impl ServiceImpl { pub fn take_request( &mut self, @@ -442,3 +584,173 @@ pub extern "C" fn rmw_client_response_subscription_get_actual_qos( ) -> rmw_ret_t { rmw_client_request_publisher_get_actual_qos(client, qos) } + +#[cfg(test)] +mod service_gil_deadlock_tests { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::{Duration, Instant}; + + use crate::c_void; + use crate::ros::rmw_service_new_request_callback_t; + + static GIL_ACQUIRED: AtomicBool = AtomicBool::new(false); + static ENTERED_CALLBACK: AtomicBool = AtomicBool::new(false); + static CALLBACK_RAN: AtomicBool = AtomicBool::new(false); + + unsafe extern "C" fn gil_wanting_callback(_user_data: *const std::ffi::c_void, _n: usize) { + ENTERED_CALLBACK.store(true, Ordering::SeqCst); + pyo3::Python::with_gil(|_py| { + CALLBACK_RAN.store(true, Ordering::SeqCst); + }); + } + + fn wait_flag(flag: &AtomicBool, timeout: Duration) -> bool { + let start = Instant::now(); + while !flag.load(Ordering::SeqCst) { + if start.elapsed() > timeout { + return false; + } + std::thread::sleep(Duration::from_millis(1)); + } + true + } + + /// Same shape as `pubsub::gil_deadlock_tests`, for `ServiceImpl`. + #[test] + fn service_notify_and_setter_do_not_deadlock_under_gil_contention() { + pyo3::prepare_freethreaded_python(); + GIL_ACQUIRED.store(false, Ordering::SeqCst); + ENTERED_CALLBACK.store(false, Ordering::SeqCst); + CALLBACK_RAN.store(false, Ordering::SeqCst); + + let notifier = Arc::new(crate::utils::Notifier::default()); + let callback_holder: Arc> = Arc::new(Mutex::new( + Some(gil_wanting_callback as unsafe extern "C" fn(*const std::ffi::c_void, usize)), + )); + let user_data_holder = Arc::new(Mutex::new(0usize)); + let unread_count_holder = Arc::new(Mutex::new(0usize)); + + let notify = super::build_service_notify_callback( + notifier, + callback_holder.clone(), + user_data_holder.clone(), + unread_count_holder.clone(), + ); + + let cb2 = callback_holder.clone(); + let ud2 = user_data_holder.clone(); + let un2 = unread_count_holder.clone(); + let setter_thread = std::thread::spawn(move || { + pyo3::Python::with_gil(|_py| { + GIL_ACQUIRED.store(true, Ordering::SeqCst); + wait_flag(&ENTERED_CALLBACK, Duration::from_secs(3)); + super::set_service_callback_core( + &cb2, + &ud2, + &un2, + Some(gil_wanting_callback), + std::ptr::null_mut::(), + ); + }); + }); + + let notify_thread = std::thread::spawn(move || { + wait_flag(&GIL_ACQUIRED, Duration::from_secs(3)); + notify(); + }); + + notify_thread.join().unwrap(); + setter_thread.join().unwrap(); + + assert!( + CALLBACK_RAN.load(Ordering::SeqCst), + "the callback never actually ran -- the repro did not exercise the real call-out" + ); + } +} + +#[cfg(test)] +mod client_gil_deadlock_tests { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::{Duration, Instant}; + + use crate::c_void; + use crate::ros::rmw_client_new_response_callback_t; + + static GIL_ACQUIRED: AtomicBool = AtomicBool::new(false); + static ENTERED_CALLBACK: AtomicBool = AtomicBool::new(false); + static CALLBACK_RAN: AtomicBool = AtomicBool::new(false); + + unsafe extern "C" fn gil_wanting_callback(_user_data: *const std::ffi::c_void, _n: usize) { + ENTERED_CALLBACK.store(true, Ordering::SeqCst); + pyo3::Python::with_gil(|_py| { + CALLBACK_RAN.store(true, Ordering::SeqCst); + }); + } + + fn wait_flag(flag: &AtomicBool, timeout: Duration) -> bool { + let start = Instant::now(); + while !flag.load(Ordering::SeqCst) { + if start.elapsed() > timeout { + return false; + } + std::thread::sleep(Duration::from_millis(1)); + } + true + } + + /// Same shape as `pubsub::gil_deadlock_tests`, for `ClientImpl`. + #[test] + fn client_notify_and_setter_do_not_deadlock_under_gil_contention() { + pyo3::prepare_freethreaded_python(); + GIL_ACQUIRED.store(false, Ordering::SeqCst); + ENTERED_CALLBACK.store(false, Ordering::SeqCst); + CALLBACK_RAN.store(false, Ordering::SeqCst); + + let notifier = Arc::new(crate::utils::Notifier::default()); + let callback_holder: Arc> = Arc::new(Mutex::new( + Some(gil_wanting_callback as unsafe extern "C" fn(*const std::ffi::c_void, usize)), + )); + let user_data_holder = Arc::new(Mutex::new(0usize)); + let unread_count_holder = Arc::new(Mutex::new(0usize)); + + let notify = super::build_client_notify_callback( + notifier, + callback_holder.clone(), + user_data_holder.clone(), + unread_count_holder.clone(), + ); + + let cb2 = callback_holder.clone(); + let ud2 = user_data_holder.clone(); + let un2 = unread_count_holder.clone(); + let setter_thread = std::thread::spawn(move || { + pyo3::Python::with_gil(|_py| { + GIL_ACQUIRED.store(true, Ordering::SeqCst); + wait_flag(&ENTERED_CALLBACK, Duration::from_secs(3)); + super::set_client_callback_core( + &cb2, + &ud2, + &un2, + Some(gil_wanting_callback), + std::ptr::null_mut::(), + ); + }); + }); + + let notify_thread = std::thread::spawn(move || { + wait_flag(&GIL_ACQUIRED, Duration::from_secs(3)); + notify(); + }); + + notify_thread.join().unwrap(); + setter_thread.join().unwrap(); + + assert!( + CALLBACK_RAN.load(Ordering::SeqCst), + "the callback never actually ran -- the repro did not exercise the real call-out" + ); + } +}