Skip to content

Commit 3538bf5

Browse files
mergify[bot]cferreiragonz
mergify[bot]
and
cferreiragonz
authored
Remove stale TCP channels (#6421) (#6458)
* Remove stale tcp channels (#6421) * Refs #24549: Test for removing stale channel_resources_ of finished clients Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com> * Refs #24549: Improve TCP clean up Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com> * Refs #24549: Ensure proper destruction in test Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com> * Refs #24549: Protect smart wait from potential data race Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com> * Refs #24549: Fix test Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com> --------- Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com> (cherry picked from commit 8188b41) * Uncrustify Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com> --------- Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com> Co-authored-by: Carlos Ferreira González <carlosferreira@eprosima.com>
1 parent 9ada349 commit 3538bf5

3 files changed

Lines changed: 124 additions & 6 deletions

File tree

‎src/cpp/rtps/transport/TCPTransportInterface.cpp‎

Lines changed: 58 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -585,7 +585,8 @@ bool TCPTransportInterface::init(
585585
{
586586
auto ioContextTimersFunction = [&]()
587587
{
588-
asio::executor_work_guard<asio::io_context::executor_type> work = make_work_guard(io_context_timers_.
588+
asio::executor_work_guard<asio::io_context::executor_type> work =
589+
make_work_guard(io_context_timers_.
589590
get_executor());
590591
io_context_timers_.run();
591592
};
@@ -1068,8 +1069,8 @@ bool TCPTransportInterface::OpenInputChannel(
10681069
}
10691070

10701071
EPROSIMA_LOG_INFO(RTCP, " OpenInputChannel (physical: " << IPLocator::getPhysicalPort(
1071-
locator) << "; logical: " << \
1072-
IPLocator::getLogicalPort(locator) << ")");
1072+
locator) << "; logical: " \
1073+
<< IPLocator::getLogicalPort(locator) << ")");
10731074
}
10741075
}
10751076
return success;
@@ -1221,6 +1222,55 @@ void TCPTransportInterface::perform_listen_operation(
12211222
}
12221223

12231224
EPROSIMA_LOG_INFO(RTCP, "End PerformListenOperation " << channel->locator());
1225+
1226+
// If we get here, the channel has been disconnected. We might need to clean it up if
1227+
// the remote endpoint is the one that initiated the disconnection.
1228+
// We only delete acceptor channels, as connect channels need to be kept in channel_resources_ to restart the connection
1229+
if (channel && channel->tcp_connection_type() == TCPChannelResource::TCPConnectionType::TCP_ACCEPT_TYPE)
1230+
{
1231+
// Defer the erase to io_context_ so the TCPChannelResource destructor runs off the listener thread that is about to exit.
1232+
// Weak_ptr is used to avoid keeping the channel alive if it has already been removed from the maps by another thread.
1233+
asio::post(io_context_, [this, channel_weak]()
1234+
{
1235+
auto ch = channel_weak.lock();
1236+
if (!ch)
1237+
{
1238+
return;
1239+
}
1240+
{
1241+
// Channel resources map case
1242+
std::unique_lock<std::mutex> scoped_lock(sockets_map_mutex_);
1243+
bool erased = false;
1244+
// There might be multiple entries with the same channel. Delete them all
1245+
for (auto it = channel_resources_.begin(); it != channel_resources_.end(); )
1246+
{
1247+
if (it->second == ch)
1248+
{
1249+
it = channel_resources_.erase(it);
1250+
erased = true;
1251+
}
1252+
else
1253+
{
1254+
++it;
1255+
}
1256+
}
1257+
if (erased)
1258+
{
1259+
return;
1260+
}
1261+
}
1262+
// Unbound channel resources map case
1263+
std::unique_lock<std::mutex> unbound_lock(unbound_map_mutex_);
1264+
auto it = std::find(unbound_channel_resources_.begin(),
1265+
unbound_channel_resources_.end(), ch);
1266+
if (it != unbound_channel_resources_.end())
1267+
{
1268+
unbound_channel_resources_.erase(it);
1269+
}
1270+
});
1271+
// Drop the listener's reference so the destructor cannot run on this thread
1272+
channel.reset();
1273+
}
12241274
}
12251275

12261276
bool TCPTransportInterface::read_body(
@@ -1584,7 +1634,8 @@ bool TCPTransportInterface::send(
15841634
// Logical port might be under negotiation. Wait a little and check again. This prevents from
15851635
// losing first messages.
15861636
scoped_lock.unlock();
1587-
bool logical_port_opened = channel->wait_logical_port_under_negotiation(logical_port, std::chrono::milliseconds(
1637+
bool logical_port_opened = channel->wait_logical_port_under_negotiation(logical_port,
1638+
std::chrono::milliseconds(
15881639
configuration()->tcp_negotiation_timeout));
15891640
if (!logical_port_opened)
15901641
{
@@ -1607,8 +1658,9 @@ bool TCPTransportInterface::send(
16071658

16081659
if (sent != static_cast<uint32_t>(TCPHeader::size() + total_bytes) || ec)
16091660
{
1610-
EPROSIMA_LOG_WARNING(DEBUG, "Failed to send RTCP message (" << sent << " of " <<
1611-
TCPHeader::size() + total_bytes << " b): " << ec.message());
1661+
EPROSIMA_LOG_WARNING(DEBUG, "Failed to send RTCP message (" << sent << " of "
1662+
<< TCPHeader::size() + total_bytes
1663+
<< " b): " << ec.message());
16121664
success = false;
16131665
}
16141666
else

‎test/unittest/transport/TCPv4Tests.cpp‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2429,6 +2429,60 @@ TEST_F(TCPv4Tests, add_logical_port_on_send_resource_creation)
24292429
}
24302430
}
24312431

2432+
// This test verifies that TCP channels of type ACCEPT are correctly removed from the channel resources map when
2433+
// the channel is disabled by asio. This is the case when a client disconnects from the server. There is no need
2434+
// maintain the channel resource of a disconnected client because new connections will generate new channel resources
2435+
// and no unbind operation is needed at destruction time for a removed participant (eDisconnected channel).
2436+
TEST_F(TCPv4Tests, remove_stale_channel_resources_of_server)
2437+
{
2438+
// Server
2439+
TCPv4TransportDescriptor serverDescriptor;
2440+
serverDescriptor.add_listener_port(g_default_port);
2441+
MockTCPv4Transport server(serverDescriptor);
2442+
ASSERT_TRUE(server.init());
2443+
2444+
// Client
2445+
{
2446+
TCPv4TransportDescriptor clientDescriptor;
2447+
auto client = std::unique_ptr<TCPv4Transport>(new TCPv4Transport(clientDescriptor));
2448+
ASSERT_TRUE(client->init());
2449+
2450+
Locator_t outputLocator;
2451+
outputLocator.kind = LOCATOR_KIND_TCPv4;
2452+
IPLocator::setIPv4(outputLocator, 127, 0, 0, 1);
2453+
IPLocator::setPhysicalPort(outputLocator, g_default_port);
2454+
IPLocator::setLogicalPort(outputLocator, 7410);
2455+
2456+
SendResourceList send_resource_list;
2457+
ASSERT_TRUE(client->OpenOutputChannel(send_resource_list, outputLocator));
2458+
2459+
// Wait for the server to finish the BindConnectionRequest handshake
2460+
auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
2461+
while (server.get_channel_resources_size() == 0 &&
2462+
std::chrono::steady_clock::now() < deadline)
2463+
{
2464+
std::this_thread::sleep_for(std::chrono::milliseconds(20));
2465+
}
2466+
// Ensure there are channel resources in the server. Bind socket adds an entry per interface available, so there could be more than one.
2467+
ASSERT_GT(server.get_channel_resources_size(), 0u);
2468+
2469+
// Tear down the client: clean send_resource_list and then close the TCP socket.
2470+
send_resource_list.clear();
2471+
client.reset();
2472+
}
2473+
2474+
// Check that the server correctly removes the channel resource of type ACCEPT after the client disconnection
2475+
auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
2476+
while (server.get_channel_resources_size() != 0 &&
2477+
std::chrono::steady_clock::now() < deadline)
2478+
{
2479+
std::this_thread::sleep_for(std::chrono::milliseconds(100));
2480+
}
2481+
EXPECT_EQ(server.get_channel_resources_size(), 0u);
2482+
EXPECT_EQ(server.get_unbound_channel_resources_size(), 0u);
2483+
}
2484+
2485+
24322486
void TCPv4Tests::HELPER_SetDescriptorDefaults()
24332487
{
24342488
descriptor.add_listener_port(g_default_port);

‎test/unittest/transport/mock/MockTCPv4Transport.h‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,11 +40,23 @@ class MockTCPv4Transport : public TCPv4Transport
4040
return channel_resources_;
4141
}
4242

43+
size_t get_channel_resources_size() const
44+
{
45+
std::lock_guard<std::mutex> lock(sockets_map_mutex_);
46+
return channel_resources_.size();
47+
}
48+
4349
const std::vector<std::shared_ptr<TCPChannelResource>> get_unbound_channel_resources() const
4450
{
4551
return unbound_channel_resources_;
4652
}
4753

54+
size_t get_unbound_channel_resources_size() const
55+
{
56+
std::lock_guard<std::mutex> lock(unbound_map_mutex_);
57+
return unbound_channel_resources_.size();
58+
}
59+
4860
const std::vector<asio::ip::address_v4>& get_interface_whitelist() const
4961
{
5062
return interface_whitelist_;

0 commit comments

Comments
 (0)