From bae0b8df2e638a2eba8ec34056ad899e35675a44 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jes=C3=BAs=20P=C3=A9rez?= <78275223+jepemi@users.noreply.github.com> Date: Wed, 21 Feb 2024 21:11:34 +0100 Subject: [PATCH] Fix TCP reconnection after open logical port failure (#4324) * Refs #20141: Relocate pending ports erasing Signed-off-by: Jesus Perez * Refs #20141: Add test Signed-off-by: Jesus Perez * Refs #20141: Uncrustify Signed-off-by: Jesus Perez * Refs #20141: Fix asan tests Signed-off-by: Jesus Perez * Refs #20141: Changes after rebase Signed-off-by: Jesus Perez * Refs #20141: Fix windows tests Signed-off-by: Jesus Perez * Refs #20141: Uncrustify Signed-off-by: Jesus Perez * Refs #20141: Apply suggestions Signed-off-by: Jesus Perez --------- Signed-off-by: Jesus Perez (cherry picked from commit 3db2db893c0d59e740c5c31d05e8ed439cb71280) --- src/cpp/rtps/transport/TCPChannelResource.cpp | 2 +- test/unittest/transport/TCPv4Tests.cpp | 86 +++++++++++++++++++ test/unittest/transport/TCPv6Tests.cpp | 86 +++++++++++++++++++ .../transport/mock/MockTCPv4Transport.h | 9 ++ .../transport/mock/MockTCPv6Transport.h | 9 ++ 5 files changed, 191 insertions(+), 1 deletion(-) diff --git a/src/cpp/rtps/transport/TCPChannelResource.cpp b/src/cpp/rtps/transport/TCPChannelResource.cpp index ab95d4eaf2b..315e9f6cbe0 100644 --- a/src/cpp/rtps/transport/TCPChannelResource.cpp +++ b/src/cpp/rtps/transport/TCPChannelResource.cpp @@ -176,9 +176,9 @@ void TCPChannelResource::add_logical_port_response( negotiating_logical_ports_.erase(it); if (portIt != pending_logical_output_ports_.end()) { - pending_logical_output_ports_.erase(portIt); if (success) { + pending_logical_output_ports_.erase(portIt); logical_output_ports_.push_back(port); EPROSIMA_LOG_INFO(RTCP, "OpenedLogicalPort: " << port); } diff --git a/test/unittest/transport/TCPv4Tests.cpp b/test/unittest/transport/TCPv4Tests.cpp index 5e4dab9e691..9fbe3dccb77 100644 --- a/test/unittest/transport/TCPv4Tests.cpp +++ b/test/unittest/transport/TCPv4Tests.cpp @@ -1961,6 +1961,92 @@ TEST_F(TCPv4Tests, non_blocking_send) } #endif // ifndef _WIN32 +// This test verifies that a server can reconnect to a client after the client has once failed in a +// openLogicalPort request +TEST_F(TCPv4Tests, reconnect_after_open_port_failure) +{ + eprosima::fastdds::dds::Log::SetVerbosity(eprosima::fastdds::dds::Log::Warning); + uint16_t port = g_default_port; + // Create a TCP Server transport + TCPv4TransportDescriptor serverDescriptor; + serverDescriptor.add_listener_port(port); + std::unique_ptr serverTransportUnderTest(new TCPv4Transport(serverDescriptor)); + serverTransportUnderTest->init(); + + // Create a TCP Client transport + TCPv4TransportDescriptor clientDescriptor; + std::unique_ptr clientTransportUnderTest(new MockTCPv4Transport(clientDescriptor)); + clientTransportUnderTest->init(); + + // Add initial peer to the client + Locator_t initialPeerLocator; + IPLocator::createLocator(LOCATOR_KIND_TCPv4, "127.0.0.1", port, initialPeerLocator); + IPLocator::setLogicalPort(initialPeerLocator, 7410); + + // Connect client to server + EXPECT_TRUE(serverTransportUnderTest->OpenInputChannel(initialPeerLocator, nullptr, 0x00FF)); + SendResourceList client_resource_list; + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, initialPeerLocator)); + ASSERT_FALSE(client_resource_list.empty()); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + auto channel = clientTransportUnderTest->get_channel_resources().begin()->second; + + // Logical port is opened + ASSERT_TRUE(channel->is_logical_port_opened(7410)); + + // Disconnect server + EXPECT_TRUE(serverTransportUnderTest->CloseInputChannel(initialPeerLocator)); + serverTransportUnderTest.reset(); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + // Client should have passed logical port to pending list + ASSERT_FALSE(channel->is_logical_port_opened(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7410)); + + // Now try normal reconnection + serverTransportUnderTest.reset(new TCPv4Transport(serverDescriptor)); + serverTransportUnderTest->init(); + ASSERT_TRUE(serverTransportUnderTest->OpenInputChannel(initialPeerLocator, nullptr, 0x00FF)); + clientTransportUnderTest->send(nullptr, 0, channel->locator(), initialPeerLocator); // connect() + + // Logical port is opened (moved from pending list) + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + ASSERT_TRUE(channel->is_logical_port_opened(7410)); + + // Disconnect server + EXPECT_TRUE(serverTransportUnderTest->CloseInputChannel(initialPeerLocator)); + serverTransportUnderTest.reset(); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + // Client should have passed logical port to pending list + ASSERT_FALSE(channel->is_logical_port_opened(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7410)); + + // Now try reconnect the server and close server's input channel before client's open logical + // port request, and then delete server and reconnect + serverTransportUnderTest.reset(new TCPv4Transport(serverDescriptor)); + serverTransportUnderTest->init(); + ASSERT_TRUE(serverTransportUnderTest->OpenInputChannel(initialPeerLocator, nullptr, 0x00FF)); + EXPECT_TRUE(serverTransportUnderTest->CloseInputChannel(initialPeerLocator)); + clientTransportUnderTest->send(nullptr, 0, channel->locator(), initialPeerLocator); // connect() + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + serverTransportUnderTest.reset(); + ASSERT_FALSE(channel->is_logical_port_opened(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7410)); + + // Now try normal reconnection + serverTransportUnderTest.reset(new TCPv4Transport(serverDescriptor)); + serverTransportUnderTest->init(); + ASSERT_TRUE(serverTransportUnderTest->OpenInputChannel(initialPeerLocator, nullptr, 0x00FF)); + clientTransportUnderTest->send(nullptr, 0, channel->locator(), initialPeerLocator); // connect() + + // Logical port is opened (moved from pending list) + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + ASSERT_TRUE(channel->is_logical_port_opened(7410)); + + // Clear test + EXPECT_TRUE(serverTransportUnderTest->CloseInputChannel(initialPeerLocator)); + client_resource_list.clear(); +} + void TCPv4Tests::HELPER_SetDescriptorDefaults() { descriptor.add_listener_port(g_default_port); diff --git a/test/unittest/transport/TCPv6Tests.cpp b/test/unittest/transport/TCPv6Tests.cpp index 1834e28cef0..596e2460acc 100644 --- a/test/unittest/transport/TCPv6Tests.cpp +++ b/test/unittest/transport/TCPv6Tests.cpp @@ -400,6 +400,92 @@ TEST_F(TCPv6Tests, non_blocking_send) } #endif // ifndef _WIN32 +// This test verifies that a server can reconnect to a client after the client has once failed in a +// openLogicalPort request +TEST_F(TCPv6Tests, reconnect_after_open_port_failure) +{ + eprosima::fastdds::dds::Log::SetVerbosity(eprosima::fastdds::dds::Log::Warning); + uint16_t port = g_default_port; + // Create a TCP Server transport + TCPv6TransportDescriptor serverDescriptor; + serverDescriptor.add_listener_port(port); + std::unique_ptr serverTransportUnderTest(new TCPv6Transport(serverDescriptor)); + serverTransportUnderTest->init(); + + // Create a TCP Client transport + TCPv6TransportDescriptor clientDescriptor; + std::unique_ptr clientTransportUnderTest(new MockTCPv6Transport(clientDescriptor)); + clientTransportUnderTest->init(); + + // Add initial peer to the client + Locator_t initialPeerLocator; + IPLocator::createLocator(LOCATOR_KIND_TCPv6, "::1", port, initialPeerLocator); + IPLocator::setLogicalPort(initialPeerLocator, 7410); + + // Connect client to server + EXPECT_TRUE(serverTransportUnderTest->OpenInputChannel(initialPeerLocator, nullptr, 0x00FF)); + SendResourceList client_resource_list; + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, initialPeerLocator)); + ASSERT_FALSE(client_resource_list.empty()); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + auto channel = clientTransportUnderTest->get_channel_resources().begin()->second; + + // Logical port is opened + ASSERT_TRUE(channel->is_logical_port_opened(7410)); + + // Disconnect server + EXPECT_TRUE(serverTransportUnderTest->CloseInputChannel(initialPeerLocator)); + serverTransportUnderTest.reset(); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + // Client should have passed logical port to pending list + ASSERT_FALSE(channel->is_logical_port_opened(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7410)); + + // Now try normal reconnection + serverTransportUnderTest.reset(new TCPv6Transport(serverDescriptor)); + serverTransportUnderTest->init(); + ASSERT_TRUE(serverTransportUnderTest->OpenInputChannel(initialPeerLocator, nullptr, 0x00FF)); + clientTransportUnderTest->send(nullptr, 0, channel->locator(), initialPeerLocator); // connect() + + // Logical port is opened (moved from pending list) + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + ASSERT_TRUE(channel->is_logical_port_opened(7410)); + + // Disconnect server + EXPECT_TRUE(serverTransportUnderTest->CloseInputChannel(initialPeerLocator)); + serverTransportUnderTest.reset(); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + // Client should have passed logical port to pending list + ASSERT_FALSE(channel->is_logical_port_opened(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7410)); + + // Now try reconnect the server and close server's input channel before client's open logical + // port request, and then delete server and reconnect + serverTransportUnderTest.reset(new TCPv6Transport(serverDescriptor)); + serverTransportUnderTest->init(); + ASSERT_TRUE(serverTransportUnderTest->OpenInputChannel(initialPeerLocator, nullptr, 0x00FF)); + EXPECT_TRUE(serverTransportUnderTest->CloseInputChannel(initialPeerLocator)); + clientTransportUnderTest->send(nullptr, 0, channel->locator(), initialPeerLocator); // connect() + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + serverTransportUnderTest.reset(); + ASSERT_FALSE(channel->is_logical_port_opened(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7410)); + + // Now try normal reconnection + serverTransportUnderTest.reset(new TCPv6Transport(serverDescriptor)); + serverTransportUnderTest->init(); + ASSERT_TRUE(serverTransportUnderTest->OpenInputChannel(initialPeerLocator, nullptr, 0x00FF)); + clientTransportUnderTest->send(nullptr, 0, channel->locator(), initialPeerLocator); // connect() + + // Logical port is opened (moved from pending list) + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + ASSERT_TRUE(channel->is_logical_port_opened(7410)); + + // Clear test + EXPECT_TRUE(serverTransportUnderTest->CloseInputChannel(initialPeerLocator)); + client_resource_list.clear(); +} + /* TEST_F(TCPv6Tests, send_and_receive_between_both_secure_ports) { diff --git a/test/unittest/transport/mock/MockTCPv4Transport.h b/test/unittest/transport/mock/MockTCPv4Transport.h index f5aff8b452b..08569dbd08a 100644 --- a/test/unittest/transport/mock/MockTCPv4Transport.h +++ b/test/unittest/transport/mock/MockTCPv4Transport.h @@ -56,6 +56,15 @@ class MockTCPv4Transport : public TCPv4Transport return acceptors_; } + bool send( + const fastrtps::rtps::octet* send_buffer, + uint32_t send_buffer_size, + const fastrtps::rtps::Locator_t& send_resource_locator, + const Locator_t& remote_locator) + { + return TCPv4Transport::send(send_buffer, send_buffer_size, send_resource_locator, remote_locator); + } + }; } // namespace rtps diff --git a/test/unittest/transport/mock/MockTCPv6Transport.h b/test/unittest/transport/mock/MockTCPv6Transport.h index 33f9155e2dc..c407b5fd650 100644 --- a/test/unittest/transport/mock/MockTCPv6Transport.h +++ b/test/unittest/transport/mock/MockTCPv6Transport.h @@ -56,6 +56,15 @@ class MockTCPv6Transport : public TCPv6Transport return acceptors_; } + bool send( + const fastrtps::rtps::octet* send_buffer, + uint32_t send_buffer_size, + const fastrtps::rtps::Locator_t& send_resource_locator, + const Locator_t& remote_locator) + { + return TCPv6Transport::send(send_buffer, send_buffer_size, send_resource_locator, remote_locator); + } + }; } // namespace rtps