Fix ABBA deadlock.

onNewDataMessage() is called as a callback from Fast-RTPS when
there is new data.  When that happens, some locks inside of
Fast-RTPS are taken, followed by internalMutex_ in SubListener.
However, the rest of SubListener takes internalMutex_ first,
followed by calls into Fast-RTPS where locks are taken.  This
is an ABBA deadlock.  Break the deadlock by doing the calls
into Fast-RTPS before taking the internalMutex_.

Signed-off-by: Chris Lalancette <clalancette@openrobotics.org>

rmw_fastrtps_shared_cpp/include/rmw_fastrtps_shared_cpp/custom_subscriber_info.hpp

@@ -78,13 +78,17 @@ class SubListener : public EventListenerInterface, public eprosima::fastrtps::Su
   void
   onNewDataMessage(eprosima::fastrtps::Subscriber * sub) final
   {
+    // Make sure to call into Fast-RTPS before taking the lock to avoid an
+    // ABBA deadlock between internalMutex_ and mutexes inside of Fast-RTPS.
+    uint64_t unread_count = sub->getUnreadCount();
+
     std::lock_guard<std::mutex> lock(internalMutex_);
 
     // the change to liveliness_lost_count_ needs to be mutually exclusive with
     // rmw_wait() which checks hasEvent() and decides if wait() needs to be called
     ConditionalScopedLock clock(conditionMutex_, conditionVariable_);
 
-    data_.store(sub->getUnreadCount(), std::memory_order_relaxed);
+    data_.store(unread_count, std::memory_order_relaxed);
   }
 
   RMW_FASTRTPS_SHARED_CPP_PUBLIC
@@ -134,9 +138,13 @@ class SubListener : public EventListenerInterface, public eprosima::fastrtps::Su
   void
   data_taken(eprosima::fastrtps::Subscriber * sub)
   {
+    // Make sure to call into Fast-RTPS before taking the lock to avoid an
+    // ABBA deadlock between internalMutex_ and mutexes inside of Fast-RTPS.
+    uint64_t unread_count = sub->getUnreadCount();
+
     std::lock_guard<std::mutex> lock(internalMutex_);
     ConditionalScopedLock clock(conditionMutex_);
-    data_.store(sub->getUnreadCount(), std::memory_order_relaxed);
+    data_.store(unread_count, std::memory_order_relaxed);
   }
 
   size_t publisherCount()