diff --git a/crates/hiroz-tests/tests/reentrant_event_status.rs b/crates/hiroz-tests/tests/reentrant_event_status.rs new file mode 100644 index 000000000..9dc408787 --- /dev/null +++ b/crates/hiroz-tests/tests/reentrant_event_status.rs @@ -0,0 +1,161 @@ +//! Re-entrancy audit for endpoint event-status updates. +//! +//! `EventsManager` is shared as `Arc>`, so `&mut self` proves the +//! caller holds that mutex — and `update_event_status` fired the callback from +//! there. The callback is user code the rmw layer hands to an rclcpp executor, +//! and the first thing it usually does is ask the handle that fired for its +//! status (`rmw_take_event` → `RmEventHandle::take_event`), locking the same +//! mutex on the same thread. Non-reentrant, no race needed. +//! +//! Fix: record under the lock, drop the guard, invoke. +//! `record_event_status_with_policy` returns the callback rather than calling +//! it, and `update_shared_event_status[_with_policy]` is what holders use. +//! +//! Each scenario runs on its own thread behind a deadline, so a deadlock fails +//! the test instead of wedging the suite. +//! +//! # What these detect, and what they do not +//! +//! Both call `update_shared_event_status`, which this change *introduces* — so +//! a wholesale revert does not turn them red, it stops this file compiling. +//! They are not evidence the old shape deadlocked; `reentrant_graph_event.rs` +//! carries that. +//! +//! They are still detectors, for the property rather than the history: +//! reinstate the callout inside `update_shared_event_status_with_policy` and +//! both fail on their deadline. That is the regression worth guarding, since +//! the old entry point is gone. + +use std::{ + sync::{ + Arc, Mutex, + atomic::{AtomicI32, Ordering}, + mpsc, + }, + thread, + time::Duration, +}; + +use hiroz::{ + GidArray, + event::{ + EventsManager, RmEventHandle, ZenohEventType, update_shared_event_status, + update_shared_event_status_with_policy, + }, +}; + +/// Budget for one scenario. Generous relative to the work done — anything +/// slower than this is a hang, not slowness. +const SCENARIO_TIMEOUT: Duration = Duration::from_secs(30); + +/// Run `scenario` on its own thread; fail (rather than hang) past the deadline. +/// +/// On timeout the worker is deliberately left running: it is blocked on a lock +/// that will never be released, and there is no sound way to unwind it. +fn with_deadline(name: &'static str, scenario: impl FnOnce() + Send + 'static) { + let (tx, rx) = mpsc::channel(); + thread::spawn(move || { + scenario(); + let _ = tx.send(()); + }); + match rx.recv_timeout(SCENARIO_TIMEOUT) { + Ok(()) => {} + // The worker panicked and dropped the sender. That is an assertion + // failure inside the scenario, NOT a deadlock — reporting it as one + // would turn every ordinary test failure into a false deadlock report. + Err(mpsc::RecvTimeoutError::Disconnected) => { + panic!("{name}: scenario panicked — see the worker thread's panic above") + } + Err(mpsc::RecvTimeoutError::Timeout) => { + panic!("{name}: scenario did not finish within {SCENARIO_TIMEOUT:?} — deadlock") + } + } +} + +fn gid(n: u8) -> GidArray { + let mut g = [0u8; 16]; + g[0] = n; + g +} + +/// The canonical rmw shape: the matched-event callback immediately takes the +/// status that triggered it. +/// +/// `take_event` locks the same `Arc>` the update path +/// holds, so with the callback invoked under that guard this never returns. +#[test] +fn event_callback_taking_its_own_status_does_not_deadlock() { + with_deadline("event_callback_take_event", || { + let mgr = Arc::new(Mutex::new(EventsManager::new(gid(1)))); + let handle = Arc::new(RmEventHandle::new( + mgr.clone(), + ZenohEventType::SubscriptionMatched, + )); + + let observed = Arc::new(AtomicI32::new(-1)); + { + let handle_in_cb = handle.clone(); + let observed = observed.clone(); + handle.set_callback(move |_change| { + let status = handle_in_cb.take_event(); + observed.store(status.total_count, Ordering::SeqCst); + }); + } + + update_shared_event_status(&mgr, ZenohEventType::SubscriptionMatched, 1); + + assert_eq!( + observed.load(Ordering::SeqCst), + 1, + "the callback did not observe the status change that triggered it" + ); + // The callback consumed the change counters via `take_event`. + assert!( + !handle.is_ready(), + "take_event inside the callback should have cleared the changed flag" + ); + }); +} + +/// A QoS-incompatibility callback that re-arms itself. +/// +/// `set_callback` locks the manager to install, so a callback that replaces +/// itself re-enters the outer mutex exactly like `take_event` does. This also +/// covers the `_with_policy` entry point, which carries the encoded policy kind. +#[test] +fn event_callback_reinstalling_itself_does_not_deadlock() { + with_deadline("event_callback_reinstall", || { + let mgr = Arc::new(Mutex::new(EventsManager::new(gid(2)))); + let handle = Arc::new(RmEventHandle::new( + mgr.clone(), + ZenohEventType::RequestedQosIncompatible, + )); + + let fired = Arc::new(AtomicI32::new(0)); + { + let handle_in_cb = handle.clone(); + let fired = fired.clone(); + handle.set_callback(move |_change| { + fired.fetch_add(1, Ordering::SeqCst); + // Re-arm with a no-op. Installing takes the manager lock. + handle_in_cb.set_callback(|_| {}); + }); + } + + update_shared_event_status_with_policy( + &mgr, + ZenohEventType::RequestedQosIncompatible, + 1, + 42, + ); + + assert_eq!( + fired.load(Ordering::SeqCst), + 1, + "the re-arming callback did not run" + ); + let status = handle.take_event(); + assert_eq!(status.total_count, 1); + assert_eq!(status.last_policy_kind, 42); + }); +} diff --git a/crates/hiroz-tests/tests/reentrant_graph_event.rs b/crates/hiroz-tests/tests/reentrant_graph_event.rs new file mode 100644 index 000000000..eed9e5da7 --- /dev/null +++ b/crates/hiroz-tests/tests/reentrant_graph_event.rs @@ -0,0 +1,310 @@ +//! Re-entrancy audit for graph-change and endpoint event callbacks. +//! +//! `GraphEventManager` invoked its registered callbacks *while holding the +//! registries they are registered in*: `trigger_event_with_policy` called the +//! callback under the `event_callbacks` guard, and `trigger_graph_change` held +//! both `event_callbacks` and `entity_topics` for the whole notification loop. +//! +//! Worse, the hot path into `trigger_graph_change` is the liveliness subscriber +//! installed by `Graph::new_with_pattern`, which used to hold the `GraphData` +//! mutex across the call. So on every liveliness token the callback ran with +//! three or four non-reentrant locks held — and these callbacks are user code: +//! the rmw layer hands them straight to an rclcpp executor. Any callback that +//! re-entered hiroz (counting publishers, unregistering an entity, registering a +//! new one) self-deadlocked on the thread that already held the guard. +//! +//! The fix is the same shape zenoh core itself uses in `resolve_put`: collect +//! the callbacks under the lock, drop every guard, then invoke. Callbacks are +//! `Arc` rather than `Box` so they can be cloned out cheaply. +//! +//! Every scenario runs on a dedicated thread behind a hard deadline, so a +//! re-entrancy deadlock fails the test instead of wedging the suite — the same +//! shape as `reentrant_publish.rs` and `reentrant_service.rs`. + +mod common; + +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, + mpsc, + }, + thread, + time::Duration, +}; + +use common::{TestRouter, create_hiroz_context_with_endpoint}; +use hiroz::{ + Builder, GidArray, TypeHash, + entity::EndpointKind, + event::{GraphEventManager, ZenohEventType}, + ros_msg::MessageTypeInfo, +}; +use serde::{Deserialize, Serialize}; +use serial_test::serial; + +/// A self-contained message type, so this file does not depend on the +/// `ros-msgs` feature. Same shape as `reentrant_publish.rs`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +struct Tick { + counter: u64, +} + +impl MessageTypeInfo for Tick { + fn type_name() -> &'static str { + "test_msgs::msg::dds_::Tick_" + } + + fn type_hash() -> TypeHash { + TypeHash::zero() + } +} + +impl hiroz::ros_msg::WithTypeInfo for Tick {} + +impl hiroz::msg::ZMessage for Tick { + type Serdes = hiroz::msg::SerdeCdrSerdes; +} + +/// Budget for one scenario. Generous relative to the work done — anything slower +/// than this is a hang, not slowness. +const SCENARIO_TIMEOUT: Duration = Duration::from_secs(30); + +/// Run `scenario` on its own thread; fail (rather than hang) past the deadline. +/// +/// On timeout the worker is deliberately left running: it is blocked on a lock +/// that will never be released, and there is no sound way to unwind it. +fn with_deadline(name: &'static str, scenario: impl FnOnce() + Send + 'static) { + let (tx, rx) = mpsc::channel(); + thread::spawn(move || { + scenario(); + let _ = tx.send(()); + }); + match rx.recv_timeout(SCENARIO_TIMEOUT) { + Ok(()) => {} + // The worker panicked and dropped the sender. That is an assertion + // failure inside the scenario, NOT a deadlock — reporting it as one + // would turn every ordinary test failure into a false deadlock report. + Err(mpsc::RecvTimeoutError::Disconnected) => { + panic!("{name}: scenario panicked — see the worker thread's panic above") + } + Err(mpsc::RecvTimeoutError::Timeout) => { + panic!("{name}: scenario did not finish within {SCENARIO_TIMEOUT:?} — deadlock") + } + } +} + +/// A fixed, valid `ZenohId`. `trigger_graph_change` does not read it. +fn test_zid() -> zenoh::session::ZenohId { + use std::str::FromStr; + zenoh::session::ZenohId::from_str("221b72df20924c15b8794c6bdb471150").expect("zid") +} + +fn gid(n: u8) -> GidArray { + let mut g = [0u8; 16]; + g[0] = n; + g +} + +// --------------------------------------------------------------------------- +// Registry-level scenarios (deterministic, no router needed) +// --------------------------------------------------------------------------- + +/// An endpoint event callback that unregisters an entity. +/// +/// `trigger_event_with_policy` used to hold `event_callbacks` across the call and +/// `unregister_entity` takes the same `std::sync::Mutex`, so this re-entered a +/// mutex the calling thread already held — a self-deadlock with no race needed. +/// +/// Tearing down an endpoint from inside its own matched-event callback is exactly +/// what an rmw/rclcpp user does when a "publisher went away" event triggers +/// cleanup, so this is a reachable shape rather than a contrived one. +#[test] +fn event_callback_unregistering_does_not_deadlock() { + with_deadline("event_callback_unregister", || { + let mgr = Arc::new(GraphEventManager::new()); + let ran = Arc::new(AtomicBool::new(false)); + + let ran_c = ran.clone(); + // Weak, so the callback (owned by the manager) does not keep it alive. + let mgr_weak = Arc::downgrade(&mgr); + mgr.register_event_callback( + gid(1), + "/reentrant".to_string(), + ZenohEventType::PublicationMatched, + move |_change| { + ran_c.store(true, Ordering::SeqCst); + if let Some(m) = mgr_weak.upgrade() { + // Re-entrant registry access from inside the callback. + m.unregister_entity(&gid(2)); + } + }, + ) + .expect("register"); + + mgr.trigger_event(&gid(1), ZenohEventType::PublicationMatched, 1); + + assert!(ran.load(Ordering::SeqCst), "the event callback never ran"); + }); +} + +/// A graph-change callback that registers a *new* entity. +/// +/// `trigger_graph_change` used to hold both `event_callbacks` and `entity_topics` +/// for the whole notification loop; `register_event_callback` takes both. Same +/// self-deadlock, on the graph-change path rather than the endpoint-event path. +#[test] +fn graph_change_callback_registering_does_not_deadlock() { + with_deadline("graph_change_register", || { + use hiroz::entity::{EndpointEntity, Entity}; + + let mgr = Arc::new(GraphEventManager::new()); + let ran = Arc::new(AtomicUsize::new(0)); + + let ran_c = ran.clone(); + let mgr_weak = Arc::downgrade(&mgr); + mgr.register_event_callback( + gid(1), + "/reentrant".to_string(), + // A Publisher appearing notifies subscriptions. + ZenohEventType::SubscriptionMatched, + move |_change| { + // Only re-enter once, or this registers forever. + if ran_c.fetch_add(1, Ordering::SeqCst) == 0 + && let Some(m) = mgr_weak.upgrade() + { + // Re-entrant registration from inside the callback. + let _ = m.register_event_callback( + gid(9), + "/other".to_string(), + ZenohEventType::SubscriptionMatched, + |_| {}, + ); + } + }, + ) + .expect("register"); + + let appearing = Entity::Endpoint(EndpointEntity { + id: 1, + node: None, + kind: EndpointKind::Publisher, + topic: "/reentrant".to_string(), + type_info: None, + qos: Default::default(), + }); + mgr.trigger_graph_change(&appearing, true, test_zid()); + + assert_eq!( + ran.load(Ordering::SeqCst), + 1, + "the graph-change callback never ran" + ); + }); +} + +// --------------------------------------------------------------------------- +// End-to-end: the liveliness path that holds GraphData +// --------------------------------------------------------------------------- + +/// A graph-change callback that queries the graph, driven by a *remote* entity. +/// +/// This is the full hazard, not just the registry half. A remote entity arrives +/// on the liveliness subscriber declared in `Graph::new_with_pattern`, whose +/// callback held the `GraphData` mutex across `trigger_graph_change`. A +/// graph-change callback that asks the graph anything — `count`, +/// `get_topic_names_and_types`, `node_exists` — takes that same mutex on the same +/// thread and never returns. +/// +/// Counting matched endpoints from inside a matched-event callback is the +/// canonical rmw use, so this is the shape that actually ships. +#[test] +#[serial] +fn graph_change_callback_querying_the_graph_does_not_deadlock() { + with_deadline("graph_change_query_graph", || { + const TOPIC: &str = "/reentrant_graph_event"; + + let router = TestRouter::new(); + + // Observer side: registers the graph-change callback. + let ctx_a = + create_hiroz_context_with_endpoint(router.endpoint()).expect("observer context"); + let node_a = ctx_a + .create_node("graph_evt_observer") + .build() + .expect("node a"); + + // A local publisher, only so we can read back the *qualified* topic name + // the graph indexes entities under. + let local_pub = node_a.create_pub::(TOPIC).build().expect("local pub"); + let qualified_topic = local_pub.entity().topic.clone(); + + let graph_a = node_a.graph().clone(); + let observed = Arc::new(AtomicUsize::new(0)); + let counted = Arc::new(AtomicUsize::new(0)); + + let observed_c = observed.clone(); + let counted_c = counted.clone(); + // Weak: the callback is owned by the graph's event manager, which the + // graph owns — an Arc here would be a cycle. + let graph_weak = Arc::downgrade(&graph_a); + graph_a + .event_manager + .register_event_callback( + gid(7), + qualified_topic.clone(), + // A Publisher appearing notifies subscriptions. + ZenohEventType::SubscriptionMatched, + move |_change| { + observed_c.fetch_add(1, Ordering::SeqCst); + if let Some(g) = graph_weak.upgrade() { + // Re-entrant graph query from inside the callback: this + // takes the same `GraphData` mutex the liveliness + // callback holds. + let n = g.count(EndpointKind::Publisher, &qualified_topic); + counted_c.store(n, Ordering::SeqCst); + } + }, + ) + .expect("register graph event callback"); + + // Remote side: a second context whose publisher reaches the observer + // through a liveliness token, i.e. through the subscriber callback that + // used to hold `GraphData`. + let ctx_b = create_hiroz_context_with_endpoint(router.endpoint()).expect("remote context"); + let node_b = ctx_b + .create_node("graph_evt_remote") + .build() + .expect("node b"); + let _remote_pub = node_b + .create_pub::(TOPIC) + .build() + .expect("remote pub"); + + // Wait for the liveliness token to propagate and the callback to run. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while observed.load(Ordering::SeqCst) == 0 && std::time::Instant::now() < deadline { + thread::sleep(Duration::from_millis(50)); + } + + assert!( + observed.load(Ordering::SeqCst) >= 1, + "the graph-change callback never ran for the remote publisher — \ + the scenario proved nothing" + ); + // Two, not one. `local_pub` is on this node for the whole scenario, so + // `>= 1` was satisfiable without the remote publisher ever being in the + // graph — which is precisely the regression this is meant to catch: a + // callback invoked *before* the entity is inserted would still observe + // the local publisher and pass. Requiring both makes the assertion + // actually depend on the ordering it claims to verify. + assert!( + counted.load(Ordering::SeqCst) >= 2, + "the re-entrant graph query saw {} publisher(s) on the topic; it must \ + see both the local one and the remote one whose appearance triggered \ + this callback. Seeing exactly 1 means the callback ran before the \ + remote entity was inserted into the graph", + counted.load(Ordering::SeqCst) + ); + }); +} diff --git a/crates/hiroz/src/event.rs b/crates/hiroz/src/event.rs index 912c63806..9dea9e7a1 100644 --- a/crates/hiroz/src/event.rs +++ b/crates/hiroz/src/event.rs @@ -1,6 +1,8 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; + +use crate::reentrancy::TrackedMutex; use zenoh::Result; use crate::GidArray; @@ -35,10 +37,29 @@ pub struct ZenohEventStatus { pub last_policy_kind: u32, // RMW QoS policy kind that caused incompatibility } -// Event callback type -pub type EventCallback = Box; - -// EventsManager - manages event state for a single publisher/subscription +// Event callback type. +// +// `Arc` rather than `Box` so that a callback can be *collected* while the +// registry lock is held and *invoked* after it has been released. Event +// callbacks are user code (the rmw layer hands them straight to an rclcpp +// executor), and they routinely re-enter hiroz — querying the graph, +// publishing, unregistering an entity. Invoking them under the registry guard +// makes any such re-entry a self-deadlock on a non-reentrant `Mutex`. +pub type EventCallback = Arc; + +/// Event state for a single publisher/subscription. +/// +/// # The `&mut self` rule +/// +/// This type is shared as `Arc>`, so **`&mut self` proves +/// the caller holds that outer mutex**. Any `&mut self` method that invoked a +/// callback would therefore run user code under a lock the same thread already +/// holds — and these callbacks re-enter (`rmw_take_event` → +/// [`RmEventHandle::take_event`], or `set_callback` to re-arm). Non-reentrant +/// mutex, one thread, no race: a guaranteed deadlock. +/// +/// So the `&mut self` methods here *return* the callback instead, and +/// [`update_shared_event_status`] is the entry point holders actually use. pub struct EventsManager { event_statuses: Vec, event_callbacks: Vec>, @@ -60,33 +81,78 @@ impl EventsManager { } } + /// Install a callback, delivering any backlog immediately. + /// + /// **Only for a caller that owns this manager outright.** It fires the + /// backlog while the caller's outer guard is still live — the lock this + /// releases is only the *inner* `event_mutex` — so a re-entering callback + /// deadlocks. See the `&mut self` note on [`EventsManager`]; holders of an + /// `Arc>` want [`RmEventHandle::set_callback`] instead. pub fn set_callback(&mut self, event_type: ZenohEventType, callback: F) where F: Fn(i32) + Send + Sync + 'static, { + let callback: EventCallback = Arc::new(callback); + let unread_count = self.install_callback(event_type, callback.clone()); + // Outside the inner `event_mutex` only — see the note above. + if unread_count != 0 { + callback(unread_count); + } + } + + /// Install `callback` and take (clearing) the unread-event backlog. + /// + /// Deliberately does *not* invoke the callback: the caller fires it after + /// releasing every lock it holds, including the outer `Mutex` + /// that [`RmEventHandle`] uses. Returns the backlog count, or 0 if none. + pub fn install_callback(&mut self, event_type: ZenohEventType, callback: EventCallback) -> i32 { let event_id = event_type as usize; let _lock = self.event_mutex.lock().unwrap(); - // If there are unread events, trigger the callback immediately let unread_count = self.event_statuses[event_id].total_count_change; if unread_count != 0 { - callback(unread_count); self.event_statuses[event_id].total_count_change = 0; } + self.event_callbacks[event_id] = Some(callback); - self.event_callbacks[event_id] = Some(Box::new(callback)); + unread_count } + /// Record a status change and invoke the registered callback, if any. + /// + /// Only safe when the caller does not hold the outer `Mutex` + /// — see [`update_shared_event_status`], which is what every holder of an + /// `Arc>` must use instead. pub fn update_event_status(&mut self, event_type: ZenohEventType, change: i32) { self.update_event_status_with_policy(event_type, change, 0); } + /// See [`EventsManager::update_event_status`] for the locking caveat. pub fn update_event_status_with_policy( &mut self, event_type: ZenohEventType, change: i32, policy_kind: u32, ) { + if let Some(callback) = + self.record_event_status_with_policy(event_type, change, policy_kind) + { + callback(change); + } + } + + /// Record a status change and hand back the callback that owes a + /// notification, *without* invoking it. + /// + /// Not invoking is the point: see the `&mut self` note on + /// [`EventsManager`]. The caller invokes once every guard is dropped. + #[must_use = "the returned callback must be invoked after every guard is dropped"] + pub fn record_event_status_with_policy( + &mut self, + event_type: ZenohEventType, + change: i32, + policy_kind: u32, + ) -> Option { let event_id = event_type as usize; { @@ -104,10 +170,7 @@ impl EventsManager { } } - // Trigger callback if registered - if let Some(ref callback) = self.event_callbacks[event_id] { - callback(change); - } + self.event_callbacks[event_id].clone() } pub fn take_event_status(&mut self, event_type: ZenohEventType) -> ZenohEventStatus { @@ -128,15 +191,33 @@ impl EventsManager { } } -// Callback type for triggering graph guard conditions -pub type GraphGuardConditionTrigger = Box; +/// A graph guard condition this manager may trigger on a graph change. +/// +/// Registrations are **owned**, not raw pointers, and that is the whole point. +/// Triggering happens with the registry lock released — it must, since the +/// trigger re-enters hiroz — and a raw pointer cloned out of the lock is only +/// valid if something keeps the target alive across the call. Nothing did: +/// `rmw_destroy_node` unregisters and immediately frees, so a destroy landing +/// mid-call dereferenced freed memory. +/// +/// **Implementors must keep [`trigger`] valid after the owning C object is +/// gone.** The natural shape is state behind its own `Arc`, one reference held +/// by the C handle and one by this registry. +/// +/// [`trigger`]: GraphGuardCondition::trigger +pub trait GraphGuardCondition: Send + Sync { + /// Wake whatever is waiting on this guard condition. + /// + /// Called with no hiroz lock held, possibly concurrently, and possibly + /// after the corresponding C handle has been destroyed. + fn trigger(&self); +} // GraphCache event integration pub struct GraphEventManager { - event_callbacks: Mutex>>, - entity_topics: Mutex>, // Topic name per registered entity - graph_guard_conditions: Mutex>, // Pointers as usize for Send - trigger_guard_condition: Mutex>, + event_callbacks: TrackedMutex>>, + entity_topics: TrackedMutex>, // Topic name per registered entity + graph_guard_conditions: TrackedMutex>>, } impl Default for GraphEventManager { @@ -148,17 +229,12 @@ impl Default for GraphEventManager { impl GraphEventManager { pub fn new() -> Self { Self { - event_callbacks: Mutex::new(HashMap::new()), - entity_topics: Mutex::new(HashMap::new()), - graph_guard_conditions: Mutex::new(Vec::new()), - trigger_guard_condition: Mutex::new(None), + event_callbacks: TrackedMutex::new(HashMap::new()), + entity_topics: TrackedMutex::new(HashMap::new()), + graph_guard_conditions: TrackedMutex::new(Vec::new()), } } - pub fn set_guard_condition_trigger(&self, trigger: GraphGuardConditionTrigger) { - *self.trigger_guard_condition.lock().unwrap() = Some(trigger); - } - pub fn register_event_callback( &self, entity_gid: GidArray, @@ -169,9 +245,11 @@ impl GraphEventManager { where F: Fn(i32) + Send + Sync + 'static, { - let mut callbacks = self.event_callbacks.lock().unwrap(); - let entity_callbacks = callbacks.entry(entity_gid).or_default(); - entity_callbacks.insert(event_type, Box::new(callback)); + { + let mut callbacks = self.event_callbacks.lock().unwrap(); + let entity_callbacks = callbacks.entry(entity_gid).or_default(); + entity_callbacks.insert(event_type, Arc::new(callback)); + } let mut topics = self.entity_topics.lock().unwrap(); topics.insert(entity_gid, topic); @@ -180,21 +258,36 @@ impl GraphEventManager { } pub fn unregister_entity(&self, entity_gid: &GidArray) { - let mut callbacks = self.event_callbacks.lock().unwrap(); - callbacks.remove(entity_gid); + // Scoped so the two registries are never held at the same time. + { + let mut callbacks = self.event_callbacks.lock().unwrap(); + callbacks.remove(entity_gid); + } let mut topics = self.entity_topics.lock().unwrap(); topics.remove(entity_gid); } - pub fn register_graph_guard_condition(&self, guard_condition: *mut std::ffi::c_void) { + /// Register a guard condition to be triggered on every graph change. + /// + /// The manager keeps the `Arc` alive for as long as it is registered, and + /// for the duration of any trigger already in flight — see + /// [`GraphGuardCondition`] for why that ownership is load-bearing. + pub fn register_graph_guard_condition(&self, guard_condition: Arc) { let mut conditions = self.graph_guard_conditions.lock().unwrap(); - conditions.push(guard_condition as usize); - } - - pub fn unregister_graph_guard_condition(&self, guard_condition: *mut std::ffi::c_void) { + conditions.push(guard_condition); + } + + /// Stop triggering `guard_condition`. + /// + /// Identity is `Arc::ptr_eq`, so the caller must pass the same allocation it + /// registered. Returning does **not** mean no trigger is in flight: a + /// concurrent [`Self::trigger_graph_change`] may already hold its own clone + /// and be calling into it. That is exactly why the registration is owned — + /// the in-flight call keeps the target alive, so a caller that frees its own + /// handle immediately after this returns is still safe. + pub fn unregister_graph_guard_condition(&self, guard_condition: &Arc) { let mut conditions = self.graph_guard_conditions.lock().unwrap(); - let gc_usize = guard_condition as usize; - conditions.retain(|&gc| gc != gc_usize); + conditions.retain(|gc| !Arc::ptr_eq(gc, guard_condition)); } pub fn trigger_event(&self, entity_gid: &GidArray, event_type: ZenohEventType, change: i32) { @@ -223,11 +316,20 @@ impl GraphEventManager { change }; - let callbacks = self.event_callbacks.lock().unwrap(); - if let Some(entity_callbacks) = callbacks.get(entity_gid) - && let Some(callback) = entity_callbacks.get(&event_type) - { - callback(encoded_change); + // Collect under the lock, invoke after it is released — see [`EventCallback`]. + let callback = { + let callbacks = self.event_callbacks.lock().unwrap(); + callbacks + .get(entity_gid) + .and_then(|entity_callbacks| entity_callbacks.get(&event_type)) + .cloned() + }; + + if let Some(callback) = callback { + crate::invoke_user_callback!( + "GraphEventManager::trigger_event_with_policy", + callback(encoded_change) + ); } } @@ -241,13 +343,17 @@ impl GraphEventManager { let change = if appeared { 1 } else { -1 }; - // Trigger graph guard conditions for ALL graph changes (local and remote) - if let Some(ref trigger) = *self.trigger_guard_condition.lock().unwrap() { - let guard_conditions = self.graph_guard_conditions.lock().unwrap(); - for &gc_usize in guard_conditions.iter() { - let gc = gc_usize as *mut std::ffi::c_void; - trigger(gc); - } + // Trigger graph guard conditions for ALL graph changes (local and remote). + // + // Snapshot, release the lock, then call — the trigger is rmw-side code + // and may re-enter this manager, so it must not run under the guard. + // The snapshot clones `Arc`s rather than raw pointers, which is what + // makes releasing the lock safe: a concurrent `rmw_destroy_node` can + // unregister and free its C handle here, and each in-flight trigger + // still holds the target alive until it returns. + let guard_conditions = self.graph_guard_conditions.lock().unwrap().clone(); + for gc in guard_conditions { + gc.trigger(); } // Determine which event type based on entity kind @@ -268,16 +374,29 @@ impl GraphEventManager { _ => return, }; - let entity_topics = self.entity_topics.lock().unwrap(); - let callbacks = self.event_callbacks.lock().unwrap(); - for (entity_gid, entity_callbacks) in callbacks.iter() { - // Only notify entities on the same topic - if let Some(registered_topic) = entity_topics.get(entity_gid) - && registered_topic == changed_topic - && let Some(callback) = entity_callbacks.get(&event_type) - { - callback(change); - } + // Collect the callbacks to notify, then drop both registry guards before + // invoking any of them — see [`EventCallback`]. Locks are taken in the + // same order as `register_event_callback` (callbacks, then topics). + let to_notify: Vec = { + let callbacks = self.event_callbacks.lock().unwrap(); + let entity_topics = self.entity_topics.lock().unwrap(); + callbacks + .iter() + .filter(|(entity_gid, _)| { + // Only notify entities on the same topic + entity_topics + .get(*entity_gid) + .is_some_and(|registered_topic| registered_topic == changed_topic) + }) + .filter_map(|(_, entity_callbacks)| entity_callbacks.get(&event_type).cloned()) + .collect() + }; + + for callback in to_notify { + crate::invoke_user_callback!( + "GraphEventManager::trigger_graph_change", + callback(change) + ); } } } @@ -310,6 +429,56 @@ impl EventWaitData { } } +/// Record an event-status change on a *shared* [`EventsManager`] and fire its +/// callback with the manager lock released. +/// +/// Holders of an `Arc>` must use this rather than locking +/// and calling [`EventsManager::update_event_status`] directly: that keeps the +/// outer guard alive across the callback, and the callback is user code handed +/// to an rclcpp executor which routinely calls straight back into the same +/// manager (`rmw_take_event` → [`RmEventHandle::take_event`]). +pub fn update_shared_event_status( + events_mgr: &Mutex, + event_type: ZenohEventType, + change: i32, +) { + update_shared_event_status_with_policy(events_mgr, event_type, change, 0) +} + +/// [`update_shared_event_status`] with a QoS policy kind. +/// +/// # Known hazard +/// +/// Releasing the guard also releases the mutual exclusion that used to +/// serialise this against `RmEventHandle::set_callback`, so a concurrent detach +/// can return — and rclcpp can free `user_data` — between the clone below and +/// the call. The `Arc` keeps the *closure* alive; nothing owns the raw C +/// pointer it captured. +/// +/// Do not "fix" this by re-locking (that is the deadlock this removes) or by a +/// flag checked in the closure (the free can land between check and call). +/// Tracked with the design in hiroz#287. +pub fn update_shared_event_status_with_policy( + events_mgr: &Mutex, + event_type: ZenohEventType, + change: i32, + policy_kind: u32, +) { + // The guard is bound to its own `let` inside a block on purpose. Written as + // a `match`/`if let` scrutinee it would stay alive across the invocation + // below and silently reinstate the deadlock this function exists to remove. + let callback = { + let Ok(mut mgr) = events_mgr.lock() else { + return; + }; + mgr.record_event_status_with_policy(event_type, change, policy_kind) + }; + + if let Some(callback) = callback { + callback(change); + } +} + // RMW-style event handle pub struct RmEventHandle { pub events_mgr: Arc>, @@ -349,8 +518,16 @@ impl RmEventHandle { where F: Fn(i32) + Send + Sync + 'static, { - let mut mgr = self.events_mgr.lock().unwrap(); - mgr.set_callback(self.event_type, callback); + let callback: EventCallback = Arc::new(callback); + // Install under the manager lock; fire the backlog notification after it + // is released so the callback may re-enter this handle. + let unread_count = { + let mut mgr = self.events_mgr.lock().unwrap(); + mgr.install_callback(self.event_type, callback.clone()) + }; + if unread_count != 0 { + callback(unread_count); + } } } @@ -567,4 +744,47 @@ mod tests { .update_event_status(ZenohEventType::LivelinessChanged, 3); assert_eq!(*fired.lock().unwrap(), 3); } + + /// The registry must **own** what it registers — the invariant that makes + /// triggering outside the lock safe. See [`GraphGuardCondition`]. + /// + /// Pins the deterministic half: the registry keeps the value alive after + /// the registrant drops its handle, and releases it on unregister. It does + /// not try to schedule the destroy-during-trigger race, which would be a + /// timing test — ownership is what makes that race harmless. + #[test] + fn graph_guard_condition_registration_is_owned_by_the_manager() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + struct CountingGc(Arc); + impl GraphGuardCondition for CountingGc { + fn trigger(&self) { + self.0.fetch_add(1, Ordering::SeqCst); + } + } + + let mgr = GraphEventManager::new(); + let hits = Arc::new(AtomicUsize::new(0)); + let gc: Arc = Arc::new(CountingGc(hits.clone())); + let weak = Arc::downgrade(&gc); + + mgr.register_graph_guard_condition(gc.clone()); + drop(gc); + let held = weak.upgrade().expect( + "the manager must keep the registration alive after the registrant drops its handle; \ + otherwise a trigger issued outside the lock dereferences freed memory", + ); + + // Still reachable and callable through the registry's own reference. + held.trigger(); + assert_eq!(hits.load(Ordering::SeqCst), 1); + + mgr.unregister_graph_guard_condition(&held); + drop(held); + assert!( + weak.upgrade().is_none(), + "unregister must release the registry's reference, or registrations leak for the \ + life of the process", + ); + } } diff --git a/crates/hiroz/src/graph.rs b/crates/hiroz/src/graph.rs index 1edac3b07..1e4d8c577 100644 --- a/crates/hiroz/src/graph.rs +++ b/crates/hiroz/src/graph.rs @@ -600,7 +600,6 @@ impl Graph { .declare_subscriber(&liveliness_pattern) .history(true) .callback(move |sample| { - let mut graph_data_guard = c_graph_data.lock(); let key_expr = sample.key_expr().to_owned(); let ke = LivelinessKE(key_expr.clone()); tracing::debug!( @@ -609,6 +608,11 @@ impl Graph { sample.kind() ); + // `trigger_graph_change` runs user/rmw event callbacks, which routinely + // re-enter the graph (counting publishers, publishing, registering + // entities). `data` is a non-reentrant mutex, so it must never be held + // across that call — hence the tight scopes below. `add_local_entity` + // already follows this rule. match sample.kind() { SampleKind::Put => { debug!("[GRF] Entity appeared: {}", ke.0); @@ -625,26 +629,35 @@ impl Graph { } }; - // Only insert if not already parsed (avoid duplicates from liveliness query) - let already_parsed = graph_data_guard.parsed.contains_key(&ke); - let already_cached = graph_data_guard.cached.contains(&ke); - tracing::debug!( - " Check: parsed={}, cached={}, parsed.len()={}, cached.len()={}", - already_parsed, - already_cached, - graph_data_guard.parsed.len(), - graph_data_guard.cached.len() - ); - if already_parsed { - tracing::debug!(" Skipping - already in parsed"); - } else if already_cached { - tracing::debug!(" Skipping - already in cached"); - } else { - tracing::debug!(" Adding to cached"); - graph_data_guard.insert(ke.clone()); - } + let already_parsed = { + let mut graph_data_guard = c_graph_data.lock(); + + // Only insert if not already parsed (avoid duplicates from + // liveliness query) + let already_parsed = graph_data_guard.parsed.contains_key(&ke); + let already_cached = graph_data_guard.cached.contains(&ke); + tracing::debug!( + " Check: parsed={}, cached={}, parsed.len()={}, cached.len()={}", + already_parsed, + already_cached, + graph_data_guard.parsed.len(), + graph_data_guard.cached.len() + ); + if already_parsed { + tracing::debug!(" Skipping - already in parsed"); + } else if already_cached { + tracing::debug!(" Skipping - already in cached"); + } else { + tracing::debug!(" Adding to cached"); + graph_data_guard.insert(ke.clone()); + } + already_parsed + }; + // Only fire the event for genuinely new entities; if add_local_entity // already inserted and fired for this key, don't fire a second time. + // Fires after the insert, as before — a callback that queries the + // graph sees the new entity. if !already_parsed && let Some(entity) = parsed_entity { tracing::debug!("Successfully parsed entity: {:?}", entity); c_event_manager.trigger_graph_change(&entity, true, c_zid); @@ -655,20 +668,20 @@ impl Graph { SampleKind::Delete => { debug!("[GRF] Entity disappeared: {}", ke.0); tracing::debug!("Graph subscriber: DELETE {}", key_expr.as_str()); - // Trigger graph change events before removal using backend-specific parser + // Trigger graph change events before removal using backend-specific + // parser — a callback still observes the disappearing entity, as + // before. if let Ok(entity) = callback_parser(&key_expr) { c_event_manager.trigger_graph_change(&entity, false, c_zid); } - graph_data_guard.remove(&ke); + c_graph_data.lock().remove(&ke); c_change_notify.notify_waiters(); } } - // Release graph.data before signaling sync waiters. - // Lock ordering: sync waiters acquire change_signal.0 then (briefly) data; - // the callback holds data then acquires change_signal.0 — so we must drop - // data first to ensure the two locks are never held simultaneously. - drop(graph_data_guard); + // graph.data is released above before signaling sync waiters. + // Lock ordering: sync waiters acquire change_signal.0 then (briefly) data, + // so the callback must never hold data while acquiring change_signal.0. c_change_signal.1.notify_all(); }) .wait()?; diff --git a/crates/rmw-zenoh-rs/src/context.rs b/crates/rmw-zenoh-rs/src/context.rs index d294318ec..bf96457e6 100644 --- a/crates/rmw-zenoh-rs/src/context.rs +++ b/crates/rmw-zenoh-rs/src/context.rs @@ -95,18 +95,6 @@ impl ContextImpl { .build() .map_err(|e| format!("Failed to create ZContext: {}", e))?; - // Set up the guard condition trigger function for graph events - // This allows graph changes to trigger RMW guard conditions - let trigger_fn: hiroz::event::GraphGuardConditionTrigger = Box::new(|gc_ptr| { - crate::guard_condition::rmw_trigger_guard_condition( - gc_ptr as *const crate::ros::rmw_guard_condition_t, - ); - }); - zcontext - .graph() - .event_manager - .set_guard_condition_trigger(trigger_fn); - Ok(Self { zcontext: Arc::new(zcontext), enclave, diff --git a/crates/rmw-zenoh-rs/src/guard_condition.rs b/crates/rmw-zenoh-rs/src/guard_condition.rs index 8ae469239..1c5215531 100644 --- a/crates/rmw-zenoh-rs/src/guard_condition.rs +++ b/crates/rmw-zenoh-rs/src/guard_condition.rs @@ -3,30 +3,92 @@ use crate::ros::*; use crate::traits::*; use crate::utils::Notifier; use std::sync::Arc; - -/// Guard condition implementation for RMW +use std::sync::atomic::{AtomicBool, Ordering}; + +/// The triggerable state of a guard condition, separated from the C handle. +/// +/// Behind an `Arc` so it can outlive `rmw_destroy_guard_condition`: hiroz's +/// graph-event manager registers a clone and triggers it **after** dropping its +/// registry lock, so a racing `rmw_destroy_node` would otherwise free the target +/// mid-call. The C handle holds one reference and the registry another. +/// +/// `triggered` is atomic because triggering no longer happens under any lock. #[derive(Debug, Default)] -pub struct GuardConditionImpl { +pub struct GuardConditionState { pub(crate) notifier: Option>, - pub(crate) triggered: bool, + pub(crate) triggered: AtomicBool, } -impl GuardConditionImpl { - pub(crate) fn trigger(&mut self) -> Result<(), ()> { +impl GuardConditionState { + pub(crate) fn fire(&self) -> Result<(), ()> { let notifier = self.notifier.as_ref().ok_or(())?; - self.triggered = true; + self.triggered.store(true, Ordering::SeqCst); notifier.notify_all(); Ok(()) } - pub fn reset(&mut self) { - self.triggered = false; + pub(crate) fn reset(&self) { + self.triggered.store(false, Ordering::SeqCst); + } + + /// Consume the triggered flag, returning what it was. + /// + /// One atomic op, so a `trigger()` racing this either wins or is preserved. + /// `is_triggered()` then `reset()` is two ops and silently drops a trigger + /// landing between them: the waiter reports this wake, and the next + /// `rmw_wait` blocks until timeout. + pub(crate) fn take_triggered(&self) -> bool { + self.triggered.swap(false, Ordering::SeqCst) + } + + pub(crate) fn is_triggered(&self) -> bool { + self.triggered.load(Ordering::SeqCst) + } +} + +impl hiroz::event::GraphGuardCondition for GuardConditionState { + fn trigger(&self) { + // A guard condition with no notifier cannot wake anyone; that is not an + // error worth propagating across the registry. + let _ = self.fire(); + } +} + +/// Guard condition implementation for RMW +#[derive(Debug, Default)] +pub struct GuardConditionImpl { + pub(crate) state: Arc, +} + +impl GuardConditionImpl { + // `&self`, not `&mut self`: `rmw_wait` holds shared references while + // scanning the wait set, and a zenoh thread can trigger concurrently. + // Handing out `&mut` to an object another thread holds a `&` to is UB + // regardless of field types — the atomic defines the *data race* but says + // nothing about the aliasing. All mutation goes through the atomics. + pub(crate) fn trigger(&self) -> Result<(), ()> { + self.state.fire() + } + + pub fn reset(&self) { + self.state.reset(); + } + + /// Consume the triggered flag. See [`GuardConditionState::take_triggered`]. + pub(crate) fn take_triggered(&self) -> bool { + self.state.take_triggered() + } + + /// A shared handle to this guard condition's state, for registration with + /// hiroz's graph-event manager. + pub(crate) fn share_state(&self) -> Arc { + self.state.clone() } } impl crate::traits::Waitable for GuardConditionImpl { fn is_ready(&self) -> bool { - self.triggered + self.state.is_triggered() } } @@ -52,8 +114,10 @@ pub extern "C" fn rmw_create_guard_condition( let notifier = Some(context_impl.share_notifier()); let gc_impl = GuardConditionImpl { - notifier, - triggered: false, + state: Arc::new(GuardConditionState { + notifier, + triggered: AtomicBool::new(false), + }), }; let gc = Box::new(rmw_guard_condition_t { implementation_identifier: crate::RMW_ZENOH_IDENTIFIER.as_ptr() as *const _, @@ -90,7 +154,8 @@ pub extern "C" fn rmw_trigger_guard_condition( return RMW_RET_INVALID_ARGUMENT as _; } - if let Ok(gc_impl) = (guard_condition as *mut rmw_guard_condition_t).borrow_mut_data() { + // Immutable borrow: see the note on `GuardConditionImpl::trigger`. + if let Ok(gc_impl) = (guard_condition as *mut rmw_guard_condition_t).borrow_data() { let _ = gc_impl.trigger(); } diff --git a/crates/rmw-zenoh-rs/src/node.rs b/crates/rmw-zenoh-rs/src/node.rs index 245160ad6..b2f877b2d 100644 --- a/crates/rmw-zenoh-rs/src/node.rs +++ b/crates/rmw-zenoh-rs/src/node.rs @@ -12,6 +12,13 @@ pub struct NodeImpl { pub namespace: CString, pub fq_name: CString, pub graph_guard_condition: *mut rmw_guard_condition_t, + /// The shared state registered with hiroz's graph-event manager. + /// + /// Kept so teardown can unregister the exact allocation it registered + /// (identity is `Arc::ptr_eq`). Holding it here also means the state + /// survives `rmw_destroy_guard_condition` below, which is what makes it safe + /// to free the C handle while a graph-change trigger may still be in flight. + pub graph_guard_condition_state: Option>, } impl NodeImpl { @@ -44,6 +51,7 @@ impl NodeImpl { namespace: namespace_cstr, fq_name: fq_name_cstr, graph_guard_condition: std::ptr::null_mut(), + graph_guard_condition_state: None, }) } } @@ -122,12 +130,23 @@ pub extern "C" fn rmw_create_node( } node_impl.graph_guard_condition = graph_guard_condition; - // Register the graph guard condition with the graph event manager + // Register the graph guard condition with the graph event manager. + // + // Register the *shared state*, not the C pointer: the manager triggers with + // its registry lock released, so a raw pointer could be freed by a + // concurrent `rmw_destroy_node` between snapshot and call. Handing over an + // `Arc` keeps the target alive for the duration of any in-flight trigger. + let gc_state: std::sync::Arc = + match graph_guard_condition.borrow_data() { + Ok(gc_impl) => gc_impl.share_state(), + Err(_) => return std::ptr::null_mut(), + }; + node_impl.graph_guard_condition_state = Some(gc_state.clone()); node_impl .inner .graph() .event_manager - .register_graph_guard_condition(graph_guard_condition as *mut std::ffi::c_void); + .register_graph_guard_condition(gc_state); // Add node to local graph for immediate discovery if let Err(e) = node_impl @@ -175,7 +194,7 @@ pub extern "C" fn rmw_destroy_node(node: *mut rmw_node_t) -> rmw_ret_t { } // Remove node from local graph and destroy the graph guard condition - if let Ok(node_impl) = node.borrow_data() { + if let Ok(node_impl) = node.borrow_mut_data() { // Remove node from local graph if let Err(e) = node_impl .inner @@ -188,14 +207,17 @@ pub extern "C" fn rmw_destroy_node(node: *mut rmw_node_t) -> rmw_ret_t { } if !node_impl.graph_guard_condition.is_null() { - // Unregister from graph event manager - node_impl - .inner - .graph() - .event_manager - .unregister_graph_guard_condition( - node_impl.graph_guard_condition as *mut std::ffi::c_void, - ); + // Unregister from graph event manager, by the same allocation we + // registered. Destroying the C handle immediately afterwards is + // safe even if a graph-change trigger is in flight: that trigger + // holds its own `Arc` to the state, which outlives this handle. + if let Some(gc_state) = node_impl.graph_guard_condition_state.take() { + node_impl + .inner + .graph() + .event_manager + .unregister_graph_guard_condition(&gc_state); + } crate::guard_condition::rmw_destroy_guard_condition(node_impl.graph_guard_condition); } } diff --git a/crates/rmw-zenoh-rs/src/rmw.rs b/crates/rmw-zenoh-rs/src/rmw.rs index 884cb9e6a..fc0408805 100644 --- a/crates/rmw-zenoh-rs/src/rmw.rs +++ b/crates/rmw-zenoh-rs/src/rmw.rs @@ -208,9 +208,11 @@ pub extern "C" fn rmw_create_publisher( qualified_topic.clone(), hiroz::event::ZenohEventType::PublicationMatched, move |change| { - if let Ok(mut mgr) = events_mgr.lock() { - mgr.update_event_status(hiroz::event::ZenohEventType::PublicationMatched, change); - } + hiroz::event::update_shared_event_status( + &events_mgr, + hiroz::event::ZenohEventType::PublicationMatched, + change, + ); // Wake up wait sets notifier_clone_for_matched.notify_all(); }, @@ -230,13 +232,12 @@ pub extern "C" fn rmw_create_publisher( // Decode policy_kind from upper 16 bits and change from lower 16 bits let policy_kind = ((encoded_change >> 16) & 0xFFFF) as u32; let change = encoded_change & 0xFFFF; - if let Ok(mut mgr) = events_mgr_clone.lock() { - mgr.update_event_status_with_policy( - hiroz::event::ZenohEventType::OfferedQosIncompatible, - change, - policy_kind, - ); - } + hiroz::event::update_shared_event_status_with_policy( + &events_mgr_clone, + hiroz::event::ZenohEventType::OfferedQosIncompatible, + change, + policy_kind, + ); // Wake up wait sets notifier_clone_for_incompatible.notify_all(); }, @@ -250,12 +251,11 @@ pub extern "C" fn rmw_create_publisher( // Check if there are already existing subscriptions for this topic and trigger the event let matching_sub_count = graph.count(hiroz::entity::EndpointKind::Subscription, &entity.topic); if matching_sub_count > 0 { - if let Ok(mut mgr) = zpub.events_mgr().lock() { - mgr.update_event_status( - hiroz::event::ZenohEventType::PublicationMatched, - matching_sub_count as i32, - ); - } + hiroz::event::update_shared_event_status( + zpub.events_mgr(), + hiroz::event::ZenohEventType::PublicationMatched, + matching_sub_count as i32, + ); notifier_clone_for_init.notify_all(); } @@ -307,13 +307,12 @@ pub extern "C" fn rmw_create_publisher( } } if incompatible_count > 0 { - if let Ok(mut mgr) = zpub.events_mgr().lock() { - mgr.update_event_status_with_policy( - hiroz::event::ZenohEventType::OfferedQosIncompatible, - incompatible_count, - last_policy_kind, - ); - } + hiroz::event::update_shared_event_status_with_policy( + zpub.events_mgr(), + hiroz::event::ZenohEventType::OfferedQosIncompatible, + incompatible_count, + last_policy_kind, + ); notifier_clone_for_init.notify_all(); } @@ -595,9 +594,11 @@ pub extern "C" fn rmw_create_subscription( sub_topic.clone(), hiroz::event::ZenohEventType::SubscriptionMatched, move |change| { - if let Ok(mut mgr) = events_mgr.lock() { - mgr.update_event_status(hiroz::event::ZenohEventType::SubscriptionMatched, change); - } + hiroz::event::update_shared_event_status( + &events_mgr, + hiroz::event::ZenohEventType::SubscriptionMatched, + change, + ); // Wake up wait sets notifier_clone_for_matched.notify_all(); }, @@ -617,13 +618,12 @@ pub extern "C" fn rmw_create_subscription( // Decode policy_kind from upper 16 bits and change from lower 16 bits let policy_kind = ((encoded_change >> 16) & 0xFFFF) as u32; let change = encoded_change & 0xFFFF; - if let Ok(mut mgr) = events_mgr_clone.lock() { - mgr.update_event_status_with_policy( - hiroz::event::ZenohEventType::RequestedQosIncompatible, - change, - policy_kind, - ); - } + hiroz::event::update_shared_event_status_with_policy( + &events_mgr_clone, + hiroz::event::ZenohEventType::RequestedQosIncompatible, + change, + policy_kind, + ); // Wake up wait sets notifier_clone_for_incompatible.notify_all(); }, @@ -637,12 +637,11 @@ pub extern "C" fn rmw_create_subscription( // Check if there are already existing publishers for this topic and trigger the event let matching_pub_count = graph.count(hiroz::entity::EndpointKind::Publisher, &entity.topic); if matching_pub_count > 0 { - if let Ok(mut mgr) = zsub.events_mgr().lock() { - mgr.update_event_status( - hiroz::event::ZenohEventType::SubscriptionMatched, - matching_pub_count as i32, - ); - } + hiroz::event::update_shared_event_status( + zsub.events_mgr(), + hiroz::event::ZenohEventType::SubscriptionMatched, + matching_pub_count as i32, + ); notifier_clone_for_init.notify_all(); } @@ -694,13 +693,12 @@ pub extern "C" fn rmw_create_subscription( } } if incompatible_count > 0 { - if let Ok(mut mgr) = zsub.events_mgr().lock() { - mgr.update_event_status_with_policy( - hiroz::event::ZenohEventType::RequestedQosIncompatible, - incompatible_count, - last_policy_kind, - ); - } + hiroz::event::update_shared_event_status_with_policy( + zsub.events_mgr(), + hiroz::event::ZenohEventType::RequestedQosIncompatible, + incompatible_count, + last_policy_kind, + ); notifier_clone_for_init.notify_all(); } @@ -2406,6 +2404,39 @@ pub extern "C" fn rmw_subscription_event_init( RMW_RET_OK as _ } +/// The `user_data` pointer rmw handed us, carried into the event callback. +/// +/// The callback is `Fn(i32) + Send + Sync` and a raw pointer is neither, so +/// capturing `user_data` directly does not compile. The previous `as usize` +/// round-trip made it compile by silencing exactly the check that was flagging +/// the hazard, and destroyed the pointer's provenance with it. This asserts the +/// same thing explicitly, once. +/// +/// The `unsafe impl`s below assert only that the pointer may be *moved* between +/// threads. They say nothing about how long the pointee lives — see the known +/// hazard on `update_shared_event_status_with_policy`. +struct EventUserData(*mut c_void); + +impl EventUserData { + /// Hand the pointer back in the form the C callback expects. + /// + /// A method, not a field read: closures capture the most precise path they + /// use (RFC 2229), so `move |..| { ud.0 }` captures the bare pointer and + /// the newtype's `Send`/`Sync` never apply. `&self` forces whole-struct + /// capture. + fn as_ptr(&self) -> *const ::std::os::raw::c_void { + self.0 as *const ::std::os::raw::c_void + } +} + +// SAFETY: the pointer is opaque to hiroz — it is never dereferenced here, only +// handed back to the C callback that supplied it. rmw's contract is that the +// callback may be invoked from any thread, so the caller has already accepted +// that `user_data` is reachable from other threads. +unsafe impl Send for EventUserData {} +// SAFETY: as above. `&EventUserData` exposes no operation on the pointee. +unsafe impl Sync for EventUserData {} + #[unsafe(no_mangle)] pub extern "C" fn rmw_event_set_callback( event: *mut rmw_event_t, @@ -2421,11 +2452,13 @@ pub extern "C" fn rmw_event_set_callback( } let rm_event_handle = unsafe { &mut *((*event).data as *mut RmEventHandle) }; - let user_data_ptr = user_data as usize; + let user_data = EventUserData(user_data); rm_event_handle.set_callback(move |change: i32| { if let Some(cb) = callback { - let ud = user_data_ptr as *mut ::std::os::raw::c_void; - unsafe { cb(ud as *const ::std::os::raw::c_void, change as usize) }; + // SAFETY: `cb` and the pointer were supplied together by rmw and + // this is the pair's only use. Pointee validity is NOT established + // here — a concurrent detach can free it first. See hiroz#287. + unsafe { cb(user_data.as_ptr(), change as usize) }; } }); diff --git a/crates/rmw-zenoh-rs/src/wait_set.rs b/crates/rmw-zenoh-rs/src/wait_set.rs index f9b0ab4aa..90685fbf4 100644 --- a/crates/rmw-zenoh-rs/src/wait_set.rs +++ b/crates/rmw-zenoh-rs/src/wait_set.rs @@ -388,15 +388,15 @@ pub extern "C" fn rmw_wait( unsafe { *gc_array.guard_conditions.add(i) as *mut rmw_guard_condition_impl_t }; if !gc_impl_ptr.is_null() { unsafe { + // Shared, not exclusive: a delivery thread may be + // inside `trigger` on this same object right now. let gc_impl = - &mut *(gc_impl_ptr as *mut crate::guard_condition::GuardConditionImpl); - if !gc_impl.is_ready() { - // Not ready - set to NULL in place + &*(gc_impl_ptr as *const crate::guard_condition::GuardConditionImpl); + // Check and consume in one atomic op. Reading then + // resetting would drop a trigger landing between the + // two: this wake is reported, the next one is lost. + if !gc_impl.take_triggered() { *gc_array.guard_conditions.add(i) = std::ptr::null_mut(); - } else { - // Reset the guard condition after it's been detected as ready - // This prevents it from staying triggered forever - gc_impl.reset(); } } }