Solve TCP disconnect while incomplete read deadlock (#6066) (#6091)

* Refs #23655. Add regression test



* Refs #23655. Solve deadlock and improve thread safety



* Refs #23655. Bonus track: protect status getter



* Refs #23655. Fix test in Windows



* Refs #23655. Remove deprecated code



* Refs #23655. Restore noexcept overloads usage



* Refs #23655. Apply suggestions



---------


(cherry picked from commit f2e173c)

Signed-off-by: Juan Lopez Fernandez <[email protected]>
Co-authored-by: juanlofer-eprosima <[email protected]>

Author: mergify[bot]

src/cpp/fastdds/core/condition/StatusConditionImpl.cpp

@@ -70,6 +70,12 @@ const StatusMask& StatusConditionImpl::get_enabled_statuses() const
     return mask_;
 }
 
+const StatusMask& StatusConditionImpl::get_raw_status() const
+{
+    std::lock_guard<std::mutex> guard(mutex_);
+    return status_;
+}
+
 void StatusConditionImpl::set_status(
         const StatusMask& status,
         bool trigger_value)

src/cpp/fastdds/core/condition/StatusConditionImpl.hpp

@@ -78,10 +78,7 @@ struct StatusConditionImpl
      * @brief Retrieves the list of communication statuses that are currently triggered.
      * @return Triggered status.
      */
-    const StatusMask& get_raw_status() const
-    {
-        return status_;
-    }
+    const StatusMask& get_raw_status() const;
 
     /**
      * @brief Set the trigger value of a specific status

src/cpp/rtps/transport/TCPAcceptorSecure.cpp

@@ -55,7 +55,6 @@ void TCPAcceptorSecure::accept(
 
     try
     {
-#if ASIO_VERSION >= 101200
         acceptor_.async_accept(
             [locator, parent, &ssl_context](const std::error_code& error, tcp::socket socket)
             {
@@ -82,33 +81,6 @@ void TCPAcceptorSecure::accept(
                     parent->SecureSocketAccepted(nullptr, locator, error); // This method manages errors too.
                 }
             });
-#else
-        auto secure_socket = std::make_shared<asio::ssl::stream<asio::ip::tcp::socket>>(*io_context_, ssl_context);
-
-        acceptor_.async_accept(secure_socket->lowest_layer(),
-                [locator, parent, secure_socket](const std::error_code& error)
-                {
-                    if (!error)
-                    {
-                        ssl::stream_base::handshake_type role = ssl::stream_base::server;
-                        if (parent->configuration()->tls_config.handshake_role == TLSHSRole::CLIENT)
-                        {
-                            role = ssl::stream_base::client;
-                        }
-
-                        secure_socket->async_handshake(role,
-                        [secure_socket, locator, parent](const std::error_code& error)
-                        {
-                            //EPROSIMA_LOG_ERROR(RTCP_TLS, "Handshake: " << error.message());
-                            parent->SecureSocketAccepted(secure_socket, locator, error);
-                        });
-                    }
-                    else
-                    {
-                        parent->SecureSocketAccepted(nullptr, locator, error); // This method manages errors too.
-                    }
-                });
-#endif // if ASIO_VERSION >= 101200
     }
     catch (std::error_code& error)
     {

src/cpp/rtps/transport/TCPChannelResource.cpp

@@ -66,7 +66,7 @@ TCPChannelResource::TCPChannelResource(
     , parent_(parent)
     , locator_()
     , waiting_for_keep_alive_(false)
-    , connection_status_(eConnectionStatus::eConnected)
+    , connection_status_(eConnectionStatus::eDisconnected)
     , tcp_connection_type_(TCPConnectionType::TCP_ACCEPT_TYPE)
 {
 }
@@ -376,6 +376,9 @@ void TCPChannelResource::set_socket_options(
         asio::basic_socket<asio::ip::tcp>& socket,
         const TCPTransportDescriptor* options)
 {
+    // Options setting should be done before connection is established
+    assert(!connected());
+
     uint32_t minimum_value = options->maxMessageSize;
 
     // Set the send buffer size

src/cpp/rtps/transport/TCPChannelResource.h

@@ -53,7 +53,8 @@ class TCPChannelResource : public ChannelResource
 
     enum eConnectionStatus
     {
-        eDisconnected = 0,
+        eDisconnecting = -1,        // Transition state to disconnected, used to protect from concurrent (dis)connects.
+        eDisconnected = 0,          // Initial state.
         eConnecting,                // Output -> Trying connection.
         eConnected,                 // Output -> Send bind message.
         eWaitingForBind,            // Input -> Waiting for the bind message.
@@ -71,7 +72,6 @@ class TCPChannelResource : public ChannelResource
     std::vector<uint16_t> pending_logical_output_ports_; // Must be accessed after lock pending_logical_mutex_
     std::vector<uint16_t> logical_output_ports_;
     std::condition_variable_any logical_output_ports_updated_cv;
-    std::mutex read_mutex_;
     std::recursive_mutex pending_logical_mutex_;
     std::atomic<eConnectionStatus> connection_status_;
 
@@ -113,6 +113,17 @@ class TCPChannelResource : public ChannelResource
         return connection_status_ == eConnectionStatus::eEstablished;
     }
 
+    bool connected()
+    {
+        return connection_status_ >= eConnectionStatus::eConnected;
+    }
+
+    bool disconnected()
+    {
+        // NOTE: covers both eDisconnected and eDisconnecting states
+        return connection_status_ <= eConnectionStatus::eDisconnected;
+    }
+
     eConnectionStatus connection_status()
     {
         return connection_status_;
@@ -275,7 +286,7 @@ class TCPChannelResource : public ChannelResource
      * @param socket Socket on which to set the options.
      * @param options Descriptor with the options to set.
      */
-    static void set_socket_options(
+    void set_socket_options(
             asio::basic_socket<asio::ip::tcp>& socket,
             const TCPTransportDescriptor* options);
 

src/cpp/rtps/transport/TCPChannelResourceBasic.cpp

@@ -14,8 +14,9 @@
 
 #include <rtps/transport/TCPChannelResourceBasic.h>
 
-#include <future>
 #include <array>
+#include <future>
+#include <mutex>
 
 #include <asio.hpp>
 #include <fastdds/utils/IPLocator.hpp>
@@ -77,13 +78,7 @@ void TCPChannelResourceBasic::connect(
             asio::async_connect(
                 *socket_,
                 endpoints,
-                [this, channel_weak_ptr](std::error_code ec
-#if ASIO_VERSION >= 101200
-                , ip::tcp::endpoint
-#else
-                , ip::tcp::resolver::iterator
-#endif // if ASIO_VERSION >= 101200
-                )
+                [this, channel_weak_ptr](std::error_code ec, ip::tcp::endpoint)
                 {
                     if (!channel_weak_ptr.expired())
                     {
@@ -101,25 +96,22 @@ void TCPChannelResourceBasic::connect(
 
 void TCPChannelResourceBasic::disconnect()
 {
-    if (eConnecting < change_status(eConnectionStatus::eDisconnected) && alive())
+    // Go to disconnecting state to protect from concurrent connects and disconnects
+    auto prev_status = change_status(eConnectionStatus::eDisconnecting);
+    if (eConnecting < prev_status && alive())
     {
-        std::lock_guard<std::mutex> read_lock(read_mutex_);
-        auto socket = socket_;
+        // Shutdown the socket to abort any ongoing read and write operations
+        shutdown(asio::ip::tcp::socket::shutdown_both);
 
-        std::error_code ec;
-        socket->shutdown(asio::ip::tcp::socket::shutdown_both, ec);
+        cancel();
+        close(); // Blocks until all read and write operations have finished
 
-        asio::post(context_, [&, socket]()
-                {
-                    try
-                    {
-                        socket->cancel();
-                        socket->close();
-                    }
-                    catch (std::exception&)
-                    {
-                    }
-                });
+        // Change to disconnected state as the last step
+        change_status(eConnectionStatus::eDisconnected);
+    }
+    else if (eConnectionStatus::eDisconnecting != prev_status || !alive())
+    {
+        change_status(eConnectionStatus::eDisconnected);
     }
 }
 
@@ -128,10 +120,9 @@ uint32_t TCPChannelResourceBasic::read(
         std::size_t size,
         asio::error_code& ec)
 {
-    std::unique_lock<std::mutex> read_lock(read_mutex_);
-
-    if (eConnecting < connection_status_)
+    if (connected())
     {
+        std::unique_lock<std::mutex> read_lock(read_mutex_);
         return static_cast<uint32_t>(asio::read(*socket_, asio::buffer(buffer, size), transfer_exactly(size), ec));
     }
 
@@ -147,7 +138,7 @@ size_t TCPChannelResourceBasic::send(
 {
     size_t bytes_sent = 0;
 
-    if (eConnecting < connection_status_)
+    if (connected())
     {
         std::lock_guard<std::mutex> send_guard(send_mutex_);
 
@@ -195,23 +186,32 @@ asio::ip::tcp::endpoint TCPChannelResourceBasic::local_endpoint(
 void TCPChannelResourceBasic::set_options(
         const TCPTransportDescriptor* options)
 {
-    TCPChannelResource::set_socket_options(*socket_, options);
+    set_socket_options(*socket_, options);
 }
 
 void TCPChannelResourceBasic::cancel()
 {
-    socket_->cancel();
+    std::error_code ec;
+    socket_->cancel(ec); // thread safe with respect to asio's read and write methods
 }
 
 void TCPChannelResourceBasic::close()
 {
-    socket_->close();
+    // Wait for read and write operations to finish before closing the socket (otherwise not thread safe)
+    // NOTE: shutdown should have been called before closing to abort any ongoing operation
+    std::unique_lock<std::mutex> send_lk(send_mutex_, std::defer_lock);
+    std::unique_lock<std::mutex> read_lk(read_mutex_, std::defer_lock);
+    std::lock(send_lk, read_lk); // Pre C++17 alternative to std::scoped_lock
+
+    std::error_code ec;
+    socket_->close(ec);
 }
 
 void TCPChannelResourceBasic::shutdown(
         asio::socket_base::shutdown_type what)
 {
-    socket_->shutdown(what);
+    std::error_code ec;
+    socket_->shutdown(what, ec); // thread safe with respect to asio's read and write methods
 }
 
 } // namespace rtps

src/cpp/rtps/transport/TCPChannelResourceBasic.h

@@ -28,6 +28,7 @@ class TCPChannelResourceBasic : public TCPChannelResource
     asio::io_context& context_;
 
     std::mutex send_mutex_;
+    std::mutex read_mutex_;
     std::shared_ptr<asio::ip::tcp::socket> socket_;
 
 public:

src/cpp/rtps/transport/TCPChannelResourceSecure.cpp

@@ -91,13 +91,7 @@ void TCPChannelResourceSecure::connect(
             const auto secure_socket = secure_socket_;
 
             asio::async_connect(secure_socket_->lowest_layer(), endpoints,
-                    [secure_socket, channel_weak_ptr, parent](const std::error_code& error
-#if ASIO_VERSION >= 101200
-                    , ip::tcp::endpoint
-#else
-                    , const tcp::resolver::iterator& /*endpoint*/
-#endif // if ASIO_VERSION >= 101200
-                    )
+                    [secure_socket, channel_weak_ptr, parent](const std::error_code& error, ip::tcp::endpoint)
                     {
                         if (!error)
                         {
@@ -138,19 +132,33 @@ void TCPChannelResourceSecure::connect(
 
 void TCPChannelResourceSecure::disconnect()
 {
-    if (eConnecting < change_status(eConnectionStatus::eDisconnected) && alive())
+    // Go to disconnecting state to protect from concurrent connects and disconnects
+    auto prev_status = change_status(eConnectionStatus::eDisconnecting);
+    if (eConnecting < prev_status && alive())
     {
         auto socket = secure_socket_;
 
         post(context_, [&, socket]()
                 {
                     std::error_code ec;
-                    socket->lowest_layer().close(ec);
                     socket->async_shutdown([&, socket](const std::error_code&)
                     {
                     });
+
+                    // Close the underlying socket after SSL shutdown
+                    // NOTE: the (async) SSL shutdown may not complete before the socket is closed
+                    socket->lowest_layer().shutdown(asio::ip::tcp::socket::shutdown_both, ec);
+                    socket->lowest_layer().cancel(ec);
+                    socket->lowest_layer().close(ec);
+
+                    // Change to disconnected state as the last step
+                    this->change_status(eConnectionStatus::eDisconnected);
                 });
     }
+    else if (eConnectionStatus::eDisconnecting != prev_status || !alive())
+    {
+        change_status(eConnectionStatus::eDisconnected);
+    }
 }
 
 uint32_t TCPChannelResourceSecure::read(
@@ -160,7 +168,7 @@ uint32_t TCPChannelResourceSecure::read(
 {
     size_t bytes_read = 0;
 
-    if (eConnecting < connection_status_)
+    if (connected())
     {
         std::promise<size_t> read_bytes_promise;
         auto bytes_future = read_bytes_promise.get_future();
@@ -205,7 +213,7 @@ size_t TCPChannelResourceSecure::send(
 {
     size_t bytes_sent = 0;
 
-    if (eConnecting < connection_status_)
+    if (connected())
     {
         if (parent_->configuration()->non_blocking_send &&
                 !check_socket_send_buffer(header_size + total_bytes,
@@ -278,7 +286,7 @@ asio::ip::tcp::endpoint TCPChannelResourceSecure::local_endpoint(
 void TCPChannelResourceSecure::set_options(
         const TCPTransportDescriptor* options)
 {
-    TCPChannelResource::set_socket_options(secure_socket_->lowest_layer(), options);
+    set_socket_options(secure_socket_->lowest_layer(), options);
 }
 
 void TCPChannelResourceSecure::set_tls_verify_mode(
@@ -339,6 +347,7 @@ void TCPChannelResourceSecure::close()
 void TCPChannelResourceSecure::shutdown(
         asio::socket_base::shutdown_type)
 {
+    // WARNING: This function blocks until receiving the peer’s close_notify (or an error occurs).
     secure_socket_->shutdown();
 }