Fix RAII for subscription, client, and service (#463)

mxgrey · web-flow · commit 0fe1f2154516 · 2025-03-10T15:15:12.000+01:00
* Fix RAII for subscription, client, and service

Signed-off-by: Michael X. Grey &lt;greyxmike@gmail.com&gt;

* Fix style

Signed-off-by: Michael X. Grey &lt;greyxmike@gmail.com&gt;

* Fix compilation errors

Signed-off-by: Michael X. Grey &lt;mxgrey@intrinsic.ai&gt;

---------

Signed-off-by: Michael X. Grey &lt;greyxmike@gmail.com&gt;
Signed-off-by: Michael X. Grey &lt;mxgrey@intrinsic.ai&gt;
diff --git a/rclrs/src/client.rs b/rclrs/src/client.rs
@@ -11,7 +11,7 @@ use rosidl_runtime_rs::Message;
 use crate::{
     error::{RclReturnCode, ToResult},
     rcl_bindings::*,
-    MessageCow, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
+    MessageCow, Node, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
 };
 
 // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread
@@ -76,14 +76,18 @@ where
     pub(crate) handle: Arc<ClientHandle>,
     requests: Mutex<HashMap<RequestId, RequestValue<T::Response>>>,
     futures: Arc<Mutex<HashMap<RequestId, oneshot::Sender<T::Response>>>>,
+    /// Ensure the parent node remains alive as long as the subscription is held.
+    /// This implementation will change in the future.
+    #[allow(unused)]
+    node: Arc<Node>,
 }
 
 impl<T> Client<T>
 where
     T: rosidl_runtime_rs::Service,
 {
     /// Creates a new client.
-    pub(crate) fn new(node_handle: Arc<NodeHandle>, topic: &str) -> Result<Self, RclrsError>
+    pub(crate) fn new(node: &Arc<Node>, topic: &str) -> Result<Self, RclrsError>
     // This uses pub(crate) visibility to avoid instantiating this struct outside
     // [`Node::create_client`], see the struct's documentation for the rationale
     where
@@ -102,7 +106,7 @@ where
         let client_options = unsafe { rcl_client_get_default_options() };
 
         {
-            let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let rcl_node = node.handle.rcl_node.lock().unwrap();
             let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
 
             // SAFETY:
@@ -126,7 +130,7 @@ where
 
         let handle = Arc::new(ClientHandle {
             rcl_client: Mutex::new(rcl_client),
-            node_handle,
+            node_handle: Arc::clone(&node.handle),
             in_use_by_wait_set: Arc::new(AtomicBool::new(false)),
         });
 
@@ -136,6 +140,7 @@ where
             futures: Arc::new(Mutex::new(
                 HashMap::<RequestId, oneshot::Sender<T::Response>>::new(),
             )),
+            node: Arc::clone(node),
         })
     }
 
diff --git a/rclrs/src/node.rs b/rclrs/src/node.rs
@@ -225,11 +225,11 @@ impl Node {
     ///
     /// [1]: crate::Client
     // TODO: make client's lifetime depend on node's lifetime
-    pub fn create_client<T>(&self, topic: &str) -> Result<Arc<Client<T>>, RclrsError>
+    pub fn create_client<T>(self: &Arc<Self>, topic: &str) -> Result<Arc<Client<T>>, RclrsError>
     where
         T: rosidl_runtime_rs::Service,
     {
-        let client = Arc::new(Client::<T>::new(Arc::clone(&self.handle), topic)?);
+        let client = Arc::new(Client::<T>::new(self, topic)?);
         { self.clients_mtx.lock().unwrap() }.push(Arc::downgrade(&client) as Weak<dyn ClientBase>);
         Ok(client)
     }
@@ -292,19 +292,15 @@ impl Node {
     ///
     /// [1]: crate::Service
     pub fn create_service<T, F>(
-        &self,
+        self: &Arc<Self>,
         topic: &str,
         callback: F,
     ) -> Result<Arc<Service<T>>, RclrsError>
     where
         T: rosidl_runtime_rs::Service,
         F: Fn(&rmw_request_id_t, T::Request) -> T::Response + 'static + Send,
     {
-        let service = Arc::new(Service::<T>::new(
-            Arc::clone(&self.handle),
-            topic,
-            callback,
-        )?);
+        let service = Arc::new(Service::<T>::new(self, topic, callback)?);
         { self.services_mtx.lock().unwrap() }
             .push(Arc::downgrade(&service) as Weak<dyn ServiceBase>);
         Ok(service)
@@ -314,20 +310,15 @@ impl Node {
     ///
     /// [1]: crate::Subscription
     pub fn create_subscription<T, Args>(
-        &self,
+        self: &Arc<Self>,
         topic: &str,
         qos: QoSProfile,
         callback: impl SubscriptionCallback<T, Args>,
     ) -> Result<Arc<Subscription<T>>, RclrsError>
     where
         T: Message,
     {
-        let subscription = Arc::new(Subscription::<T>::new(
-            Arc::clone(&self.handle),
-            topic,
-            qos,
-            callback,
-        )?);
+        let subscription = Arc::new(Subscription::<T>::new(self, topic, qos, callback)?);
         { self.subscriptions_mtx.lock() }
             .unwrap()
             .push(Arc::downgrade(&subscription) as Weak<dyn SubscriptionBase>);
diff --git a/rclrs/src/parameter.rs b/rclrs/src/parameter.rs
@@ -789,7 +789,7 @@ impl ParameterInterface {
         }
     }
 
-    pub(crate) fn create_services(&self, node: &Node) -> Result<(), RclrsError> {
+    pub(crate) fn create_services(&self, node: &Arc<Node>) -> Result<(), RclrsError> {
         *self.services.lock().unwrap() =
             Some(ParameterService::new(node, self.parameter_map.clone())?);
         Ok(())
diff --git a/rclrs/src/parameter/service.rs b/rclrs/src/parameter/service.rs
@@ -239,7 +239,7 @@ fn set_parameters_atomically(
 
 impl ParameterService {
     pub(crate) fn new(
-        node: &Node,
+        node: &Arc<Node>,
         parameter_map: Arc<Mutex<ParameterMap>>,
     ) -> Result<Self, RclrsError> {
         let fqn = node.fully_qualified_name();
diff --git a/rclrs/src/service.rs b/rclrs/src/service.rs
@@ -9,7 +9,7 @@ use rosidl_runtime_rs::Message;
 use crate::{
     error::{RclReturnCode, ToResult},
     rcl_bindings::*,
-    MessageCow, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
+    MessageCow, Node, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
 };
 
 // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread
@@ -73,18 +73,18 @@ where
     pub(crate) handle: Arc<ServiceHandle>,
     /// The callback function that runs when a request was received.
     pub callback: Mutex<ServiceCallback<T::Request, T::Response>>,
+    /// Ensure the parent node remains alive as long as the subscription is held.
+    /// This implementation will change in the future.
+    #[allow(unused)]
+    node: Arc<Node>,
 }
 
 impl<T> Service<T>
 where
     T: rosidl_runtime_rs::Service,
 {
     /// Creates a new service.
-    pub(crate) fn new<F>(
-        node_handle: Arc<NodeHandle>,
-        topic: &str,
-        callback: F,
-    ) -> Result<Self, RclrsError>
+    pub(crate) fn new<F>(node: &Arc<Node>, topic: &str, callback: F) -> Result<Self, RclrsError>
     // This uses pub(crate) visibility to avoid instantiating this struct outside
     // [`Node::create_service`], see the struct's documentation for the rationale
     where
@@ -104,7 +104,7 @@ where
         let service_options = unsafe { rcl_service_get_default_options() };
 
         {
-            let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let rcl_node = node.handle.rcl_node.lock().unwrap();
             let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             unsafe {
                 // SAFETY:
@@ -127,12 +127,13 @@ where
 
         let handle = Arc::new(ServiceHandle {
             rcl_service: Mutex::new(rcl_service),
-            node_handle,
+            node_handle: Arc::clone(&node.handle),
             in_use_by_wait_set: Arc::new(AtomicBool::new(false)),
         });
 
         Ok(Self {
             handle,
+            node: Arc::clone(node),
             callback: Mutex::new(Box::new(callback)),
         })
     }
diff --git a/rclrs/src/subscription.rs b/rclrs/src/subscription.rs
@@ -10,7 +10,7 @@ use crate::{
     error::{RclReturnCode, ToResult},
     qos::QoSProfile,
     rcl_bindings::*,
-    NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
+    Node, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
 };
 
 mod callback;
@@ -84,6 +84,10 @@ where
     pub(crate) handle: Arc<SubscriptionHandle>,
     /// The callback function that runs when a message was received.
     pub callback: Mutex<AnySubscriptionCallback<T>>,
+    /// Ensure the parent node remains alive as long as the subscription is held.
+    /// This implementation will change in the future.
+    #[allow(unused)]
+    node: Arc<Node>,
     message: PhantomData<T>,
 }
 
@@ -93,7 +97,7 @@ where
 {
     /// Creates a new subscription.
     pub(crate) fn new<Args>(
-        node_handle: Arc<NodeHandle>,
+        node: &Arc<Node>,
         topic: &str,
         qos: QoSProfile,
         callback: impl SubscriptionCallback<T, Args>,
@@ -117,7 +121,7 @@ where
         subscription_options.qos = qos.into();
 
         {
-            let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let rcl_node = node.handle.rcl_node.lock().unwrap();
             let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             unsafe {
                 // SAFETY:
@@ -139,13 +143,14 @@ where
 
         let handle = Arc::new(SubscriptionHandle {
             rcl_subscription: Mutex::new(rcl_subscription),
-            node_handle,
+            node_handle: Arc::clone(&node.handle),
             in_use_by_wait_set: Arc::new(AtomicBool::new(false)),
         });
 
         Ok(Self {
             handle,
             callback: Mutex::new(callback.into_callback()),
+            node: Arc::clone(node),
             message: PhantomData,
         })
     }
@@ -396,4 +401,41 @@ mod tests {
         );
         Ok(())
     }
+
+    #[test]
+    fn test_node_subscription_raii() {
+        use crate::*;
+        use std::sync::atomic::Ordering;
+
+        let mut executor = Context::default().create_basic_executor();
+
+        let triggered = Arc::new(AtomicBool::new(false));
+        let inner_triggered = Arc::clone(&triggered);
+        let callback = move |_: msg::Empty| {
+            inner_triggered.store(true, Ordering::Release);
+        };
+
+        let (_subscription, publisher) = {
+            let node = executor
+                .create_node(&format!("test_node_subscription_raii_{}", line!()))
+                .unwrap();
+
+            let qos = QoSProfile::default().keep_all().reliable();
+            let subscription = node
+                .create_subscription::<msg::Empty, _>("test_topic", qos, callback)
+                .unwrap();
+            let publisher = node
+                .create_publisher::<msg::Empty>("test_topic", qos)
+                .unwrap();
+
+            (subscription, publisher)
+        };
+
+        publisher.publish(msg::Empty::default()).unwrap();
+        let start_time = std::time::Instant::now();
+        while !triggered.load(Ordering::Acquire) {
+            assert!(executor.spin(SpinOptions::spin_once()).is_empty());
+            assert!(start_time.elapsed() < std::time::Duration::from_secs(10));
+        }
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -789,7 +789,7 @@ impl ParameterInterface {`
`789`	`789`	`}`
`790`	`790`	`}`
`791`	`791`
`792`		`- pub(crate) fn create_services(&self, node: &Node) -> Result<(), RclrsError> {`
	`792`	`+ pub(crate) fn create_services(&self, node: &Arc<Node>) -> Result<(), RclrsError> {`
`793`	`793`	`*self.services.lock().unwrap() =`
`794`	`794`	`Some(ParameterService::new(node, self.parameter_map.clone())?);`
`795`	`795`	`Ok(())`