From bf0063873213348a66872f71ebf2a1fe8368bb65 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Ferreira=20Gonz=C3=A1lez?= Date: Fri, 12 Jun 2026 07:51:46 +0200 Subject: [PATCH 1/2] Remove stale tcp channels (#6421) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Refs #24549: Test for removing stale channel_resources_ of finished clients Signed-off-by: Carlos Ferreira González * Refs #24549: Improve TCP clean up Signed-off-by: Carlos Ferreira González * Refs #24549: Ensure proper destruction in test Signed-off-by: Carlos Ferreira González * Refs #24549: Protect smart wait from potential data race Signed-off-by: Carlos Ferreira González * Refs #24549: Fix test Signed-off-by: Carlos Ferreira González --------- Signed-off-by: Carlos Ferreira González (cherry picked from commit 8188b416373081216996fd219a2d0eadf4c9ed48) --- .../rtps/transport/TCPTransportInterface.cpp | 49 +++++++++++++++++ test/unittest/transport/TCPv4Tests.cpp | 54 +++++++++++++++++++ .../transport/mock/MockTCPv4Transport.h | 12 +++++ 3 files changed, 115 insertions(+) diff --git a/src/cpp/rtps/transport/TCPTransportInterface.cpp b/src/cpp/rtps/transport/TCPTransportInterface.cpp index 28a54d698cf..73f422c73a8 100644 --- a/src/cpp/rtps/transport/TCPTransportInterface.cpp +++ b/src/cpp/rtps/transport/TCPTransportInterface.cpp @@ -1221,6 +1221,55 @@ void TCPTransportInterface::perform_listen_operation( } EPROSIMA_LOG_INFO(RTCP, "End PerformListenOperation " << channel->locator()); + + // If we get here, the channel has been disconnected. We might need to clean it up if + // the remote endpoint is the one that initiated the disconnection. + // We only delete acceptor channels, as connect channels need to be kept in channel_resources_ to restart the connection + if (channel && channel->tcp_connection_type() == TCPChannelResource::TCPConnectionType::TCP_ACCEPT_TYPE) + { + // Defer the erase to io_context_ so the TCPChannelResource destructor runs off the listener thread that is about to exit. + // Weak_ptr is used to avoid keeping the channel alive if it has already been removed from the maps by another thread. + asio::post(io_context_, [this, channel_weak]() + { + auto ch = channel_weak.lock(); + if (!ch) + { + return; + } + { + // Channel resources map case + std::unique_lock scoped_lock(sockets_map_mutex_); + bool erased = false; + // There might be multiple entries with the same channel. Delete them all + for (auto it = channel_resources_.begin(); it != channel_resources_.end(); ) + { + if (it->second == ch) + { + it = channel_resources_.erase(it); + erased = true; + } + else + { + ++it; + } + } + if (erased) + { + return; + } + } + // Unbound channel resources map case + std::unique_lock unbound_lock(unbound_map_mutex_); + auto it = std::find(unbound_channel_resources_.begin(), + unbound_channel_resources_.end(), ch); + if (it != unbound_channel_resources_.end()) + { + unbound_channel_resources_.erase(it); + } + }); + // Drop the listener's reference so the destructor cannot run on this thread + channel.reset(); + } } bool TCPTransportInterface::read_body( diff --git a/test/unittest/transport/TCPv4Tests.cpp b/test/unittest/transport/TCPv4Tests.cpp index b441e7d0e7d..c467c8c1b52 100644 --- a/test/unittest/transport/TCPv4Tests.cpp +++ b/test/unittest/transport/TCPv4Tests.cpp @@ -2429,6 +2429,60 @@ TEST_F(TCPv4Tests, add_logical_port_on_send_resource_creation) } } +// This test verifies that TCP channels of type ACCEPT are correctly removed from the channel resources map when +// the channel is disabled by asio. This is the case when a client disconnects from the server. There is no need +// maintain the channel resource of a disconnected client because new connections will generate new channel resources +// and no unbind operation is needed at destruction time for a removed participant (eDisconnected channel). +TEST_F(TCPv4Tests, remove_stale_channel_resources_of_server) +{ + // Server + TCPv4TransportDescriptor serverDescriptor; + serverDescriptor.add_listener_port(g_default_port); + MockTCPv4Transport server(serverDescriptor); + ASSERT_TRUE(server.init()); + + // Client + { + TCPv4TransportDescriptor clientDescriptor; + auto client = std::unique_ptr(new TCPv4Transport(clientDescriptor)); + ASSERT_TRUE(client->init()); + + Locator_t outputLocator; + outputLocator.kind = LOCATOR_KIND_TCPv4; + IPLocator::setIPv4(outputLocator, 127, 0, 0, 1); + IPLocator::setPhysicalPort(outputLocator, g_default_port); + IPLocator::setLogicalPort(outputLocator, 7410); + + SendResourceList send_resource_list; + ASSERT_TRUE(client->OpenOutputChannel(send_resource_list, outputLocator)); + + // Wait for the server to finish the BindConnectionRequest handshake + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (server.get_channel_resources_size() == 0 && + std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + // Ensure there are channel resources in the server. Bind socket adds an entry per interface available, so there could be more than one. + ASSERT_GT(server.get_channel_resources_size(), 0u); + + // Tear down the client: clean send_resource_list and then close the TCP socket. + send_resource_list.clear(); + client.reset(); + } + + // Check that the server correctly removes the channel resource of type ACCEPT after the client disconnection + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (server.get_channel_resources_size() != 0 && + std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + EXPECT_EQ(server.get_channel_resources_size(), 0u); + EXPECT_EQ(server.get_unbound_channel_resources_size(), 0u); +} + + void TCPv4Tests::HELPER_SetDescriptorDefaults() { descriptor.add_listener_port(g_default_port); diff --git a/test/unittest/transport/mock/MockTCPv4Transport.h b/test/unittest/transport/mock/MockTCPv4Transport.h index c02a26d4ca9..075a3a80695 100644 --- a/test/unittest/transport/mock/MockTCPv4Transport.h +++ b/test/unittest/transport/mock/MockTCPv4Transport.h @@ -40,11 +40,23 @@ class MockTCPv4Transport : public TCPv4Transport return channel_resources_; } + size_t get_channel_resources_size() const + { + std::lock_guard lock(sockets_map_mutex_); + return channel_resources_.size(); + } + const std::vector> get_unbound_channel_resources() const { return unbound_channel_resources_; } + size_t get_unbound_channel_resources_size() const + { + std::lock_guard lock(unbound_map_mutex_); + return unbound_channel_resources_.size(); + } + const std::vector& get_interface_whitelist() const { return interface_whitelist_; From c07b3fb8ce2c80bc19c9a5907dd494525c032268 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Ferreira=20Gonz=C3=A1lez?= Date: Tue, 14 Jul 2026 12:20:11 +0200 Subject: [PATCH 2/2] Uncrustify MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Carlos Ferreira González --- src/cpp/rtps/transport/TCPTransportInterface.cpp | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/src/cpp/rtps/transport/TCPTransportInterface.cpp b/src/cpp/rtps/transport/TCPTransportInterface.cpp index 73f422c73a8..533cd57d546 100644 --- a/src/cpp/rtps/transport/TCPTransportInterface.cpp +++ b/src/cpp/rtps/transport/TCPTransportInterface.cpp @@ -585,7 +585,8 @@ bool TCPTransportInterface::init( { auto ioContextTimersFunction = [&]() { - asio::executor_work_guard work = make_work_guard(io_context_timers_. + asio::executor_work_guard work = + make_work_guard(io_context_timers_. get_executor()); io_context_timers_.run(); }; @@ -1068,8 +1069,8 @@ bool TCPTransportInterface::OpenInputChannel( } EPROSIMA_LOG_INFO(RTCP, " OpenInputChannel (physical: " << IPLocator::getPhysicalPort( - locator) << "; logical: " << \ - IPLocator::getLogicalPort(locator) << ")"); + locator) << "; logical: " \ + << IPLocator::getLogicalPort(locator) << ")"); } } return success; @@ -1619,7 +1620,8 @@ bool TCPTransportInterface::send( // Logical port might be under negotiation. Wait a little and check again. This prevents from // losing first messages. scoped_lock.unlock(); - bool logical_port_opened = channel->wait_logical_port_under_negotiation(logical_port, std::chrono::milliseconds( + bool logical_port_opened = channel->wait_logical_port_under_negotiation(logical_port, + std::chrono::milliseconds( configuration()->tcp_negotiation_timeout)); if (!logical_port_opened) { @@ -1642,8 +1644,9 @@ bool TCPTransportInterface::send( if (sent != static_cast(TCPHeader::size() + total_bytes) || ec) { - EPROSIMA_LOG_WARNING(DEBUG, "Failed to send RTCP message (" << sent << " of " << - TCPHeader::size() + total_bytes << " b): " << ec.message()); + EPROSIMA_LOG_WARNING(DEBUG, "Failed to send RTCP message (" << sent << " of " + << TCPHeader::size() + total_bytes + << " b): " << ec.message()); success = false; } else