TCP Client&Server Participant Decision-Making (#4277) (#4386)

* TCP Client&Server Participant Decision-Making (#4277) * Refs #20180: Spelling fix Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Create function to check pending_logical_ports Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Add new TCP Connection type Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Add new method to enable TCP channel after connection in LARGE DATA Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Refactor OpenOutputChannel method Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Include LARGE DATA logic in SocketAccepted method Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Include LARGE DATA logic in processBindConnectionRequest method Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Add new TCP connection and methods SECURITY Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Add SECURITY support to OpenOutputChannel Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Include LARGE DATA logic in SecureSocketAccepted method Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Bugfix setting environment variable in tests Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Add multicast IPv6 to LARGE_DATA locator Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Add TCP test in LARGE_DATA Topology Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Clean code & Uncrustify Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Modified participant populated test Signed-off-by: Jesus Perez <jesusperez@eprosima.com> * Refs #20180: Add new api to update channels Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Refactor OpenOutputChannel with new channel map Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Refactor bind_socket with new channel map Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Refactor SocketAccepted with new channel map (Revert to old behavior) Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Refactor CloseOutputChannel to support cases with new channel map Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Update Secure channel logic with new channel map Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Use same IPv6 as defaultMetatrafficMulticast Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Refactor using locator inside sender resource instead of channel Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Data races associated Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Apply suggestions to tests Signed-off-by: Jesus Perez <jesusperez@eprosima.com> * Refs #20180: Add OpenLogicalPortRequest scenario before Bind Response is processed Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180. Remove unnecessary methods. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #20180. Early exit when logical_port is 0. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #20180. Refactor to avoid lambda. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #20180. Minor changes. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #20180. Simplify with WanToLanLocator. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #20180: Apply suggestions Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Revert "Refs #20180: Add new TCP Connection type" This reverts commit 4b4eb4153553cfdbf007896ff524d3e904f722c9. Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Remove unnecessary lock Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Revert "Refs #20180: Add new TCP connection and methods SECURITY" This reverts commit c69d92e56dcfab3c3c7287145a7a93e57947d3cb. Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180: Fix test (file descriptors limit) Signed-off-by: Jesus Perez <jesusperez@eprosima.com> * Refs #20180: Uncrustify Signed-off-by: Jesus Perez <jesusperez@eprosima.com> * Refs #20180. Fix port comparison. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> Co-authored-by: Carlos Ferreira González <carlosferreira@eprosima.com> * Refs #20180: Fix double lock and update old test with current behavior Signed-off-by: Jesus Perez <jesusperez@eprosima.com> * Refs #20180: Apply suggestions Signed-off-by: Jesus Perez <jesusperez@eprosima.com> * Refs #20180: Fix windows test build Signed-off-by: Jesus Perez <jesusperez@eprosima.com> * Refs #20180: Apply test suggestions Signed-off-by: Jesus Perez <jesusperez@eprosima.com> * Refs #20180: Bugfix with cv in receiver_resources_ destruction Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20180. Delete participant on failing ParticipantTests.ParticipantCreationWithBuiltinTransport Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #20180. Do not use non_blocking_send on TCP transport for LARGE_DATA. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> --------- Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> Signed-off-by: Jesus Perez <jesusperez@eprosima.com> Signed-off-by: Miguel Company <miguelcompany@eprosima.com> Co-authored-by: Jesus Perez <jesusperez@eprosima.com> Co-authored-by: Miguel Company <miguelcompany@eprosima.com> (cherry picked from commit 709b140) * Fix Conflicts Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> --------- Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> Co-authored-by: Carlos Ferreira González <carlosferreira@eprosima.com>
eProsima · Apr 1, 2024 · 74315c2 · 74315c2
1 parent b29c6c0
commit 74315c2
Show file tree

Hide file tree

Showing 19 changed files with 553 additions and 122 deletions.
diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.cpp b/src/cpp/fastdds/publisher/DataWriterImpl.cpp
@@ -1055,6 +1055,7 @@ const DataWriterQos& DataWriterImpl::get_qos() const
 ReturnCode_t DataWriterImpl::set_listener(
         DataWriterListener* listener)
 {
+    std::lock_guard<std::mutex> scoped_lock(listener_mutex_);
     listener_ = listener;
     return ReturnCode_t::RETCODE_OK;
 }
@@ -1779,6 +1780,7 @@ bool DataWriterImpl::can_qos_be_updated(
 DataWriterListener* DataWriterImpl::get_listener_for(
         const StatusMask& status)
 {
+    std::lock_guard<std::mutex> scoped_lock(listener_mutex_);
     if (listener_ != nullptr &&
             user_datawriter_->get_status_mask().is_active(status))
     {

diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.hpp b/src/cpp/fastdds/publisher/DataWriterImpl.hpp
@@ -340,6 +340,9 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
     //! DataWriterListener
     DataWriterListener* listener_ = nullptr;
 
+    //! Mutex to protect listener_
+    std::mutex listener_mutex_;
+
     //!Listener to capture the events of the Writer
     class InnerDataWriterListener : public fastrtps::rtps::WriterListener
     {

diff --git a/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp b/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp
@@ -198,7 +198,6 @@ static void setup_transports_large_data(
 
         auto tcp_transport = create_tcpv4_transport(att);
         att.userTransports.push_back(tcp_transport);
-        att.properties.properties().emplace_back("fastdds.tcp_transport.non_blocking_send", "true");
 
         Locator_t tcp_loc;
         tcp_loc.kind = LOCATOR_KIND_TCPv4;
@@ -235,7 +234,6 @@ static void setup_transports_large_datav6(
 
         auto tcp_transport = create_tcpv6_transport(att);
         att.userTransports.push_back(tcp_transport);
-        att.properties.properties().emplace_back("fastdds.tcp_transport.non_blocking_send", "true");
 
         Locator_t tcp_loc;
         tcp_loc.kind = LOCATOR_KIND_TCPv6;
@@ -253,6 +251,7 @@ static void setup_transports_large_datav6(
     {
         Locator_t pdp_locator;
         pdp_locator.kind = LOCATOR_KIND_UDPv6;
+        IPLocator::setIPv6(pdp_locator, "ff1e::ffff:efff:1");
         att.builtin.metatrafficMulticastLocatorList.push_back(pdp_locator);
     }
 }

diff --git a/src/cpp/rtps/transport/TCPChannelResource.cpp b/src/cpp/rtps/transport/TCPChannelResource.cpp
@@ -84,7 +84,7 @@ ResponseCode TCPChannelResource::process_bind_request(
     if (connection_status_.compare_exchange_strong(expected, eConnectionStatus::eEstablished))
     {
         locator_ = IPLocator::toPhysicalLocator(locator);
-        logInfo(RTCP_MSG, "Connection Stablished");
+        logInfo(RTCP_MSG, "Connection Established");
         return RETCODE_OK;
     }
     else if (expected == eConnectionStatus::eEstablished)
@@ -139,9 +139,7 @@ void TCPChannelResource::add_logical_port(
             pending_logical_output_ports_.emplace_back(port);
             if (connection_established())
             {
-                scopedLock.unlock();
                 TCPTransactionId id = rtcp_manager->sendOpenLogicalPortRequest(this, port);
-                scopedLock.lock();
                 negotiating_logical_ports_[id] = port;
             }
         }

diff --git a/src/cpp/rtps/transport/TCPChannelResourceBasic.cpp b/src/cpp/rtps/transport/TCPChannelResourceBasic.cpp
@@ -105,6 +105,7 @@ void TCPChannelResourceBasic::disconnect()
 {
     if (eConnecting < change_status(eConnectionStatus::eDisconnected) && alive())
     {
+        std::lock_guard<std::mutex> read_lock(read_mutex_);
         auto socket = socket_;
 
         std::error_code ec;

diff --git a/src/cpp/rtps/transport/TCPSenderResource.hpp b/src/cpp/rtps/transport/TCPSenderResource.hpp
@@ -32,14 +32,14 @@ class TCPSenderResource : public fastrtps::rtps::SenderResource
 
     TCPSenderResource(
             TCPTransportInterface& transport,
-            std::shared_ptr<TCPChannelResource>& channel)
+            eprosima::fastrtps::rtps::Locator_t& locator)
         : fastrtps::rtps::SenderResource(transport.kind())
-        , channel_(channel)
+        , locator_(locator)
     {
         // Implementation functions are bound to the right transport parameters
         clean_up = [this, &transport]()
                 {
-                    transport.CloseOutputChannel(channel_);
+                    transport.CloseOutputChannel(locator_);
                 };
 
         send_lambda_ = [this, &transport](
@@ -49,7 +49,7 @@ class TCPSenderResource : public fastrtps::rtps::SenderResource
             fastrtps::rtps::LocatorsIterator* destination_locators_end,
             const std::chrono::steady_clock::time_point&) -> bool
                 {
-                    return transport.send(data, dataSize, channel_, destination_locators_begin,
+                    return transport.send(data, dataSize, locator_, destination_locators_begin,
                                    destination_locators_end);
                 };
     }
@@ -62,9 +62,9 @@ class TCPSenderResource : public fastrtps::rtps::SenderResource
         }
     }
 
-    std::shared_ptr<TCPChannelResource>& channel()
+    fastrtps::rtps::Locator_t& locator()
     {
-        return channel_;
+        return locator_;
     }
 
     static TCPSenderResource* cast(
@@ -102,7 +102,7 @@ class TCPSenderResource : public fastrtps::rtps::SenderResource
     TCPSenderResource& operator =(
             const SenderResource&) = delete;
 
-    std::shared_ptr<TCPChannelResource> channel_;
+    fastrtps::rtps::Locator_t locator_;
 };
 
 } // namespace rtps