Apply suggestions from code review

Co-authored-by: Nikolai Morin <nnmmgit@gmail.com>

Author: christopherjreid

rclrs/src/wait.rs

@@ -19,15 +19,14 @@ use crate::error::{to_rclrs_result, RclReturnCode, RclrsError, ToResult};
 use crate::rcl_bindings::*;
 use crate::{ClientBase, Context, ServiceBase, SubscriptionBase};
 
-use std::sync::{atomic::Ordering, Arc, Mutex};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use std::vec::Vec;
 
 mod exclusivity_guard;
-use exclusivity_guard::*;
-
 mod guard_condition;
-use guard_condition::*;
+use exclusivity_guard::*;
+pub use guard_condition::*;
 
 /// A struct for waiting on subscriptions and other waitable entities to become ready.
 pub struct WaitSet {
@@ -40,7 +39,7 @@ pub struct WaitSet {
     subscriptions: Vec<ExclusivityGuard<Arc<dyn SubscriptionBase>>>,
     clients: Vec<ExclusivityGuard<Arc<dyn ClientBase>>>,
     // The guard conditions that are currently registered in the wait set.
-    guard_conditions: Vec<ExclusivityGuard<Arc<Mutex<GuardCondition>>>>,
+    guard_conditions: Vec<ExclusivityGuard<Arc<GuardCondition>>>,
     services: Vec<ExclusivityGuard<Arc<dyn ServiceBase>>>,
 }
 
@@ -50,8 +49,6 @@ pub struct ReadyEntities {
     pub subscriptions: Vec<Arc<dyn SubscriptionBase>>,
     /// A list of clients that have potentially received responses.
     pub clients: Vec<Arc<dyn ClientBase>>,
-    /// A list of guard conditions that have potentially been triggered.
-    pub guard_conditions: Vec<Arc<Mutex<GuardCondition>>>,
     /// A list of services that have potentially received requests.
     pub services: Vec<Arc<dyn ServiceBase>>,
 }
@@ -115,14 +112,7 @@ impl WaitSet {
     /// [`WaitSet::new`].
     pub fn clear(&mut self) {
         self.subscriptions.clear();
-        // Guard conditions must be told that they are no longer tied to a wait set
-        self.guard_conditions.drain(..).for_each(|elem| {
-            elem.waitable
-                .lock()
-                .unwrap()
-                .in_use_by_wait_set
-                .store(false, Ordering::Relaxed)
-        });
+        self.guard_conditions.clear();
         self.clients.clear();
         self.services.clear();
         // This cannot fail – the rcl_wait_set_clear function only checks that the input handle is
@@ -178,18 +168,18 @@ impl WaitSet {
     /// [2]: crate::RclReturnCode
     pub fn add_guard_condition(
         &mut self,
-        guard_condition: Arc<Mutex<GuardCondition>>,
+        guard_condition: Arc<GuardCondition>,
     ) -> Result<(), RclrsError> {
         let exclusive_guard_condition = ExclusivityGuard::new(
             Arc::clone(&guard_condition),
-            Arc::clone(&guard_condition.lock().unwrap().in_use_by_wait_set),
+            Arc::clone(&guard_condition.in_use_by_wait_set),
         )?;
 
         unsafe {
             // SAFETY: Safe if the wait set and guard condition are initialized
             rcl_wait_set_add_guard_condition(
                 &mut self.rcl_wait_set,
-                &*guard_condition.lock().unwrap().lock(),
+                &*guard_condition.lock(),
                 std::ptr::null_mut(),
             )
             .ok()?;
@@ -300,7 +290,6 @@ impl WaitSet {
         unsafe { rcl_wait(&mut self.rcl_wait_set, timeout_ns) }.ok()?;
         let mut ready_entities = ReadyEntities {
             subscriptions: Vec::new(),
-            guard_conditions: Vec::new(),
             clients: Vec::new(),
             services: Vec::new(),
         };
@@ -316,18 +305,6 @@ impl WaitSet {
             }
         }
 
-        for (i, guard_condition) in self.guard_conditions.iter().enumerate() {
-            // SAFETY: The `guard_conditions` entry is an array of pointers, and this dereferencing is
-            // equivalent to
-            // https://github.com/ros2/rcl/blob/35a31b00a12f259d492bf53c0701003bd7f1745c/rcl/include/rcl/wait.h#L419
-            let wait_set_entry = unsafe { *self.rcl_wait_set.guard_conditions.add(i) };
-            if !wait_set_entry.is_null() {
-                ready_entities
-                    .guard_conditions
-                    .push(Arc::clone(&guard_condition.waitable));
-            }
-        }
-
         for (i, client) in self.clients.iter().enumerate() {
             // SAFETY: The `clients` entry is an array of pointers, and this dereferencing is
             // equivalent to

rclrs/src/wait/guard_condition.rs

@@ -4,15 +4,42 @@ use crate::{Context, RclrsError, ToResult};
 use std::sync::{atomic::AtomicBool, Arc, Mutex, MutexGuard};
 
 /// A struct for encapsulating a guard condition - a waitable trigger
+///
+/// # Example
+/// ```
+/// # use rclrs::{Context, GuardCondition, WaitSet, RclrsError};
+/// # use std::sync::{Arc, atomic::Ordering};
+///
+/// let context = Context::new([])?;
+///
+/// let atomic_bool = Arc::new(std::sync::atomic::AtomicBool::new(false));
+/// let atomic_bool_for_closure = Arc::clone(&atomic_bool);
+///
+/// let gc = Arc::new(GuardCondition::new(
+///     &context,
+///     Some(Box::new(move || {
+///         atomic_bool_for_closure.store(true, Ordering::Relaxed);
+///     })),
+/// ));
+///
+/// let mut ws = WaitSet::new(0, 1, 0, 0, 0, 0, &context)?;
+/// ws.add_guard_condition(Arc::clone(&gc))?;
+///
+/// // Trigger the guard condition, firing the callback and waking any wait sets being waited on
+/// gc.trigger()?;
+///
+/// // The wait call will now immediately return
+/// ws.wait(Some(std::time::Duration::from_millis(10)))?;
+///
+/// # Ok::<(), RclrsError>(())
+/// ```
 pub struct GuardCondition {
     /// The rcl_guard_condition_t that this struct encapsulates.
     rcl_guard_condition: Arc<Mutex<rcl_guard_condition_t>>,
     /// An optional callback to call when this guard condition is triggered.
-    callback: Option<Box<dyn Fn(usize)>>,
+    callback: Option<Box<dyn Fn() + Send>>,
     /// A flag to indicate if this guard condition has already been assigned to a wait set.
     pub(crate) in_use_by_wait_set: Arc<AtomicBool>,
-    /// A count for the number of times this guard condition was triggered, but no callback was assigned.
-    unread_count: usize,
 }
 
 impl Drop for GuardCondition {
@@ -24,12 +51,19 @@ impl Drop for GuardCondition {
     }
 }
 
+// SAFETY: rcl_guard_condition is the only member that doesn't implement Send, and it is designed to be accessed from other threads
+unsafe impl Send for GuardCondition {}
+
+// SAFETY: The mutexes and atomic members ensure synchronized access to members, and the callback is reentrant
+unsafe impl Sync for GuardCondition {}
+
 impl GuardCondition {
     /// Creates a new guard condition
-    pub fn new(context: &Context) -> Self {
+    pub fn new(context: &Context, callback: Option<Box<dyn Fn() + Send>>) -> Self {
+        // SAFETY: Getting a zero initialized value is always safe
         let mut guard_condition = unsafe { rcl_get_zero_initialized_guard_condition() };
         unsafe {
-            // SAFETY: The context must be valid. No other preconditions for this function.
+            // SAFETY: The context must be valid, and the guard condition must be zero-initialized
             rcl_guard_condition_init(
                 &mut guard_condition,
                 &mut *context.rcl_context_mtx.lock().unwrap(),
@@ -39,9 +73,8 @@ impl GuardCondition {
 
         Self {
             rcl_guard_condition: Arc::new(Mutex::new(guard_condition)),
-            callback: None,
+            callback,
             in_use_by_wait_set: Arc::new(AtomicBool::new(false)),
-            unread_count: 0,
         }
     }
 
@@ -50,32 +83,14 @@ impl GuardCondition {
         self.rcl_guard_condition.lock().unwrap()
     }
 
-    /// Sets the callback to call when this guard condition is triggered.
-    pub fn set_on_trigger_callback(&mut self, callback: Option<Box<dyn Fn(usize)>>) {
-        match callback {
-            Some(callback) => {
-                if self.unread_count > 0 {
-                    callback(self.unread_count);
-                    self.unread_count = 0;
-                }
-                self.callback = Some(callback);
-            }
-            None => {
-                self.callback = None;
-                self.unread_count = 0;
-            }
-        }
-    }
-
     /// Triggers this guard condition, activating the wait set, and calling the optionally assigned callback.
-    pub fn trigger(&mut self) -> Result<(), RclrsError> {
+    pub fn trigger(&self) -> Result<(), RclrsError> {
         unsafe {
             // SAFETY: The rcl_guard_condition_t is valid.
             rcl_trigger_guard_condition(&mut *self.rcl_guard_condition.lock().unwrap()).ok()?;
         }
-        match &self.callback {
-            Some(callback) => callback(1),
-            None => self.unread_count += 1,
+        if self.callback.is_some() {
+            self.callback.as_ref().unwrap()();
         }
         Ok(())
     }
@@ -90,43 +105,44 @@ mod tests {
     #[test]
     fn test_guard_condition() -> Result<(), RclrsError> {
         let context = Context::new([])?;
-        let mut gc = GuardCondition::new(&context);
 
-        let atomic_usize = Arc::new(std::sync::atomic::AtomicUsize::new(0));
-        let atomic_usize_for_closure = Arc::clone(&atomic_usize);
+        let atomic_bool = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let atomic_bool_for_closure = Arc::clone(&atomic_bool);
 
-        gc.set_on_trigger_callback(Some(Box::new(move |count| {
-            atomic_usize_for_closure.store(count, Ordering::Relaxed);
-        })));
+        let gc = GuardCondition::new(
+            &context,
+            Some(Box::new(move || {
+                atomic_bool_for_closure.store(true, Ordering::Relaxed);
+            })),
+        );
 
         gc.trigger()?;
 
-        assert_eq!(atomic_usize.load(Ordering::Relaxed), 1);
+        assert_eq!(atomic_bool.load(Ordering::Relaxed), true);
 
         Ok(())
     }
 
     #[test]
     fn test_guard_condition_wait() -> Result<(), RclrsError> {
         let context = Context::new([])?;
-        let gc = Arc::new(Mutex::new(GuardCondition::new(&context)));
 
-        let atomic_usize = Arc::new(std::sync::atomic::AtomicUsize::new(0));
-        let atomic_usize_for_closure = Arc::clone(&atomic_usize);
+        let atomic_bool = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let atomic_bool_for_closure = Arc::clone(&atomic_bool);
 
-        gc.lock()
-            .unwrap()
-            .set_on_trigger_callback(Some(Box::new(move |count| {
-                atomic_usize_for_closure.store(count, Ordering::Relaxed);
-            })));
+        let gc = Arc::new(GuardCondition::new(
+            &context,
+            Some(Box::new(move || {
+                atomic_bool_for_closure.store(true, Ordering::Relaxed);
+            })),
+        ));
 
         let mut ws = WaitSet::new(0, 1, 0, 0, 0, 0, &context)?;
         ws.add_guard_condition(Arc::clone(&gc))?;
-        gc.lock().unwrap().trigger()?;
+        gc.trigger()?;
 
-        assert_eq!(atomic_usize.load(Ordering::Relaxed), 1);
-        let wait_result = ws.wait(Some(std::time::Duration::from_millis(10)))?;
-        assert_eq!(wait_result.guard_conditions.len(), 1);
+        assert_eq!(atomic_bool.load(Ordering::Relaxed), true);
+        ws.wait(Some(std::time::Duration::from_millis(10)))?;
 
         Ok(())
     }