Fix heartbeat and ACKs issues (#1865)

* Refs 11001. Improve blackbox test by waiting for ACKs. Signed-off-by: Miguel Company <MiguelCompany@eprosima.com> * Refs 11001. Fix SharedMemTransportDescriptor::min_send_buffer_size() Signed-off-by: Miguel Company <MiguelCompany@eprosima.com> * Refs 11001. Removed unused local variable. Signed-off-by: Miguel Company <MiguelCompany@eprosima.com> * Refs 11001. Sender resources created after starting WriterProxy. Signed-off-by: Miguel Company <MiguelCompany@eprosima.com> * Refs 11001. Uncrustify ReceiverResource.cpp Signed-off-by: Miguel Company <MiguelCompany@eprosima.com> * Refs 11001. Fix locator support check on ReceiverResource. Signed-off-by: Miguel Company <MiguelCompany@eprosima.com> * Refs 11001. Uncrustify. Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>
eProsima · Mar 29, 2021 · 6cee7cd · 6cee7cd
1 parent d3139db
commit 6cee7cd
Show file tree

Hide file tree

Showing 5 changed files with 52 additions and 35 deletions.
diff --git a/include/fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h b/include/fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h
@@ -30,17 +30,17 @@ class TransportInterface;
  */
 typedef struct SharedMemTransportDescriptor : public TransportDescriptorInterface
 {
-    virtual ~SharedMemTransportDescriptor() 
+    virtual ~SharedMemTransportDescriptor()
     {
-        
+
     }
 
     virtual TransportInterface* create_transport() const override;
     uint32_t min_send_buffer_size() const override
     {
-        return 0;
+        return segment_size_;
     }
-    
+
     RTPS_DllAPI SharedMemTransportDescriptor();
 
     RTPS_DllAPI SharedMemTransportDescriptor(
@@ -50,16 +50,16 @@ typedef struct SharedMemTransportDescriptor : public TransportDescriptorInterfac
     {
         return segment_size_;
     }
-    
+
     RTPS_DllAPI void segment_size(
             uint32_t segment_size)
     {
         segment_size_ = segment_size;
     }
 
     virtual uint32_t max_message_size() const override
-    { 
-        return maxMessageSize; 
+    {
+        return maxMessageSize;
     }
 
     RTPS_DllAPI void max_message_size(

diff --git a/src/cpp/rtps/network/ReceiverResource.cpp b/src/cpp/rtps/network/ReceiverResource.cpp
@@ -17,25 +17,25 @@
 #include <cassert>
 #include <fastdds/dds/log/Log.hpp>
 
-#define IDSTRING "(ID:" << std::this_thread::get_id() <<") "<<
+#define IDSTRING "(ID:" << std::this_thread::get_id() << ") " <<
 
 using namespace std;
 using namespace eprosima::fastdds::rtps;
 
-namespace eprosima{
-namespace fastrtps{
-namespace rtps{
+namespace eprosima {
+namespace fastrtps {
+namespace rtps {
 
 ReceiverResource::ReceiverResource(
-			TransportInterface& transport,
-			const Locator_t& locator,
-			uint32_t max_recv_buffer_size)
-        : Cleanup(nullptr)
-        , LocatorMapsToManagedChannel(nullptr)
-        , mValid(false)
-        , mtx()
-        , receiver(nullptr)
-        , max_message_size_(max_recv_buffer_size)
+        TransportInterface& transport,
+        const Locator_t& locator,
+        uint32_t max_recv_buffer_size)
+    : Cleanup(nullptr)
+    , LocatorMapsToManagedChannel(nullptr)
+    , mValid(false)
+    , mtx()
+    , receiver(nullptr)
+    , max_message_size_(max_recv_buffer_size)
 {
     // Internal channel is opened and assigned to this resource.
     mValid = transport.OpenInputChannel(locator, this, max_message_size_);
@@ -45,12 +45,18 @@ ReceiverResource::ReceiverResource(
     }
 
     // Implementation functions are bound to the right transport parameters
-    Cleanup = [&transport, locator]() { transport.CloseInputChannel(locator); };
+    Cleanup = [&transport, locator]()
+            {
+                transport.CloseInputChannel(locator);
+            };
     LocatorMapsToManagedChannel = [&transport, locator](const Locator_t& locatorToCheck) -> bool
-    { return transport.DoInputLocatorsMatch(locator, locatorToCheck); };
+            {
+                return locator.kind == locatorToCheck.kind && transport.DoInputLocatorsMatch(locator, locatorToCheck);
+            };
 }
 
-ReceiverResource::ReceiverResource(ReceiverResource&& rValueResource)
+ReceiverResource::ReceiverResource(
+        ReceiverResource&& rValueResource)
 {
     Cleanup.swap(rValueResource.Cleanup);
     LocatorMapsToManagedChannel.swap(rValueResource.LocatorMapsToManagedChannel);
@@ -61,7 +67,8 @@ ReceiverResource::ReceiverResource(ReceiverResource&& rValueResource)
     max_message_size_ = rValueResource.max_message_size_;
 }
 
-bool ReceiverResource::SupportsLocator(const Locator_t& localLocator)
+bool ReceiverResource::SupportsLocator(
+        const Locator_t& localLocator)
 {
     if (LocatorMapsToManagedChannel)
     {
@@ -70,22 +77,31 @@ bool ReceiverResource::SupportsLocator(const Locator_t& localLocator)
     return false;
 }
 
-void ReceiverResource::RegisterReceiver(MessageReceiver* rcv)
+void ReceiverResource::RegisterReceiver(
+        MessageReceiver* rcv)
 {
     std::unique_lock<std::mutex> lock(mtx);
     if (receiver == nullptr)
+    {
         receiver = rcv;
+    }
 }
 
-void ReceiverResource::UnregisterReceiver(MessageReceiver* rcv)
+void ReceiverResource::UnregisterReceiver(
+        MessageReceiver* rcv)
 {
     std::unique_lock<std::mutex> lock(mtx);
     if (receiver == rcv)
+    {
         receiver = nullptr;
+    }
 }
 
-void ReceiverResource::OnDataReceived(const octet * data, const uint32_t size,
-    const Locator_t & localLocator, const Locator_t & remoteLocator)
+void ReceiverResource::OnDataReceived(
+        const octet* data,
+        const uint32_t size,
+        const Locator_t& localLocator,
+        const Locator_t& remoteLocator)
 {
     (void)localLocator;
 

diff --git a/src/cpp/rtps/participant/RTPSParticipantImpl.cpp b/src/cpp/rtps/participant/RTPSParticipantImpl.cpp
@@ -952,7 +952,6 @@ bool RTPSParticipantImpl::assignEndpoint2LocatorList(
        one of the supported Locators is needed to make the match, and the case of new ListenResources being created has been removed
        since its the NetworkFactory the one that takes care of Resource creation.
      */
-    LocatorList_t finalList;
     for (auto lit = list.begin(); lit != list.end(); ++lit)
     {
         //Iteration of all Locators within the Locator list passed down as argument

diff --git a/src/cpp/rtps/reader/StatefulReader.cpp b/src/cpp/rtps/reader/StatefulReader.cpp
@@ -141,6 +141,12 @@ bool StatefulReader::matched_writer_add(
         matched_writers_pool_.pop_back();
     }
 
+    SequenceNumber_t initial_sequence;
+    add_persistence_guid(wdata.guid(), wdata.persistence_guid());
+    initial_sequence = get_last_notified(wdata.guid());
+
+    wp->start(wdata, initial_sequence);
+
     if (!is_same_process)
     {
         for (const Locator_t& locator : wp->remote_locators_shrinked())
@@ -149,12 +155,6 @@ bool StatefulReader::matched_writer_add(
         }
     }
 
-    SequenceNumber_t initial_sequence;
-    add_persistence_guid(wdata.guid(), wdata.persistence_guid());
-    initial_sequence = get_last_notified(wdata.guid());
-
-    wp->start(wdata, initial_sequence);
-
     matched_writers_.push_back(wp);
 
     if (liveliness_lease_duration_ < c_TimeInfinite)

diff --git a/test/blackbox/BlackboxTestsPubSubBasic.cpp b/test/blackbox/BlackboxTestsPubSubBasic.cpp
@@ -253,6 +253,7 @@ TEST_P(PubSubBasic, PubSubMoreThan256Unacknowledged)
 
     reader.startReception(expected_data);
     reader.block_for_all();
+    EXPECT_TRUE(writer.waitForAllAcked(std::chrono::seconds(10)));
 }
 
 TEST_P(PubSubBasic, PubSubAsReliableHelloworldMulticastDisabled)
@@ -286,6 +287,7 @@ TEST_P(PubSubBasic, PubSubAsReliableHelloworldMulticastDisabled)
     ASSERT_TRUE(data.empty());
     // Block reader until reception finished or timeout.
     reader.block_for_all();
+    EXPECT_TRUE(writer.waitForAllAcked(std::chrono::seconds(10)));
 }
 
 TEST_P(PubSubBasic, ReceivedDynamicDataWithNoSizeLimit)