Apply lifecycle lock to all middleware entities

Signed-off-by: Michael X. Grey <[email protected]>

Author: mxgrey

rclrs/Cargo.toml

@@ -14,15 +14,21 @@ path = "src/lib.rs"
 # Please keep the list of dependencies alphabetically sorted,
 # and also state why each dependency is needed.
 [dependencies]
-# Needed for dynamically finding type support libraries 
+# Needed for dynamically finding type support libraries
 ament_rs = { version = "0.2", optional = true }
+
 # Needed for uploading documentation to docs.rs
 cfg-if = "1.0.0"
 
 # Needed for clients
 futures = "0.3"
+
+# Needed to create a global mutex for managing the lifecycles of middleware entities
+lazy_static = "1.4"
+
 # Needed for dynamic messages
 libloading = { version = "0.8", optional = true }
+
 # Needed for the Message trait, among others
 rosidl_runtime_rs = "0.4"
 

rclrs/src/client.rs

@@ -9,7 +9,7 @@ use rosidl_runtime_rs::Message;
 
 use crate::error::{RclReturnCode, ToResult};
 use crate::MessageCow;
-use crate::{rcl_bindings::*, RclrsError, NodeHandle};
+use crate::{rcl_bindings::*, RclrsError, NodeHandle, ENTITY_LIFECYCLE_MUTEX};
 
 // 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.
@@ -32,6 +32,7 @@ impl Drop for ClientHandle {
     fn drop(&mut self) {
         let rcl_client = self.rcl_client.get_mut().unwrap();
         let mut rcl_node = self.node_handle.rcl_node.lock().unwrap();
+        let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
         // SAFETY: No preconditions for this function
         unsafe {
             rcl_client_fini(rcl_client, &mut *rcl_node);
@@ -99,6 +100,7 @@ where
             // afterwards.
             {
                 let mut rcl_node = node_handle.rcl_node.lock().unwrap();
+                let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
                 rcl_client_init(
                     &mut rcl_client,
                     &mut *rcl_node,

rclrs/src/context.rs

@@ -7,13 +7,23 @@ use std::vec::Vec;
 use crate::rcl_bindings::*;
 use crate::{RclrsError, ToResult};
 
+lazy_static::lazy_static! {
+    /// This is locked whenever initializing or dropping any middleware entity
+    /// because we have found issues in RCL and some RMW implementations that
+    /// make it unsafe to simultaneously initialize and/or drop various types of
+    /// entities. It seems these C and C++ based libraries will regularly use
+    /// unprotected global variables in their object initialization and cleanup.
+    pub(crate) static ref ENTITY_LIFECYCLE_MUTEX: Mutex<()> = Mutex::new(());
+}
+
 impl Drop for rcl_context_t {
     fn drop(&mut self) {
         unsafe {
             // The context may be invalid when rcl_init failed, e.g. because of invalid command
             // line arguments.
             // SAFETY: No preconditions for this function.
             if rcl_context_is_valid(self) {
+                let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
                 // SAFETY: These functions have no preconditions besides a valid rcl_context
                 rcl_shutdown(self);
                 rcl_context_fini(self);
@@ -102,17 +112,20 @@ impl Context {
             let mut rcl_init_options = options.into_rcl(allocator)?;
             // SAFETY: This function does not store the ephemeral init_options and c_args
             // pointers. Passing in a zero-initialized rcl_context is expected.
-            let ret = rcl_init(
-                c_args.len() as i32,
-                if c_args.is_empty() {
-                    std::ptr::null()
-                } else {
-                    c_args.as_ptr()
-                },
-                &rcl_init_options,
-                &mut rcl_context,
-            )
-            .ok();
+            let ret = {
+                let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
+                rcl_init(
+                    c_args.len() as i32,
+                    if c_args.is_empty() {
+                        std::ptr::null()
+                    } else {
+                        c_args.as_ptr()
+                    },
+                    &rcl_init_options,
+                    &mut rcl_context,
+                )
+                .ok()
+            };
             // SAFETY: It's safe to pass in an initialized object.
             // Early return will not leak memory, because this is the last fini function.
             rcl_init_options_fini(&mut rcl_init_options).ok()?;

rclrs/src/node.rs

@@ -13,7 +13,7 @@ pub use self::builder::*;
 pub use self::graph::*;
 use crate::rcl_bindings::*;
 use crate::{
-    Client, ClientBase, Clock, Context, GuardCondition, ParameterBuilder, ParameterInterface,
+    Client, ClientBase, Clock, Context, ENTITY_LIFECYCLE_MUTEX, GuardCondition, ParameterBuilder, ParameterInterface,
     ParameterVariant, Parameters, Publisher, QoSProfile, RclrsError, Service, ServiceBase,
     Subscription, SubscriptionBase, SubscriptionCallback, TimeSource, ContextHandle,
 };
@@ -78,6 +78,7 @@ impl Drop for NodeHandle {
     fn drop(&mut self) {
         let _context_lock = self.context_handle.rcl_context.lock().unwrap();
         let mut rcl_node = self.rcl_node.lock().unwrap();
+        let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
         unsafe { rcl_node_fini(&mut *rcl_node) };
     }
 }

rclrs/src/node/builder.rs

@@ -3,7 +3,7 @@ use std::sync::{Arc, Mutex};
 
 use crate::rcl_bindings::*;
 use crate::{
-    ClockType, Context, ContextHandle, Node, NodeHandle, ParameterInterface, QoSProfile, RclrsError, TimeSource, ToResult,
+    ClockType, Context, ContextHandle, ENTITY_LIFECYCLE_MUTEX, Node, NodeHandle, ParameterInterface, QoSProfile, RclrsError, TimeSource, ToResult,
     QOS_PROFILE_CLOCK,
 };
 
@@ -270,6 +270,7 @@ impl NodeBuilder {
             // The strings and node options are copied by this function, so we don't need
             // to keep them alive.
             // The rcl_context has to be kept alive because it is co-owned by the node.
+            let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             rcl_node_init(
                 &mut rcl_node,
                 node_name.as_ptr(),

rclrs/src/publisher.rs

@@ -9,7 +9,7 @@ use rosidl_runtime_rs::{Message, RmwMessage};
 use crate::error::{RclrsError, ToResult};
 use crate::qos::QoSProfile;
 use crate::rcl_bindings::*;
-use crate::NodeHandle;
+use crate::{NodeHandle, ENTITY_LIFECYCLE_MUTEX};
 
 mod loaned_message;
 pub use loaned_message::*;
@@ -28,6 +28,7 @@ impl Drop for PublisherHandle {
         unsafe {
             // SAFETY: No preconditions for this function (besides the arguments being valid).
             let mut rcl_node = self.node_handle.rcl_node.lock().unwrap();
+            let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             rcl_publisher_fini(
                 self.rcl_publisher.get_mut().unwrap(),
                 &mut *rcl_node,
@@ -98,6 +99,7 @@ where
             // afterwards.
             // TODO: type support?
             let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             rcl_publisher_init(
                 &mut rcl_publisher,
                 &*rcl_node,

rclrs/src/service.rs

@@ -7,7 +7,7 @@ use rosidl_runtime_rs::Message;
 
 use crate::error::{RclReturnCode, ToResult};
 use crate::{rcl_bindings::*, MessageCow, RclrsError};
-use crate::NodeHandle;
+use crate::{NodeHandle, ENTITY_LIFECYCLE_MUTEX};
 
 // 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.
@@ -30,6 +30,7 @@ impl Drop for ServiceHandle {
     fn drop(&mut self) {
         let rcl_service = self.rcl_service.get_mut().unwrap();
         let mut rcl_node = self.node_handle.rcl_node.lock().unwrap();
+        let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
         // SAFETY: No preconditions for this function
         unsafe {
             rcl_service_fini(rcl_service, &mut *rcl_node);
@@ -100,6 +101,7 @@ where
             // The topic name and the options are copied by this function, so they can be dropped
             // afterwards.
             let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             rcl_service_init(
                 &mut rcl_service,
                 &*rcl_node,

rclrs/src/subscription.rs

@@ -8,7 +8,7 @@ use rosidl_runtime_rs::{Message, RmwMessage};
 
 use crate::error::{RclReturnCode, ToResult};
 use crate::qos::QoSProfile;
-use crate::{rcl_bindings::*, RclrsError, NodeHandle};
+use crate::{rcl_bindings::*, RclrsError, NodeHandle, ENTITY_LIFECYCLE_MUTEX};
 
 mod callback;
 mod message_info;
@@ -38,6 +38,7 @@ impl Drop for SubscriptionHandle {
     fn drop(&mut self) {
         let rcl_subscription = self.rcl_subscription.get_mut().unwrap();
         let mut rcl_node = self.node_handle.rcl_node.lock().unwrap();
+        let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
         // SAFETY: No preconditions for this function (besides the arguments being valid).
         unsafe {
             rcl_subscription_fini(rcl_subscription, &mut *rcl_node);
@@ -114,6 +115,7 @@ where
             // afterwards.
             // TODO: type support?
             let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             rcl_subscription_init(
                 &mut rcl_subscription,
                 &*rcl_node,

rclrs/src/wait.rs

@@ -96,6 +96,7 @@ impl WaitSet {
         let rcl_wait_set = unsafe {
             // SAFETY: Getting a zero-initialized value is always safe
             let mut rcl_wait_set = rcl_get_zero_initialized_wait_set();
+            let mut rcl_context = context.handle.rcl_context.lock().unwrap();
             // SAFETY: We're passing in a zero-initialized wait set and a valid context.
             // There are no other preconditions.
             rcl_wait_set_init(
@@ -106,7 +107,7 @@ impl WaitSet {
                 number_of_clients,
                 number_of_services,
                 number_of_events,
-                &mut *context.handle.rcl_context.lock().unwrap(),
+                &mut *rcl_context,
                 rcutils_get_default_allocator(),
             )
             .ok()?;

rclrs/src/wait/guard_condition.rs

@@ -110,9 +110,9 @@ impl GuardCondition {
         callback: Option<Box<dyn Fn() + Send + Sync>>,
     ) -> Self {
         let rcl_guard_condition = {
-            let mut rcl_context = context_handle.rcl_context.lock().unwrap();
             // SAFETY: Getting a zero initialized value is always safe
             let mut guard_condition = unsafe { rcl_get_zero_initialized_guard_condition() };
+            let mut rcl_context = context_handle.rcl_context.lock().unwrap();
             unsafe {
                 // SAFETY: The context must be valid, and the guard condition must be zero-initialized
                 rcl_guard_condition_init(