Make all the things Send, and messages Sync as well (#171)

This is required to make multithreading work. For instance, calling publish() on a separate thread doesn't work without these impls.

Author: nnmm

rclrs/src/context.rs

@@ -24,6 +24,10 @@ impl Drop for rcl_context_t {
     }
 }
 
+// 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_context_t {}
+
 /// Shared state between nodes and similar entities.
 ///
 /// It is possible, but not usually necessary, to have several contexts in an application.

rclrs/src/node/mod.rs

@@ -24,6 +24,10 @@ impl Drop for rcl_node_t {
     }
 }
 
+// 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_node_t {}
+
 /// A processing unit that can communicate with other nodes.
 ///
 /// Nodes are a core concept in ROS 2. Refer to the official ["Understanding ROS 2 nodes"][1]
@@ -263,7 +267,7 @@ impl Node {
     ) -> Result<Arc<Subscription<T>>, RclrsError>
     where
         T: Message,
-        F: FnMut(T) + 'static,
+        F: FnMut(T) + 'static + Send,
     {
         let subscription = Arc::new(Subscription::<T>::new(self, topic, qos, callback)?);
         self.subscriptions

rclrs/src/node/publisher.rs

@@ -12,6 +12,10 @@ use parking_lot::{Mutex, MutexGuard};
 
 use rosidl_runtime_rs::{Message, RmwMessage};
 
+// 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_publisher_t {}
+
 pub(crate) struct PublisherHandle {
     handle: Mutex<rcl_publisher_t>,
     node_handle: Arc<Mutex<rcl_node_t>>,

rclrs/src/node/subscription.rs

@@ -13,6 +13,10 @@ use rosidl_runtime_rs::{Message, RmwMessage};
 
 use parking_lot::{Mutex, MutexGuard};
 
+// 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 {}
+
 /// Internal struct used by subscriptions.
 pub struct SubscriptionHandle {
     handle: Mutex<rcl_subscription_t>,
@@ -37,7 +41,7 @@ impl Drop for SubscriptionHandle {
 }
 
 /// Trait to be implemented by concrete [`Subscription`]s.
-pub trait SubscriptionBase {
+pub trait SubscriptionBase: Send + Sync {
     /// Internal function to get a reference to the `rcl` handle.
     fn handle(&self) -> &SubscriptionHandle;
     /// Tries to take a new message and run the callback with it.
@@ -61,7 +65,7 @@ where
 {
     pub(crate) handle: Arc<SubscriptionHandle>,
     /// The callback function that runs when a message was received.
-    pub callback: Mutex<Box<dyn FnMut(T) + 'static>>,
+    pub callback: Mutex<Box<dyn FnMut(T) + 'static + Send>>,
     message: PhantomData<T>,
 }
 
@@ -81,7 +85,7 @@ where
     ) -> Result<Self, RclrsError>
     where
         T: Message,
-        F: FnMut(T) + 'static,
+        F: FnMut(T) + 'static + Send,
     {
         // SAFETY: Getting a zero-initialized value is always safe.
         let mut subscription_handle = unsafe { rcl_get_zero_initialized_subscription() };

rosidl_runtime_rs/src/sequence.rs

@@ -235,6 +235,11 @@ impl<T: SequenceAlloc + PartialOrd> PartialOrd for Sequence<T> {
     }
 }
 
+// SAFETY: A sequence is a simple data structure, and therefore not thread-specific.
+unsafe impl<T: Send + SequenceAlloc> Send for Sequence<T> {}
+// SAFETY: A sequence does not have interior mutability, so it can be shared.
+unsafe impl<T: Sync + SequenceAlloc> Sync for Sequence<T> {}
+
 impl<T> Sequence<T>
 where
     T: SequenceAlloc,

rosidl_runtime_rs/src/string.rs

@@ -213,6 +213,11 @@ macro_rules! string_impl {
             }
         }
 
+        // SAFETY: A string is a simple data structure, and therefore not thread-specific.
+        unsafe impl Send for $string {}
+        // SAFETY: A string does not have interior mutability, so it can be shared.
+        unsafe impl Sync for $string {}
+
         impl SequenceAlloc for $string {
             fn sequence_init(seq: &mut Sequence<Self>, size: libc::size_t) -> bool {
                 // SAFETY: There are no special preconditions to the sequence_init function.

rosidl_runtime_rs/src/traits.rs

@@ -37,7 +37,7 @@ pub trait SequenceAlloc: Sized {
 /// used by user code.
 ///
 /// User code never needs to call this trait's method, much less implement this trait.
-pub trait RmwMessage: Clone + Debug + Default {
+pub trait RmwMessage: Clone + Debug + Default + Send + Sync {
     /// Get a pointer to the correct `rosidl_message_type_support_t` structure.
     fn get_type_support() -> libc::uintptr_t;
 }
@@ -126,7 +126,7 @@ pub trait RmwMessage: Clone + Debug + Default {
 ///  problem, since nothing is allocated this way.
 /// The `Drop` impl for any sequence or string will call `fini()`.
 
-pub trait Message: Clone + Debug + Default + 'static {
+pub trait Message: Clone + Debug + Default + 'static + Send + Sync {
     /// The corresponding RMW-native message type.
     type RmwMsg: RmwMessage;