From 59123f5325f6095defbd8903f6e99146de441fbe Mon Sep 17 00:00:00 2001 From: Mathieu David Date: Sun, 21 Jun 2026 01:54:23 +0200 Subject: [PATCH 1/3] feat: add push-callback registration for rcl primitives Add an opt-in way for primitives to report readiness via rcl's push callbacks (rcl_*_set_on_new_*_callback) instead of being polled in a wait set, as the foundation for an event-driven executor. `RclPrimitive::register_on_ready` installs a callback that the middleware invokes when the entity becomes ready and returns an `OnReadyHandle` (RAII) that deregisters on drop. `OnReadyRegistration` wraps the unsafe rcl setter: it boxes the callback context for a stable address and, on drop, clears the callback before freeing the context (finalizing the rcl entity first) so the middleware can never invoke a freed context during teardown. Implemented for subscriptions, services, and clients. No executor consumes this yet, so the basic executor is unchanged. --- rclrs/src/client.rs | 25 +++ rclrs/src/executor.rs | 2 + rclrs/src/executor/event_callback.rs | 243 +++++++++++++++++++++++++++ rclrs/src/service.rs | 25 +++ rclrs/src/subscription.rs | 35 ++++ rclrs/src/wait_set/rcl_primitive.rs | 30 ++++ 6 files changed, 360 insertions(+) create mode 100644 rclrs/src/executor/event_callback.rs diff --git a/rclrs/src/client.rs b/rclrs/src/client.rs index a229bac13..1535dc7ab 100644 --- a/rclrs/src/client.rs +++ b/rclrs/src/client.rs @@ -468,6 +468,31 @@ where fn kind(&self) -> RclPrimitiveKind { RclPrimitiveKind::Client } + + fn register_on_ready( + &self, + on_ready: Box, + ) -> Result>, RclrsError> { + // A client has a single readiness path; report it as `Basic`. + let on_ready = move |n| on_ready(ReadyKind::Basic, n); + let registration = crate::executor::event_callback::OnReadyRegistration::new( + Arc::clone(&self.handle), + set_client_on_new_response, + Box::new(on_ready), + )?; + Ok(Some(Box::new(registration))) + } +} + +/// Install (or, with a null callback/user_data, clear) the "on new response" +/// push callback used by the event-driven executor. Encapsulates the client +/// lock and the rcl call within this module. +unsafe fn set_client_on_new_response( + handle: &ClientHandle, + callback: rcl_event_callback_t, + user_data: *const std::os::raw::c_void, +) -> rcl_ret_t { + rcl_client_set_on_new_response_callback(&*handle.lock(), callback, user_data) } type SequenceNumber = i64; diff --git a/rclrs/src/executor.rs b/rclrs/src/executor.rs index 0d97f985f..10aae3cf1 100644 --- a/rclrs/src/executor.rs +++ b/rclrs/src/executor.rs @@ -1,6 +1,8 @@ mod basic_executor; pub use self::basic_executor::*; +pub(crate) mod event_callback; + use crate::{ Context, ContextHandle, GuardCondition, IntoNodeOptions, Node, RclrsError, Waitable, WeakActivityListener, diff --git a/rclrs/src/executor/event_callback.rs b/rclrs/src/executor/event_callback.rs new file mode 100644 index 000000000..49b08f2ec --- /dev/null +++ b/rclrs/src/executor/event_callback.rs @@ -0,0 +1,243 @@ +//! Safe RAII wrapper around rcl's "on new ___" push-callback APIs +//! (`rcl_subscription_set_on_new_message_callback` and the service/client/event +//! equivalents). +//! +//! These let an event-driven executor learn that an entity has become ready +//! *without* polling `rcl_wait`: the middleware invokes a C callback (possibly +//! from its own thread) when data arrives. We forward that to a Rust closure, +//! which an executor uses to enqueue work. +//! +//! [`OnReadyRegistration`] is generic over the entity handle type `H`. Each +//! entity module provides a [`SetOnReadyFn`] that locks its handle and calls the +//! appropriate `rcl_*_set_on_new_*_callback`; the registration owns the boxed +//! callback context and the handle, and deregisters on drop. +//! +//! # Safety model +//! +//! rcl stores the `user_data` pointer we hand it and passes it back to the C +//! callback on every event. That pointer must stay valid for as long as the +//! callback is registered. We therefore: +//! +//! - box the [`EventCallbackCtx`] so it has a stable heap address, and +//! - in `Drop`, **unregister the callback first** (so the middleware can no +//! longer invoke the trampoline) and only then free the context. +//! +//! Getting that ordering wrong is a use-after-free, since the middleware may be +//! calling the trampoline from another thread at the moment of teardown. + +use std::{os::raw::c_void, sync::Arc}; + +use crate::{rcl_bindings::*, OnReadyHandle, RclrsError, ToResult}; + +/// The context carried through rcl as `user_data`. Boxed so its address is +/// stable for the lifetime of the registration. +struct EventCallbackCtx { + on_ready: Box, +} + +/// The C trampoline that rcl/rmw invokes when an entity becomes ready. It may be +/// called from a middleware thread, so it does nothing but forward to the Rust +/// closure. It must not run user code or take locks that could deadlock the +/// middleware. +unsafe extern "C" fn on_ready_trampoline(user_data: *const c_void, number_of_events: usize) { + // SAFETY: `user_data` is the pointer to the `EventCallbackCtx` we passed to + // the rcl setter. It stays valid until the owning registration's `Drop` + // clears the callback, which always happens before the box is freed. + let ctx = unsafe { &*(user_data as *const EventCallbackCtx) }; + (ctx.on_ready)(number_of_events); +} + +/// A function that registers (or, with a null callback/user_data, clears) the +/// "on ready" push callback on an entity handle of type `H`. Implemented per +/// entity module so that `H`'s (private) lock and its specific +/// `rcl_*_set_on_new_*_callback` stay encapsulated there. +pub(crate) type SetOnReadyFn = unsafe fn(&H, rcl_event_callback_t, *const c_void) -> rcl_ret_t; + +/// RAII registration of a push "on ready" callback on an rcl entity. +/// +/// While alive, `on_ready(number_of_events)` is invoked by the middleware +/// whenever the entity becomes ready. Dropping it unregisters the callback +/// before releasing the context, so the middleware can never call into freed +/// memory. +pub(crate) struct OnReadyRegistration { + set_callback: SetOnReadyFn, + // Field order is important for teardown safety: `handle` is declared + // (and therefore dropped) before `ctx`. Dropping the last `Arc` + // finalizes the rcl entity (destroying the middleware reader), so by the + // time `ctx` is freed no middleware thread can still invoke the trampoline + // against it. This mirrors rclcpp, which frees its callback storage only + // after `rcl_*_fini`. See `Drop` below. + handle: Arc, + + // Never read directly, held only so its `Drop` frees the callback context. + #[allow(dead_code)] + ctx: CtxBox, +} + +impl OnReadyHandle for OnReadyRegistration {} + +impl OnReadyRegistration { + /// Register `on_ready` to be called by the middleware whenever the entity + /// becomes ready. `set_callback` locks `handle` and installs the trampoline. + pub(crate) fn new( + handle: Arc, + set_callback: SetOnReadyFn, + on_ready: Box, + ) -> Result { + let ctx = Box::into_raw(Box::new(EventCallbackCtx { on_ready })); + + // SAFETY: `ctx` points to a live, heap-stable context that outlives the + // registration (only freed once, when the `CtxBox` field is dropped). + let result = + unsafe { set_callback(&handle, Some(on_ready_trampoline), ctx as *const c_void).ok() }; + + if let Err(err) = result { + // Registration failed, so nothing else references `ctx`. Reclaim it. + // SAFETY: `ctx` came from `Box::into_raw` above and was never + // successfully registered. + unsafe { + drop(Box::from_raw(ctx)); + } + return Err(err); + } + + Ok(Self { + set_callback, + handle, + ctx: CtxBox(ctx), + }) + } +} + +impl Drop for OnReadyRegistration { + fn drop(&mut self) { + // Detach the callback so the middleware stops invoking the trampoline. + // The context is NOT freed here, the `ctx: CtxBox` field is dropped + // *after* `handle`, so the rcl entity is finalized before + // the context is freed, avoiding a use-after-free if a callback is still + // in flight at teardown. + // + // SAFETY: handle is valid and locked by the setter. A null callback + + // null user_data clears the registration. + unsafe { + let _ = (self.set_callback)(&self.handle, None, std::ptr::null()); + } + } +} + +/// Owns the heap-allocated [`EventCallbackCtx`] and frees it on drop. Kept as a +/// separate field of [`OnReadyRegistration`] so its drop runs *after* the +/// `handle` Arc. +struct CtxBox(*mut EventCallbackCtx); + +// SAFETY: the pointer is only dereferenced by the middleware via the trampoline +// (forwarding to a `Send + Sync` closure). It carries no thread-unsafe state. +unsafe impl Send for CtxBox {} +unsafe impl Sync for CtxBox {} + +impl Drop for CtxBox { + fn drop(&mut self) { + // SAFETY: by the time this runs, `OnReadyRegistration::drop` has cleared + // the callback and the `handle` field has been dropped (finalizing the + // rcl entity if it was the last reference), so no middleware thread can + // still be dereferencing this context. Reclaim it exactly once. + unsafe { + drop(Box::from_raw(self.0)); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{subscription::set_subscription_on_new_message, *}; + use ros_env::test_msgs::msg; + use std::{ + sync::atomic::{AtomicUsize, Ordering}, + time::{Duration, Instant}, + }; + + /// The push callback fires when messages arrive, without ever spinning the + /// executor (i.e. without `rcl_wait`). + #[test] + fn push_callback_fires_without_spinning() -> Result<(), RclrsError> { + let executor = Context::default().create_basic_executor(); + let node = executor.create_node(&format!("test_push_callback_{}", line!()))?; + let qos = QoSProfile::default().reliable().keep_last(10); + + let publisher = node.create_publisher::("test_push_topic".qos(qos))?; + let subscription = node + .create_subscription::("test_push_topic".qos(qos), |_: msg::Empty| {})?; + + let count = Arc::new(AtomicUsize::new(0)); + let count_cb = Arc::clone(&count); + let _registration = OnReadyRegistration::new( + Arc::clone(subscription.handle()), + set_subscription_on_new_message, + Box::new(move |n| { + count_cb.fetch_add(n, Ordering::Relaxed); + }), + )?; + + // Publish repeatedly (to ride out discovery) and wait for the push + // callback to fire. We deliberately never spin the executor. + let deadline = Instant::now() + Duration::from_secs(10); + while count.load(Ordering::Relaxed) == 0 && Instant::now() < deadline { + publisher.publish(msg::Empty::default())?; + std::thread::sleep(Duration::from_millis(20)); + } + + assert!( + count.load(Ordering::Relaxed) > 0, + "push callback never fired" + ); + Ok(()) + } + + /// Rapidly create and drop registrations while messages are flowing. If the + /// drop ordering is wrong (freeing the context before unregistering), the + /// middleware thread can call into freed memory; this stresses that path. + #[test] + fn rapid_register_unregister_is_sound() -> Result<(), RclrsError> { + let executor = Context::default().create_basic_executor(); + let node = executor.create_node(&format!("test_push_raii_{}", line!()))?; + let qos = QoSProfile::default().reliable().keep_last(10); + + let publisher = node.create_publisher::("test_push_raii_topic".qos(qos))?; + let subscription = node.create_subscription::( + "test_push_raii_topic".qos(qos), + |_: msg::Empty| {}, + )?; + + // A background thread floods the topic the whole time. + let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let stop_pub = Arc::clone(&stop); + let flood = std::thread::spawn(move || { + while !stop_pub.load(Ordering::Acquire) { + let _ = publisher.publish(msg::Empty::default()); + std::thread::sleep(Duration::from_micros(50)); + } + }); + + // Register/unregister many times against the live subscription. + for _ in 0..2000 { + let count = Arc::new(AtomicUsize::new(0)); + let count_cb = Arc::clone(&count); + let registration = OnReadyRegistration::new( + Arc::clone(subscription.handle()), + set_subscription_on_new_message, + Box::new(move |n| { + count_cb.fetch_add(n, Ordering::Relaxed); + }), + )?; + // Hold briefly so the middleware can fire into this context, then drop + // (which must unregister before freeing). + std::thread::sleep(Duration::from_micros(100)); + drop(registration); + } + + stop.store(true, Ordering::Release); + flood.join().unwrap(); + Ok(()) + } +} diff --git a/rclrs/src/service.rs b/rclrs/src/service.rs index a30ec8dc4..f72accb25 100644 --- a/rclrs/src/service.rs +++ b/rclrs/src/service.rs @@ -307,6 +307,31 @@ where fn handle(&self) -> RclPrimitiveHandle<'_> { RclPrimitiveHandle::Service(self.handle.lock()) } + + fn register_on_ready( + &self, + on_ready: Box, + ) -> Result>, RclrsError> { + // A service has a single readiness path; report it as `Basic`. + let on_ready = move |n| on_ready(ReadyKind::Basic, n); + let registration = crate::executor::event_callback::OnReadyRegistration::new( + Arc::clone(&self.handle), + set_service_on_new_request, + Box::new(on_ready), + )?; + Ok(Some(Box::new(registration))) + } +} + +/// Install (or, with a null callback/user_data, clear) the "on new request" +/// push callback used by the event-driven executor. Encapsulates the service +/// lock and the rcl call within this module. +pub(crate) unsafe fn set_service_on_new_request( + handle: &ServiceHandle, + callback: rcl_event_callback_t, + user_data: *const std::os::raw::c_void, +) -> rcl_ret_t { + rcl_service_set_on_new_request_callback(&*handle.lock(), callback, user_data) } // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread diff --git a/rclrs/src/subscription.rs b/rclrs/src/subscription.rs index 16a800a9c..e58767788 100644 --- a/rclrs/src/subscription.rs +++ b/rclrs/src/subscription.rs @@ -103,6 +103,16 @@ where self.handle.topic_name() } + /// Access the handle for this subscription's underlying `rcl_subscription_t`. + /// + /// Returns the subscription handle. Only the `event_callback` tests use this + /// accessor (production code reaches the handle through its own field), so it + /// is compiled only under test rather than carried as dead code. + #[cfg(test)] + pub(crate) fn handle(&self) -> &Arc { + &self.handle + } + /// Returns the QoS settings of the subscription. pub fn qos(&self) -> QoSProfile { let options = unsafe { @@ -294,6 +304,31 @@ where fn handle(&self) -> RclPrimitiveHandle<'_> { RclPrimitiveHandle::Subscription(self.handle.lock()) } + + fn register_on_ready( + &self, + on_ready: Box, + ) -> Result>, RclrsError> { + // A subscription has a single readiness path; report it as `Basic`. + let on_ready = move |n| on_ready(ReadyKind::Basic, n); + let registration = crate::executor::event_callback::OnReadyRegistration::new( + Arc::clone(&self.handle), + set_subscription_on_new_message, + Box::new(on_ready), + )?; + Ok(Some(Box::new(registration))) + } +} + +/// Install (or, with a null callback/user_data, clear) the "on new message" +/// push callback used by the event-driven executor. Encapsulates the +/// subscription lock and the rcl call within this module. +pub(crate) unsafe fn set_subscription_on_new_message( + handle: &SubscriptionHandle, + callback: rcl_event_callback_t, + user_data: *const std::os::raw::c_void, +) -> rcl_ret_t { + rcl_subscription_set_on_new_message_callback(&*handle.lock(), callback, user_data) } // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread diff --git a/rclrs/src/wait_set/rcl_primitive.rs b/rclrs/src/wait_set/rcl_primitive.rs index c59efc54c..27f2af4e7 100644 --- a/rclrs/src/wait_set/rcl_primitive.rs +++ b/rclrs/src/wait_set/rcl_primitive.rs @@ -28,8 +28,38 @@ pub trait RclPrimitive: Send + Sync { /// Provide the handle for this primitive fn handle(&self) -> RclPrimitiveHandle<'_>; + + /// Register a push "on ready" callback so an event-driven executor can learn + /// this primitive has become ready without polling a wait set. The + /// middleware invokes `on_ready` with the [`ReadyKind`] describing *which* + /// part of the primitive became ready and the number of new events. + /// + /// Most primitives have a single readiness path and call `on_ready` with + /// [`ReadyKind::Basic`]. Composite primitives (action servers and clients) + /// register one callback per internal source and call `on_ready` with a + /// [`ReadyKind::ActionServer`]/[`ReadyKind::ActionClient`] value whose single + /// matching flag is set, so the executor knows which sub-entity to run. + /// + /// Returns `Ok(None)` for primitive kinds that have no rcl push-callback API + /// (e.g. timers and guard conditions); an event-driven executor drives those + /// by other means. The returned [`OnReadyHandle`] keeps the callback(s) + /// registered; dropping it detaches them. + fn register_on_ready( + &self, + on_ready: Box, + ) -> Result>, RclrsError> { + // Default: no push-callback support. Suppress the unused parameter. + let _ = on_ready; + Ok(None) + } } +/// RAII handle that keeps a push "on ready" callback registered with the +/// middleware (see [`RclPrimitive::register_on_ready`]). Dropping it +/// unregisters the callback, before freeing the callback's context, so an +/// event-driven executor can detach an entity simply by dropping this handle. +pub trait OnReadyHandle: Send + Sync {} + /// Enum to describe the kind of an executable. #[derive(Clone, Copy, Hash, PartialEq, Eq, PartialOrd, Ord)] pub enum RclPrimitiveKind { From befda483e8f8ba269ffda0703477ad12f5616277 Mon Sep 17 00:00:00 2001 From: Mathieu David Date: Sun, 12 Jul 2026 01:14:25 +0200 Subject: [PATCH 2/3] fix: contain panic in push on-ready C trampoline Wrap user closures in catch_unwind so a panic cannot unwind across the extern "C" boundary. Disable the registration after the first panic, log once, and expose an optional hook for executors to surface the error. --- rclrs/src/executor/event_callback.rs | 150 +++++++++++++++++++++++++-- 1 file changed, 144 insertions(+), 6 deletions(-) diff --git a/rclrs/src/executor/event_callback.rs b/rclrs/src/executor/event_callback.rs index 49b08f2ec..3b76de9ca 100644 --- a/rclrs/src/executor/event_callback.rs +++ b/rclrs/src/executor/event_callback.rs @@ -24,27 +24,92 @@ //! //! Getting that ordering wrong is a use-after-free, since the middleware may be //! calling the trampoline from another thread at the moment of teardown. +//! +//! # Panic containment +//! +//! The trampoline is an `extern "C"` entry point. A Rust panic must not unwind +//! across that boundary. User closures are therefore invoked inside +//! [`std::panic::catch_unwind`]; on panic the registration is disabled and a +//! fatal log is emitted once. An optional [`OnReadyPanicHook`] lets an executor +//! record the failure and wake its spin driver without panicking or allocating +//! in the reporter itself. -use std::{os::raw::c_void, sync::Arc}; +use std::{ + os::raw::c_void, + panic::AssertUnwindSafe, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, +}; -use crate::{rcl_bindings::*, OnReadyHandle, RclrsError, ToResult}; +use crate::{log_fatal, rcl_bindings::*, OnReadyHandle, RclrsError, ToResult}; + +/// Optional hook invoked once when a push callback panics. Must not panic or +/// allocate unboundedly; an executor uses this to record a fatal spin error and +/// wake its driver. +pub(crate) type OnReadyPanicHook = Arc; /// The context carried through rcl as `user_data`. Boxed so its address is /// stable for the lifetime of the registration. struct EventCallbackCtx { + /// When false, the trampoline returns without invoking the closure. + enabled: AtomicBool, + /// Set after the first panic so we log and notify at most once. + panic_reported: AtomicBool, on_ready: Box, + on_panic: Option, +} + +impl EventCallbackCtx { + fn new(on_ready: Box, on_panic: Option) -> Self { + Self { + enabled: AtomicBool::new(true), + panic_reported: AtomicBool::new(false), + on_ready, + on_panic, + } + } + + /// Record the first panic, disable this registration, and notify an executor + /// if one was wired in. Must not panic or allocate unboundedly. + fn report_panic(&self) { + if self.panic_reported.swap(true, Ordering::AcqRel) { + return; + } + self.enabled.store(false, Ordering::Release); + log_fatal!( + "rclrs.executor.event_callback", + "A push on-ready callback panicked; the registration has been disabled \ + and will not be invoked again.", + ); + if let Some(on_panic) = &self.on_panic { + on_panic(); + } + } } /// The C trampoline that rcl/rmw invokes when an entity becomes ready. It may be /// called from a middleware thread, so it does nothing but forward to the Rust /// closure. It must not run user code or take locks that could deadlock the -/// middleware. +/// middleware, and it must not let a Rust panic unwind across the C ABI. unsafe extern "C" fn on_ready_trampoline(user_data: *const c_void, number_of_events: usize) { // SAFETY: `user_data` is the pointer to the `EventCallbackCtx` we passed to // the rcl setter. It stays valid until the owning registration's `Drop` // clears the callback, which always happens before the box is freed. let ctx = unsafe { &*(user_data as *const EventCallbackCtx) }; - (ctx.on_ready)(number_of_events); + if !ctx.enabled.load(Ordering::Acquire) { + return; + } + + let panicked = std::panic::catch_unwind(AssertUnwindSafe(|| { + (ctx.on_ready)(number_of_events); + })) + .is_err(); + + if panicked { + ctx.report_panic(); + } } /// A function that registers (or, with a null callback/user_data, clears) the @@ -84,7 +149,18 @@ impl OnReadyRegistration { set_callback: SetOnReadyFn, on_ready: Box, ) -> Result { - let ctx = Box::into_raw(Box::new(EventCallbackCtx { on_ready })); + Self::new_with_panic_hook(handle, set_callback, on_ready, None) + } + + /// Like [`Self::new`], but invokes `on_panic` once if the user closure + /// panics so an executor can surface the failure to `spin()`. + pub(crate) fn new_with_panic_hook( + handle: Arc, + set_callback: SetOnReadyFn, + on_ready: Box, + on_panic: Option, + ) -> Result { + let ctx = Box::into_raw(Box::new(EventCallbackCtx::new(on_ready, on_panic))); // SAFETY: `ctx` points to a live, heap-stable context that outlives the // registration (only freed once, when the `CtxBox` field is dropped). @@ -153,7 +229,7 @@ mod tests { use crate::{subscription::set_subscription_on_new_message, *}; use ros_env::test_msgs::msg; use std::{ - sync::atomic::{AtomicUsize, Ordering}, + sync::atomic::{AtomicBool, AtomicUsize, Ordering}, time::{Duration, Instant}, }; @@ -240,4 +316,66 @@ mod tests { flood.join().unwrap(); Ok(()) } + + /// A panicking push callback must not unwind across the C trampoline or + /// wedge later invocations. After the first panic the registration is + /// disabled and the optional hook fires once. + #[test] + fn push_callback_panic_is_contained() -> Result<(), RclrsError> { + let executor = Context::default().create_basic_executor(); + let node = executor.create_node(&format!("test_push_panic_{}", line!()))?; + let qos = QoSProfile::default().reliable().keep_last(10); + + let publisher = node.create_publisher::("test_push_panic_topic".qos(qos))?; + let subscription = node.create_subscription::( + "test_push_panic_topic".qos(qos), + |_: msg::Empty| {}, + )?; + + let invocations = Arc::new(AtomicUsize::new(0)); + let invocations_cb = Arc::clone(&invocations); + let hook_fired = Arc::new(AtomicBool::new(false)); + let hook_fired_cb = Arc::clone(&hook_fired); + let _registration = OnReadyRegistration::new_with_panic_hook( + Arc::clone(subscription.handle()), + set_subscription_on_new_message, + Box::new(move |_| { + invocations_cb.fetch_add(1, Ordering::Relaxed); + panic!("push callback panic"); + }), + Some(Arc::new(move || { + hook_fired_cb.store(true, Ordering::Release); + })), + )?; + + let deadline = Instant::now() + Duration::from_secs(10); + while invocations.load(Ordering::Relaxed) == 0 && Instant::now() < deadline { + publisher.publish(msg::Empty::default())?; + std::thread::sleep(Duration::from_millis(20)); + } + + assert_eq!( + invocations.load(Ordering::Relaxed), + 1, + "expected exactly one invocation before the registration was disabled", + ); + assert!( + hook_fired.load(Ordering::Acquire), + "panic hook should fire once", + ); + + let invocations_after = invocations.load(Ordering::Relaxed); + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + publisher.publish(msg::Empty::default())?; + std::thread::sleep(Duration::from_millis(20)); + } + + assert_eq!( + invocations.load(Ordering::Relaxed), + invocations_after, + "registration should stay disabled after panic", + ); + Ok(()) + } } From 44b741e71da460c44b0bee490dd8a7ce17556f6a Mon Sep 17 00:00:00 2001 From: Mathieu David Date: Mon, 13 Jul 2026 23:17:59 +0200 Subject: [PATCH 3/3] fix: drop of old handle clears new callback on re-registration --- rclrs/src/client.rs | 17 +- .../dynamic_message/dynamic_subscription.rs | 3 +- rclrs/src/error.rs | 18 ++ rclrs/src/executor/event_callback.rs | 197 +++++++++++++++++- rclrs/src/service.rs | 17 +- rclrs/src/subscription.rs | 17 +- 6 files changed, 262 insertions(+), 7 deletions(-) diff --git a/rclrs/src/client.rs b/rclrs/src/client.rs index 1535dc7ab..bd0dd8b1f 100644 --- a/rclrs/src/client.rs +++ b/rclrs/src/client.rs @@ -2,7 +2,7 @@ use std::{ any::Any, collections::HashMap, ffi::{CStr, CString}, - sync::{Arc, Mutex, MutexGuard}, + sync::{atomic::AtomicBool, Arc, Mutex, MutexGuard}, }; use rosidl_runtime_rs::Message; @@ -354,6 +354,7 @@ where let handle = Arc::new(ClientHandle { rcl_client: Mutex::new(rcl_client), node: Arc::clone(&node), + on_ready_slot: AtomicBool::new(false), }); let board = Arc::new(Mutex::new(ClientRequestBoard::new())); @@ -478,6 +479,7 @@ where let registration = crate::executor::event_callback::OnReadyRegistration::new( Arc::clone(&self.handle), set_client_on_new_response, + client_on_ready_slot, Box::new(on_ready), )?; Ok(Some(Box::new(registration))) @@ -495,6 +497,13 @@ unsafe fn set_client_on_new_response( rcl_client_set_on_new_response_callback(&*handle.lock(), callback, user_data) } +/// The guard for the client's single "on new response" callback slot, paired +/// with [`set_client_on_new_response`] so a push registration can enforce that +/// only one callback owns the slot at a time. +fn client_on_ready_slot(handle: &ClientHandle) -> &AtomicBool { + &handle.on_ready_slot +} + type SequenceNumber = i64; /// This is used internally to monitor the state of active requests, as well as @@ -606,6 +615,12 @@ struct ClientHandle { /// We store the whole node here because we use some of its user-facing API /// in some of the Client methods. node: Node, + /// Guards the single rcl "on new response" callback slot so only one push + /// registration can own it at a time. Selected via [`client_on_ready_slot`]; + /// see [`OnReadySlotFn`]. + /// + /// [`OnReadySlotFn`]: crate::executor::event_callback::OnReadySlotFn + on_ready_slot: AtomicBool, } impl ClientHandle { diff --git a/rclrs/src/dynamic_message/dynamic_subscription.rs b/rclrs/src/dynamic_message/dynamic_subscription.rs index ddc3de290..41ee750fc 100644 --- a/rclrs/src/dynamic_message/dynamic_subscription.rs +++ b/rclrs/src/dynamic_message/dynamic_subscription.rs @@ -3,7 +3,7 @@ use std::{ boxed::Box, ffi::CString, ops::{Deref, DerefMut}, - sync::{Arc, Mutex}, + sync::{atomic::AtomicBool, Arc, Mutex}, }; use futures::future::BoxFuture; @@ -329,6 +329,7 @@ where let handle = Arc::new(SubscriptionHandle { rcl_subscription: Mutex::new(rcl_subscription), node_handle: Arc::clone(node_handle), + on_ready_slot: AtomicBool::new(false), }); let callback = Arc::new(Mutex::new(callback.into())); diff --git a/rclrs/src/error.rs b/rclrs/src/error.rs index 6f66086a7..39a588c6d 100644 --- a/rclrs/src/error.rs +++ b/rclrs/src/error.rs @@ -80,6 +80,16 @@ pub enum RclrsError { /// However, the implementation of rclrs automatically protects from all of /// these errors except memory allocation failure. GoalAcceptanceError, + /// A second push "on ready" callback was registered on an entity that + /// already has one active. + /// + /// rcl stores a single callback per entity slot, so an overlapping + /// registration would silently clobber the existing one, and later dropping + /// the earlier registration would clear the newer callback. rclrs entities + /// register their callback exactly once and never replace it, so the second + /// registration is rejected rather than allowed to corrupt the RAII + /// handle's semantics. + OnReadyAlreadyRegistered, } impl RclrsError { @@ -168,6 +178,13 @@ impl Display for RclrsError { "An error occurred while trying to accept an action server goal", ) } + RclrsError::OnReadyAlreadyRegistered => { + write!( + f, + "A push on-ready callback is already registered for this entity. \ + rcl allows only one callback per entity slot", + ) + } } } } @@ -216,6 +233,7 @@ impl Error for RclrsError { RclrsError::PoisonedMutex => None, RclrsError::InvalidReadyInformation { .. } => None, RclrsError::GoalAcceptanceError => None, + RclrsError::OnReadyAlreadyRegistered => None, } } } diff --git a/rclrs/src/executor/event_callback.rs b/rclrs/src/executor/event_callback.rs index 3b76de9ca..4992fb02f 100644 --- a/rclrs/src/executor/event_callback.rs +++ b/rclrs/src/executor/event_callback.rs @@ -118,6 +118,27 @@ unsafe extern "C" fn on_ready_trampoline(user_data: *const c_void, number_of_eve /// `rcl_*_set_on_new_*_callback` stay encapsulated there. pub(crate) type SetOnReadyFn = unsafe fn(&H, rcl_event_callback_t, *const c_void) -> rcl_ret_t; +/// Selects the registration guard for the specific rcl callback *slot* that a +/// [`SetOnReadyFn`] installs into. +/// +/// rcl stores exactly one callback per entity slot. Registering a second +/// callback into a slot silently replaces the first, and later clearing the +/// first would then clear the *second* one, leaving that registration's +/// [`OnReadyHandle`] alive while its callback is no longer installed. rclrs +/// entities register a given slot exactly once and never replace it, so rather +/// than track generations to make replacement safe we simply reject a second, +/// overlapping registration on the same slot (see +/// [`RclrsError::OnReadyAlreadyRegistered`]). +/// +/// Most primitives expose a single slot, but composite primitives (action +/// servers and clients) expose several distinct slots on one handle +/// (goal/cancel/result, feedback/status, ...). Pairing each registration with +/// the accessor for *its* slot lets independent slots on the same handle coexist +/// while still rejecting a true overlap on any one slot. The guard is an +/// `AtomicBool` owned by the entity handle and shared by every `Arc` clone of +/// it: `false` while the slot is free, `true` while a registration owns it. +pub(crate) type OnReadySlotFn = fn(&H) -> &AtomicBool; + /// RAII registration of a push "on ready" callback on an rcl entity. /// /// While alive, `on_ready(number_of_events)` is invoked by the middleware @@ -126,6 +147,9 @@ pub(crate) type SetOnReadyFn = unsafe fn(&H, rcl_event_callback_t, *const c_v /// memory. pub(crate) struct OnReadyRegistration { set_callback: SetOnReadyFn, + // Selects the guard for the slot this registration owns, so `Drop` releases + // the same slot it claimed in `new`. + slot: OnReadySlotFn, // Field order is important for teardown safety: `handle` is declared // (and therefore dropped) before `ctx`. Dropping the last `Arc` // finalizes the rcl entity (destroying the middleware reader), so by the @@ -143,13 +167,15 @@ impl OnReadyHandle for OnReadyRegistration {} impl OnReadyRegistration { /// Register `on_ready` to be called by the middleware whenever the entity - /// becomes ready. `set_callback` locks `handle` and installs the trampoline. + /// becomes ready. `set_callback` locks `handle` and installs the trampoline; + /// `slot` selects the guard for the rcl callback slot `set_callback` targets. pub(crate) fn new( handle: Arc, set_callback: SetOnReadyFn, + slot: OnReadySlotFn, on_ready: Box, ) -> Result { - Self::new_with_panic_hook(handle, set_callback, on_ready, None) + Self::new_with_panic_hook(handle, set_callback, slot, on_ready, None) } /// Like [`Self::new`], but invokes `on_panic` once if the user closure @@ -157,9 +183,23 @@ impl OnReadyRegistration { pub(crate) fn new_with_panic_hook( handle: Arc, set_callback: SetOnReadyFn, + slot: OnReadySlotFn, on_ready: Box, on_panic: Option, ) -> Result { + // Claim this registration's callback slot. rcl stores one callback per + // slot, so allowing an overlapping registration would let this one + // silently clobber the existing callback (and dropping the earlier + // registration would later clear ours). Reject the overlap instead of + // corrupting the RAII semantics. + let slot_taken = slot(&handle) + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err(); + + if slot_taken { + return Err(RclrsError::OnReadyAlreadyRegistered); + } + let ctx = Box::into_raw(Box::new(EventCallbackCtx::new(on_ready, on_panic))); // SAFETY: `ctx` points to a live, heap-stable context that outlives the @@ -174,11 +214,14 @@ impl OnReadyRegistration { unsafe { drop(Box::from_raw(ctx)); } + // Release the slot we claimed so a later registration can succeed. + slot(&handle).store(false, Ordering::Release); return Err(err); } Ok(Self { set_callback, + slot, handle, ctx: CtxBox(ctx), }) @@ -198,6 +241,9 @@ impl Drop for OnReadyRegistration { unsafe { let _ = (self.set_callback)(&self.handle, None, std::ptr::null()); } + + // Release this registration's callback slot so a new registration can be made. + (self.slot)(&self.handle).store(false, Ordering::Release); } } @@ -226,7 +272,10 @@ impl Drop for CtxBox { #[cfg(test)] mod tests { use super::*; - use crate::{subscription::set_subscription_on_new_message, *}; + use crate::{ + subscription::{set_subscription_on_new_message, subscription_on_ready_slot}, + *, + }; use ros_env::test_msgs::msg; use std::{ sync::atomic::{AtomicBool, AtomicUsize, Ordering}, @@ -250,6 +299,7 @@ mod tests { let _registration = OnReadyRegistration::new( Arc::clone(subscription.handle()), set_subscription_on_new_message, + subscription_on_ready_slot, Box::new(move |n| { count_cb.fetch_add(n, Ordering::Relaxed); }), @@ -302,6 +352,7 @@ mod tests { let registration = OnReadyRegistration::new( Arc::clone(subscription.handle()), set_subscription_on_new_message, + subscription_on_ready_slot, Box::new(move |n| { count_cb.fetch_add(n, Ordering::Relaxed); }), @@ -339,6 +390,7 @@ mod tests { let _registration = OnReadyRegistration::new_with_panic_hook( Arc::clone(subscription.handle()), set_subscription_on_new_message, + subscription_on_ready_slot, Box::new(move |_| { invocations_cb.fetch_add(1, Ordering::Relaxed); panic!("push callback panic"); @@ -378,4 +430,143 @@ mod tests { ); Ok(()) } + + /// Only one push registration may own an entity's callback slot at a time. + /// A second overlapping registration is rejected instead of silently + /// clobbering the first, and once the first is dropped the slot is free for + /// a fresh registration (which then fires normally). This guards against the + /// RAII hazard where dropping the earlier handle would clear the later + /// callback. + #[test] + fn overlapping_registration_is_rejected() -> Result<(), RclrsError> { + let executor = Context::default().create_basic_executor(); + let node = executor.create_node("test_overlapping_registration_is_rejected")?; + let qos = QoSProfile::default().reliable().keep_last(10); + + let publisher = node.create_publisher::("test_push_overlap_topic".qos(qos))?; + let subscription = node.create_subscription::( + "test_push_overlap_topic".qos(qos), + |_: msg::Empty| {}, + )?; + + let first = OnReadyRegistration::new( + Arc::clone(subscription.handle()), + set_subscription_on_new_message, + subscription_on_ready_slot, + Box::new(|_| {}), + )?; + + // A second registration against the same slot must be rejected while + // the first is still alive. + let second = OnReadyRegistration::new( + Arc::clone(subscription.handle()), + set_subscription_on_new_message, + subscription_on_ready_slot, + Box::new(|_| {}), + ); + assert!( + matches!(second, Err(RclrsError::OnReadyAlreadyRegistered)), + "second overlapping registration should be rejected", + ); + + // Dropping the first releases the slot, so a new registration succeeds + // and receives events. + drop(first); + let count = Arc::new(AtomicUsize::new(0)); + let count_cb = Arc::clone(&count); + let _third = OnReadyRegistration::new( + Arc::clone(subscription.handle()), + set_subscription_on_new_message, + subscription_on_ready_slot, + Box::new(move |n| { + count_cb.fetch_add(n, Ordering::Relaxed); + }), + )?; + + let deadline = Instant::now() + Duration::from_secs(10); + while count.load(Ordering::Relaxed) == 0 && Instant::now() < deadline { + publisher.publish(msg::Empty::default())?; + std::thread::sleep(Duration::from_millis(20)); + } + assert!( + count.load(Ordering::Relaxed) > 0, + "re-registration after drop never fired", + ); + Ok(()) + } + + /// Models the composite-primitive (action server/client) case: several + /// distinct rcl callback *slots* live on one handle. Each slot enforces its + /// own single-registration invariant independently, so registrations on + /// different slots of the same handle coexist, while a repeat registration + /// on an already-owned slot is rejected. Uses a fake handle whose setters are + /// no-ops so the test needs no middleware. + #[test] + fn independent_slots_on_one_handle_coexist() { + struct MultiSlotHandle { + slot_a: AtomicBool, + slot_b: AtomicBool, + } + + fn slot_a(handle: &MultiSlotHandle) -> &AtomicBool { + &handle.slot_a + } + fn slot_b(handle: &MultiSlotHandle) -> &AtomicBool { + &handle.slot_b + } + + // The rcl setters are replaced with no-ops that report success, so the + // registration logic runs without touching a real entity. + unsafe fn set_noop( + _handle: &MultiSlotHandle, + _callback: rcl_event_callback_t, + _user_data: *const c_void, + ) -> rcl_ret_t { + 0 // RCL_RET_OK + } + + let handle = Arc::new(MultiSlotHandle { + slot_a: AtomicBool::new(false), + slot_b: AtomicBool::new(false), + }); + + let reg_a = + OnReadyRegistration::new(Arc::clone(&handle), set_noop, slot_a, Box::new(|_| {})) + .expect("slot A should be free"); + // A different slot on the same handle is independent and must succeed. + let reg_b = + OnReadyRegistration::new(Arc::clone(&handle), set_noop, slot_b, Box::new(|_| {})) + .expect("slot B is independent of slot A"); + + // Re-registering either owned slot is rejected. + assert!( + matches!( + OnReadyRegistration::new(Arc::clone(&handle), set_noop, slot_a, Box::new(|_| {})), + Err(RclrsError::OnReadyAlreadyRegistered), + ), + "slot A is already owned", + ); + assert!( + matches!( + OnReadyRegistration::new(Arc::clone(&handle), set_noop, slot_b, Box::new(|_| {})), + Err(RclrsError::OnReadyAlreadyRegistered), + ), + "slot B is already owned", + ); + + // Dropping one registration frees only its own slot. + drop(reg_a); + assert!( + !handle.slot_a.load(Ordering::Acquire), + "slot A should be freed", + ); + assert!( + handle.slot_b.load(Ordering::Acquire), + "slot B should still be owned by reg_b", + ); + OnReadyRegistration::new(Arc::clone(&handle), set_noop, slot_a, Box::new(|_| {})) + .expect("slot A should be reusable after its registration dropped"); + + drop(reg_b); + } } diff --git a/rclrs/src/service.rs b/rclrs/src/service.rs index f72accb25..07467e0f6 100644 --- a/rclrs/src/service.rs +++ b/rclrs/src/service.rs @@ -2,7 +2,7 @@ use std::{ any::Any, boxed::Box, ffi::{CStr, CString}, - sync::{Arc, Mutex, MutexGuard}, + sync::{atomic::AtomicBool, Arc, Mutex, MutexGuard}, }; use rosidl_runtime_rs::{Message, Service as ServiceIDL}; @@ -146,6 +146,7 @@ where rcl_service: Mutex::new(rcl_service), node_handle: Arc::clone(node.handle()), clock: node.get_clock(), + on_ready_slot: AtomicBool::new(false), }); let (waitable, lifecycle) = Waitable::new( @@ -317,6 +318,7 @@ where let registration = crate::executor::event_callback::OnReadyRegistration::new( Arc::clone(&self.handle), set_service_on_new_request, + service_on_ready_slot, Box::new(on_ready), )?; Ok(Some(Box::new(registration))) @@ -334,6 +336,13 @@ pub(crate) unsafe fn set_service_on_new_request( rcl_service_set_on_new_request_callback(&*handle.lock(), callback, user_data) } +/// The guard for the service's single "on new request" callback slot, paired +/// with [`set_service_on_new_request`] so a push registration can enforce that +/// only one callback owns the slot at a time. +fn service_on_ready_slot(handle: &ServiceHandle) -> &AtomicBool { + &handle.on_ready_slot +} + // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread // they are running in. Therefore, this type can be safely sent to another thread. unsafe impl Send for rcl_service_t {} @@ -347,6 +356,12 @@ pub struct ServiceHandle { rcl_service: Mutex, node_handle: Arc, clock: Clock, + /// Guards the single rcl "on new request" callback slot so only one push + /// registration can own it at a time. Selected via [`service_on_ready_slot`]; + /// see [`OnReadySlotFn`]. + /// + /// [`OnReadySlotFn`]: crate::executor::event_callback::OnReadySlotFn + on_ready_slot: AtomicBool, } impl ServiceHandle { diff --git a/rclrs/src/subscription.rs b/rclrs/src/subscription.rs index e58767788..41a767648 100644 --- a/rclrs/src/subscription.rs +++ b/rclrs/src/subscription.rs @@ -1,7 +1,7 @@ use std::{ any::Any, ffi::{CStr, CString}, - sync::{Arc, Mutex, MutexGuard}, + sync::{atomic::AtomicBool, Arc, Mutex, MutexGuard}, }; use rosidl_runtime_rs::{Message, RmwMessage}; @@ -181,6 +181,7 @@ where let handle = Arc::new(SubscriptionHandle { rcl_subscription: Mutex::new(rcl_subscription), node_handle: Arc::clone(node_handle), + on_ready_slot: AtomicBool::new(false), }); let (waitable, lifecycle) = Waitable::new( @@ -314,6 +315,7 @@ where let registration = crate::executor::event_callback::OnReadyRegistration::new( Arc::clone(&self.handle), set_subscription_on_new_message, + subscription_on_ready_slot, Box::new(on_ready), )?; Ok(Some(Box::new(registration))) @@ -331,6 +333,13 @@ pub(crate) unsafe fn set_subscription_on_new_message( rcl_subscription_set_on_new_message_callback(&*handle.lock(), callback, user_data) } +/// The guard for the subscription's single "on new message" callback slot, +/// paired with [`set_subscription_on_new_message`] so a push registration can +/// enforce that only one callback owns the slot at a time. +pub(crate) fn subscription_on_ready_slot(handle: &SubscriptionHandle) -> &AtomicBool { + &handle.on_ready_slot +} + // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread // they are running in. Therefore, this type can be safely sent to another thread. unsafe impl Send for rcl_subscription_t {} @@ -343,6 +352,12 @@ unsafe impl Send for rcl_subscription_t {} pub(crate) struct SubscriptionHandle { pub(crate) rcl_subscription: Mutex, pub(crate) node_handle: Arc, + /// Guards the single rcl "on new message" callback slot so only one push + /// registration can own it at a time. Selected via + /// [`subscription_on_ready_slot`]; see [`OnReadySlotFn`]. + /// + /// [`OnReadySlotFn`]: crate::executor::event_callback::OnReadySlotFn + pub(crate) on_ready_slot: AtomicBool, } impl SubscriptionHandle {