Fixing data race on UDPTransportInterface [5458] (#540)

* Refs #5458. Fixing data race on UDPTransportInterface.

* Refs #5458. Fixing timeouts on Mac.

* Refs #5458 Fixing data race on ReceiverResource dtor.

* Refs #5458 Clang-tidy warnings

* Refs #5458 Fixing Mac timeout.

* Refs #5458 Fixing valgrind invalid read.

* Refs #5458 Requested changes.

* Refs #5458 Fixing situation when socket doesn't close nicely.

* Refs #5458 Trying to fix Mac issue.

* Refs #5458 Trying to fix Mac issue.

* Refs #5458 Trying to fix Mac issue.

* Refs #5458 Fixed Mac issue.

* Refs #5458 Avoid loop when multicast. Much lesser wait time until continue.

* Refs #5458 Requested changes.

Author: MiguelCompany

include/fastrtps/rtps/network/ReceiverResource.h

@@ -68,12 +68,18 @@ class ReceiverResource : public TransportReceiverInterface
     */
     void UnregisterReceiver(MessageReceiver* receiver);
 
+    /**
+     * Closes related ChannelResources.
+     */
+    void disable();
+
     /**
      * Resources can only be transfered through move semantics. Copy, assignment, and
      * construction outside of the factory are forbidden.
      */
     ReceiverResource(ReceiverResource&&);
-    ~ReceiverResource();
+
+    ~ReceiverResource() override;
 
 private:
     ReceiverResource() = delete;

include/fastrtps/transport/UDPChannelResource.h

@@ -16,13 +16,15 @@
 #define UDP_CHANNEL_RESOURCE_INFO_
 
 #include <fastrtps/transport/ChannelResource.h>
+#include <fastrtps/rtps/common/Locator.h>
 #include <asio.hpp>
 
 namespace eprosima{
 namespace fastrtps{
 namespace rtps{
 
 class TransportReceiverInterface;
+class UDPTransportInterface;
 
 #if defined(ASIO_HAS_MOVE)
     // Typedefs
@@ -73,14 +75,16 @@ class TransportReceiverInterface;
 class UDPChannelResource : public ChannelResource
 {
 public:
-    UDPChannelResource(eProsimaUDPSocket& socket);
 
     UDPChannelResource(
+        UDPTransportInterface* transport,
         eProsimaUDPSocket& socket,
-        uint32_t maxMsgSize);
+        uint32_t maxMsgSize,
+        const Locator_t& locator,
+        const std::string& sInterface,
+        TransportReceiverInterface* receiver);
 
-    UDPChannelResource(UDPChannelResource&& channelResource);
-    virtual ~UDPChannelResource();
+    virtual ~UDPChannelResource() override;
 
     UDPChannelResource& operator=(UDPChannelResource&& channelResource)
     {
@@ -91,7 +95,7 @@ class UDPChannelResource : public ChannelResource
     void only_multicast_purpose(const bool value)
     {
         only_multicast_purpose_ = value;
-    };
+    }
 
     bool& only_multicast_purpose()
     {
@@ -132,12 +136,49 @@ class UDPChannelResource : public ChannelResource
         return message_receiver_;
     }
 
+    inline virtual void disable() override
+    {
+        ChannelResource::disable();
+    }
+
+    void release(
+            const Locator_t& locator,
+            const asio::ip::address& address);
+
+protected:
+    /**
+     * Function to be called from a new thread, which takes cares of performing a blocking receive
+     * operation on the ReceiveResource
+     * @param input_locator - Locator that triggered the creation of the resource
+    */
+    void perform_listen_operation(
+            Locator_t input_locator);
+
+    /**
+    * Blocking Receive from the specified channel.
+    * @param receive_buffer vector with enough capacity (not size) to accomodate a full receive buffer. That
+    * capacity must not be less than the receive_buffer_size supplied to this class during construction.
+    * @param receive_buffer_capacity Maximum size of the receive_buffer.
+    * @param[out] receive_buffer_size Size of the received buffer.
+    * @param[out] remote_locator Locator describing the remote restination we received a packet from.
+    */
+    bool Receive(
+            octet* receive_buffer,
+            uint32_t receive_buffer_capacity,
+            uint32_t& receive_buffer_size,
+            Locator_t& remote_locator);
+
 private:
 
     TransportReceiverInterface* message_receiver_; //Associated Readers/Writers inside of MessageReceiver
     eProsimaUDPSocket socket_;
     bool only_multicast_purpose_;
     std::string interface_;
+    UDPTransportInterface* transport_;
+    bool closing_;
+    std::mutex mtx_closing_;
+    std::condition_variable cv_closing_;
+
     UDPChannelResource(const UDPChannelResource&) = delete;
     UDPChannelResource& operator=(const UDPChannelResource&) = delete;
 };

include/fastrtps/transport/UDPTransportInterface.h

@@ -64,18 +64,6 @@ class UDPTransportInterface : public TransportInterface
            SendResourceList& sender_resource_list,
            const Locator_t&) override;
 
-   /**
-   * Blocking Receive from the specified channel.
-   * @param p_channel_resource Pointer to the channer resource that stores the socket.
-   * @param receive_buffer vector with enough capacity (not size) to accomodate a full receive buffer. That
-   * capacity must not be less than the receive_buffer_size supplied to this class during construction.
-   * @param receive_buffer_capacity Maximum size of the receive_buffer.
-   * @param[out] receive_buffer_size Size of the received buffer.
-   * @param[out] remote_locator Locator describing the remote restination we received a packet from.
-   */
-   bool Receive(UDPChannelResource* p_channel_resource, octet* receive_buffer,
-       uint32_t receive_buffer_capacity, uint32_t& receive_buffer_size, Locator_t& remote_locator);
-
    //! Release the listening socket for the specified port.
    bool ReleaseInputChannel(const Locator_t& locator, const asio::ip::address& interface_address);
 
@@ -117,6 +105,8 @@ class UDPTransportInterface : public TransportInterface
 
 protected:
 
+    friend class UDPChannelResource;
+
     // For UDPv6, the notion of channel corresponds to a port + direction tuple.
     asio::io_service io_service_;
     std::vector<IPFinder::info_IP> currentInterfaces;
@@ -162,14 +152,6 @@ class UDPTransportInterface : public TransportInterface
     virtual eProsimaUDPSocket OpenAndBindInputSocket(const std::string& sIp, uint16_t port, bool is_multicast) = 0;
     eProsimaUDPSocket OpenAndBindUnicastOutputSocket(const asio::ip::udp::endpoint& endpoint, uint16_t& port);
 
-    /**
-     * Function to be called from a new thread, which takes cares of performing a blocking receive
-     * operation on the ReceiveResource
-     * @param p_channel_resource - Associated ChannelResource
-     * @param input_locator - Locator that triggered the creation of the resource
-    */
-    void perform_listen_operation(UDPChannelResource* p_channel_resource, Locator_t input_locator);
-
     virtual void set_receive_buffer_size(uint32_t size) = 0;
     virtual void set_send_buffer_size(uint32_t size) = 0;
     virtual void SetSocketOutboundInterface(eProsimaUDPSocket&, const std::string&) = 0;

src/cpp/rtps/network/ReceiverResource.cpp

@@ -26,7 +26,10 @@ namespace fastrtps{
 namespace rtps{
 
 ReceiverResource::ReceiverResource(TransportInterface& transport, const Locator_t& locator, uint32_t max_size)
-        : mValid(false)
+        : Cleanup(nullptr)
+        , LocatorMapsToManagedChannel(nullptr)
+        , mValid(false)
+        , mtx()
         , receiver(nullptr)
         , msg(0)
 {
@@ -98,14 +101,18 @@ void ReceiverResource::OnDataReceived(const octet * data, const uint32_t size,
 
 }
 
-ReceiverResource::~ReceiverResource()
+void ReceiverResource::disable()
 {
     if (Cleanup)
     {
         Cleanup();
     }
 }
 
+ReceiverResource::~ReceiverResource()
+{
+}
+
 } // namespace rtps
 } // namespace fastrtps
 } // namespace eprosima

src/cpp/rtps/participant/RTPSParticipantImpl.cpp

@@ -122,7 +122,8 @@ RTPSParticipantImpl::RTPSParticipantImpl(const RTPSParticipantAttributes& PParam
 
     /// Creation of metatraffic locator and receiver resources
     uint32_t metatraffic_multicast_port = m_att.port.getMulticastPort(m_att.builtin.domainId);
-    uint32_t metatraffic_unicast_port = m_att.port.getUnicastPort(m_att.builtin.domainId, m_att.participantID);
+    uint32_t metatraffic_unicast_port = m_att.port.getUnicastPort(m_att.builtin.domainId,
+                                                                  static_cast<uint32_t>(m_att.participantID));
 
     /* If metatrafficMulticastLocatorList is empty, add mandatory default Locators
        Else -> Take them */
@@ -249,16 +250,17 @@ RTPSParticipantImpl::~RTPSParticipantImpl()
     for(auto& block : m_receiverResourcelist)
     {
         block.Receiver->UnregisterReceiver(block.mp_receiver);
+        block.disable();
     }
 
     while(m_userReaderList.size() > 0)
     {
-        deleteUserEndpoint((Endpoint*)*m_userReaderList.begin());
+        deleteUserEndpoint(static_cast<Endpoint*>(*m_userReaderList.begin()));
     }
 
     while(m_userWriterList.size() > 0)
     {
-        deleteUserEndpoint((Endpoint*)*m_userWriterList.begin());
+        deleteUserEndpoint(static_cast<Endpoint*>(*m_userWriterList.begin()));
     }
 
     delete(this->mp_builtinProtocols);
@@ -298,7 +300,7 @@ bool RTPSParticipantImpl::createWriter(
     std::string type = (param.endpoint.reliabilityKind == RELIABLE) ? "RELIABLE" :"BEST_EFFORT";
     logInfo(RTPS_PARTICIPANT," of type " << type);
     EntityId_t entId;
-    if(entityId== c_EntityId_Unknown)
+    if(entityId == c_EntityId_Unknown)
     {
         if(param.endpoint.topicKind == NO_KEY)
         {
@@ -309,21 +311,21 @@ bool RTPSParticipantImpl::createWriter(
             entId.value[3] = 0x02;
         }
         uint32_t idnum;
-        if(param.endpoint.getEntityID()>0)
+        if(param.endpoint.getEntityID() > 0)
         {
-            idnum = param.endpoint.getEntityID();
+            idnum = static_cast<uint32_t>(param.endpoint.getEntityID());
         }
         else
         {
             IdCounter++;
             idnum = IdCounter;
         }
 
-        octet* c = (octet*)&idnum;
+        octet* c = reinterpret_cast<octet*>(&idnum);
         entId.value[2] = c[0];
         entId.value[1] = c[1];
         entId.value[0] = c[2];
-        if(this->existsEntityId(entId,WRITER))
+        if(this->existsEntityId(entId, WRITER))
         {
             logError(RTPS_PARTICIPANT,"A writer with the same entityId already exists in this RTPSParticipant");
             return false;
@@ -375,14 +377,14 @@ bool RTPSParticipantImpl::createWriter(
     if (param.endpoint.reliabilityKind == BEST_EFFORT)
     {
         SWriter = (persistence == nullptr) ?
-            (RTPSWriter*) new StatelessWriter(this, guid, param, hist, listen) :
-            (RTPSWriter*) new StatelessPersistentWriter(this, guid, param, hist, listen, persistence);
+            new StatelessWriter(this, guid, param, hist, listen) :
+            new StatelessPersistentWriter(this, guid, param, hist, listen, persistence);
     }
     else if (param.endpoint.reliabilityKind == RELIABLE)
     {
         SWriter = (persistence == nullptr) ?
-            (RTPSWriter*) new StatefulWriter(this, guid, param, hist, listen) :
-            (RTPSWriter*) new StatefulPersistentWriter(this, guid, param, hist, listen, persistence);
+            new StatefulWriter(this, guid, param, hist, listen) :
+            new StatefulPersistentWriter(this, guid, param, hist, listen, persistence);
     }
 
     if (SWriter == nullptr)
@@ -411,10 +413,10 @@ bool RTPSParticipantImpl::createWriter(
     }
 #endif
 
-    createSendResources((Endpoint *)SWriter);
+    createSendResources(SWriter);
     if (param.endpoint.reliabilityKind == RELIABLE)
     {
-        if (!createAndAssociateReceiverswithEndpoint((Endpoint *)SWriter))
+        if (!createAndAssociateReceiverswithEndpoint(SWriter))
         {
             delete(SWriter);
             return false;
@@ -469,15 +471,15 @@ bool RTPSParticipantImpl::createReader(
         uint32_t idnum;
         if (param.endpoint.getEntityID() > 0)
         {
-            idnum = param.endpoint.getEntityID();
+            idnum = static_cast<uint32_t>(param.endpoint.getEntityID());
         }
         else
         {
             IdCounter++;
             idnum = IdCounter;
         }
 
-        octet* c = (octet*)&idnum;
+        octet* c = reinterpret_cast<octet*>(&idnum);
         entId.value[2] = c[0];
         entId.value[1] = c[1];
         entId.value[0] = c[2];
@@ -526,14 +528,14 @@ bool RTPSParticipantImpl::createReader(
     if (param.endpoint.reliabilityKind == BEST_EFFORT)
     {
         SReader = (persistence == nullptr) ?
-            (RTPSReader*) new StatelessReader(this, guid, param, hist, listen) :
-            (RTPSReader*) new StatelessPersistentReader(this, guid, param, hist, listen, persistence);
+            new StatelessReader(this, guid, param, hist, listen) :
+            new StatelessPersistentReader(this, guid, param, hist, listen, persistence);
     }
     else if (param.endpoint.reliabilityKind == RELIABLE)
     {
         SReader = (persistence == nullptr) ?
-            (RTPSReader*) new StatefulReader(this, guid, param, hist, listen) :
-            (RTPSReader*) new StatefulPersistentReader(this, guid, param, hist, listen, persistence);
+            new StatefulReader(this, guid, param, hist, listen) :
+            new StatefulPersistentReader(this, guid, param, hist, listen, persistence);
     }
 
     if (SReader == nullptr)
@@ -565,7 +567,7 @@ bool RTPSParticipantImpl::createReader(
 
     if (param.endpoint.reliabilityKind == RELIABLE)
     {
-        createSendResources((Endpoint *)SReader);
+        createSendResources(SReader);
     }
 
     if (isBuiltin)
@@ -575,7 +577,7 @@ bool RTPSParticipantImpl::createReader(
 
     if (enable)
     {
-        if (!createAndAssociateReceiverswithEndpoint((Endpoint *)SReader))
+        if (!createAndAssociateReceiverswithEndpoint(SReader))
         {
             delete(SReader);
             return false;
@@ -594,7 +596,7 @@ bool RTPSParticipantImpl::createReader(
 
 bool RTPSParticipantImpl::enableReader(RTPSReader *reader)
 {
-    if (!assignEndpointListenResources((Endpoint*)reader))
+    if (!assignEndpointListenResources(reader))
     {
         return false;
     }
@@ -794,7 +796,7 @@ void RTPSParticipantImpl::createReceiverResources(LocatorList_t& Locator_list, b
         {
             std::lock_guard<std::mutex> lock(m_receiverResourcelistMutex);
             //Push the new items into the ReceiverResource buffer
-            m_receiverResourcelist.push_back(ReceiverControlBlock(std::move(*it_buffer)));
+            m_receiverResourcelist.emplace_back(*it_buffer);
             //Create and init the MessageReceiver
             auto mr = new MessageReceiver(this, size);
             m_receiverResourcelist.back().mp_receiver = mr;
@@ -892,7 +894,7 @@ bool RTPSParticipantImpl::deleteUserEndpoint(Endpoint* p_endpoint)
         {
             if (found_in_users)
             {
-                mp_builtinProtocols->removeLocalWriter((RTPSWriter*)p_endpoint);
+                mp_builtinProtocols->removeLocalWriter(static_cast<RTPSWriter*>(p_endpoint));
             }
 
 #if HAVE_SECURITY
@@ -907,7 +909,7 @@ bool RTPSParticipantImpl::deleteUserEndpoint(Endpoint* p_endpoint)
         {
             if (found_in_users)
             {
-                mp_builtinProtocols->removeLocalReader((RTPSReader*)p_endpoint);
+                mp_builtinProtocols->removeLocalReader(static_cast<RTPSReader*>(p_endpoint));
             }
 
 #if HAVE_SECURITY