Clarify interactions between count_unread_samples() and take_next()

Signed-off-by: Andrea Sorbini <[email protected]>

Author: asorbini

rmw_connextdds_common/include/rmw_connextdds/rmw_impl.hpp

@@ -453,6 +453,30 @@ class RMW_Connext_Subscriber
     return has_data;
   }
 
+  rmw_ret_t
+  count_unread_samples(size_t & unread_count)
+  {
+    // The action of counting unread samples is not currently mutually exclusive
+    // with the action of taking samples out of the reader cache. Unfortunately
+    // we cannot use a mutex to synchronize calls between
+    // count_unread_samples() and and take_next() because we would run the risk
+    // of a deadlock. This is because count_unread_samples() is supposed to be
+    // called from within the reader's "exclusive area" (since it is
+    // executed within a listener callback), while take_next() is usually called
+    // from an executor/application thread and it must acquire the reader's
+    // "exclusive area" before taking samples.
+    // This might mean that an application which relies on data callbacks and the
+    // count provided by this function, and which at the same time polls data
+    // on the subscription, might end up missing some samples in the total count
+    // notified to it via callback because the samples were taken out of the
+    // cache before they could be notified to the listener.
+    // Fortunately, in Connext the listener callback is notified right after a
+    // sample is added to the reader cache and before any mutex is released,
+    // which should allow this function to always report a correct number, as
+    // long as it is only called within a (DDS) listener callback.
+    return rmw_connextdds_count_unread_samples(this, unread_count);
+  }
+
   DDS_Subscriber * dds_subscriber() const
   {
     return DDS_DataReader_get_subscriber(this->dds_reader);

rmw_connextdds_common/include/rmw_connextdds/rmw_waitset_std.hpp

@@ -123,7 +123,7 @@ class RMW_Connext_Condition
 
   template<typename FunctorT, typename FunctorA>
   void
-  perform_action_and_update_state(FunctorT && update_condition, FunctorA && action)
+  perform_action_and_update_state(FunctorA && action, FunctorT && update_condition)
   {
     std::lock_guard<std::mutex> internal_lock(this->mutex_internal);
 
@@ -782,28 +782,7 @@ class RMW_Connext_SubscriberStatusCondition : public RMW_Connext_StatusCondition
     }
   }
 
-  void notify_new_data()
-  {
-    size_t unread_samples = 0;
-    std::unique_lock<std::mutex> lock_mutex(new_data_event_mutex_);
-    perform_action_and_update_state(
-      [this, &unread_samples]() {
-        if (unread_samples == 0) {
-          return;
-        }
-        if (new_data_event_cb_) {
-          new_data_event_cb_(data_event_user_data_, unread_samples);
-        } else {
-          unread_data_events_count_ += unread_samples;
-        }
-      },
-      [this, &unread_samples]() {
-        const rmw_ret_t rc = rmw_connextdds_count_unread_samples(this->sub, unread_samples);
-        if (RMW_RET_OK != rc) {
-          RMW_CONNEXT_LOG_ERROR("failed to count unread samples on DDS Reader")
-        }
-      });
-  }
+  void notify_new_data();
 
 protected:
   void update_status_deadline(

rmw_connextdds_common/src/common/rmw_impl_waitset_std.cpp

@@ -750,6 +750,30 @@ RMW_Connext_SubscriberStatusCondition::update_status_sample_lost(
     this->status_sample_lost_last.total_count;
 }
 
+void
+RMW_Connext_SubscriberStatusCondition::notify_new_data()
+{
+  size_t unread_samples = 0;
+  std::unique_lock<std::mutex> lock_mutex(new_data_event_mutex_);
+  perform_action_and_update_state(
+    [this, &unread_samples]() {
+      const rmw_ret_t rc = this->sub->count_unread_samples(unread_samples);
+      if (RMW_RET_OK != rc) {
+        RMW_CONNEXT_LOG_ERROR("failed to count unread samples on DDS Reader")
+      }
+    },
+    [this, &unread_samples]() {
+      if (unread_samples == 0) {
+        return;
+      }
+      if (new_data_event_cb_) {
+        new_data_event_cb_(data_event_user_data_, unread_samples);
+      } else {
+        unread_data_events_count_ += unread_samples;
+      }
+    });
+}
+
 rmw_ret_t
 RMW_Connext_PublisherStatusCondition::install(
   RMW_Connext_Publisher * const pub)