From 9818c2a8a6e62774b419c49bd309e8251e0c2cb6 Mon Sep 17 00:00:00 2001 From: "mergify[bot]" <37929162+mergify[bot]@users.noreply.github.com> Date: Thu, 7 Nov 2024 16:14:32 +0100 Subject: [PATCH] Fix issues in Dynamic Network Interfaces (#5282) (#5304) * Fix issues in Dynamic Network Interfaces (#5282) * Refs #21690. Parse `--rescan` argument on communication applications. Signed-off-by: Miguel Company * Refs #21690. Implement rescan mechanism. Signed-off-by: Miguel Company * Refs #21690. Add docker infrastructure. Signed-off-by: Miguel Company * Refs #21690. Add CMake infrastructure. Signed-off-by: Miguel Company * Refs #21690. Ensure same domain and topic name are used. Signed-off-by: Miguel Company * Refs #21690. Add `--loops` argument to publisher. Signed-off-by: Miguel Company * Refs #21690. Publisher exits after publishing all samples. Signed-off-by: Miguel Company * Refs #21690. Improve subscriber script. Signed-off-by: Miguel Company * Refs #21690. Add test. Signed-off-by: Miguel Company * Refs #21690. Make publisher wait subscriber. Signed-off-by: Miguel Company * Refs #21690. Possible fix. Signed-off-by: Miguel Company * Refs #21690. Clear locators before recalculating them. Signed-off-by: Miguel Company * Refs #21690. Move local participant proxy update to PDP. Signed-off-by: Miguel Company * Refs #21690. Improve new method's logic. Signed-off-by: Miguel Company * Refs #21690. Include what you use. Signed-off-by: Miguel Company * Refs #21690. Add empty method to update endpoint locators. Signed-off-by: Miguel Company * Refs #21690. Add implementation for `update_endpoint_locators_if_default_nts`. Signed-off-by: Miguel Company * Refs #21690. Compare against old default locators. Signed-off-by: Miguel Company * Refs #21690. Update locators in attributes. Signed-off-by: Miguel Company * Refs #17283. Avoid early return on `PDP::local_participant_attributes_update_nts`. Signed-off-by: Miguel Company * Refs #17283. Apply suggestions. Signed-off-by: Miguel Company --------- Signed-off-by: Miguel Company (cherry picked from commit 91bd7c857d8760f1a523e878368f895fbc41a955) * Refs #17283: Fix conflicts Signed-off-by: eProsima * Refs #17283: Keep original PR functionality Signed-off-by: eProsima * Refs #17283: Please uncrustify Signed-off-by: eProsima * Restore discovery server list check code. Signed-off-by: Miguel Company * Make backported methods static methods instead of inline code. Signed-off-by: Miguel Company * Methods get_default_xxx_locators with output argument. Signed-off-by: Miguel Company * Only update attributes when necessary. Signed-off-by: Miguel Company --------- Signed-off-by: eProsima Signed-off-by: Miguel Company Co-authored-by: Miguel Company Co-authored-by: eProsima (cherry picked from commit 66ec998a804ad066b7a950ecb2237a91d782af8e) Signed-off-by: Miguel Company # Conflicts: # src/cpp/rtps/participant/RTPSParticipantImpl.cpp # test/communication/PublisherMain.cpp # test/communication/SubscriberMain.cpp # test/dds/communication/SubscriberMain.cpp # test/dds/communication/SubscriberModule.hpp # test/dds/communication/security/PublisherModule.cpp # test/dds/communication/security/PublisherModule.hpp # test/dds/communication/security/SubscriberModule.cpp # test/dds/communication/security/SubscriberModule.hpp --- .../rtps/builtin/discovery/participant/PDP.h | 5 + .../rtps/participant/RTPSParticipantImpl.cpp | 265 +++++++++++++----- .../rtps/participant/RTPSParticipantImpl.h | 6 + test/dds/communication/CMakeLists.txt | 4 + test/dds/communication/PubSubMain.cpp | 6 +- test/dds/communication/PublisherMain.cpp | 28 +- test/dds/communication/PublisherModule.cpp | 20 ++ test/dds/communication/PublisherModule.hpp | 19 +- test/dds/communication/SubscriberMain.cpp | 18 +- test/dds/communication/SubscriberModule.cpp | 25 +- test/dds/communication/SubscriberModule.hpp | 27 +- .../communication/dyn_network/CMakeLists.txt | 61 ++++ test/dds/communication/dyn_network/Dockerfile | 27 ++ .../dynamic_interfaces.compose.yml | 42 +++ .../dyn_network/launch_subscriber.bash | 33 +++ 15 files changed, 493 insertions(+), 93 deletions(-) create mode 100644 test/dds/communication/dyn_network/CMakeLists.txt create mode 100644 test/dds/communication/dyn_network/Dockerfile create mode 100644 test/dds/communication/dyn_network/dynamic_interfaces.compose.yml create mode 100755 test/dds/communication/dyn_network/launch_subscriber.bash diff --git a/include/fastdds/rtps/builtin/discovery/participant/PDP.h b/include/fastdds/rtps/builtin/discovery/participant/PDP.h index e38b532c1ee..f9df13b2264 100644 --- a/include/fastdds/rtps/builtin/discovery/participant/PDP.h +++ b/include/fastdds/rtps/builtin/discovery/participant/PDP.h @@ -22,9 +22,12 @@ #ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC #include +#include #include #include #include +#include +#include #include #include @@ -61,6 +64,7 @@ class RTPSWriter; class RTPSReader; class WriterHistory; class ReaderHistory; +struct RTPSParticipantAllocationAttributes; class RTPSParticipantImpl; class RTPSParticipantListener; class BuiltinProtocols; @@ -70,6 +74,7 @@ class ReaderProxyData; class WriterProxyData; class ParticipantProxyData; class ReaderListener; +class PDPEndpoints; class PDPListener; class PDPServerListener; class ITopicPayloadPool; diff --git a/src/cpp/rtps/participant/RTPSParticipantImpl.cpp b/src/cpp/rtps/participant/RTPSParticipantImpl.cpp index 4a77accc971..4c222b591ed 100644 --- a/src/cpp/rtps/participant/RTPSParticipantImpl.cpp +++ b/src/cpp/rtps/participant/RTPSParticipantImpl.cpp @@ -144,6 +144,128 @@ static bool get_unique_flows_parameters( return true; } +// -- The following is a backport from 3.x of a PDP method +static void local_participant_attributes_update_nts( + const RTPSParticipantAttributes& new_atts, + PDP* pdp, + RTPSParticipantImpl* participant) +{ + auto participant_data = pdp->getLocalParticipantProxyData(); + participant_data->m_userData.data_vec(new_atts.userData); + + // If we are intraprocess only, we do not need to update locators + bool announce_locators = !participant->is_intraprocess_only(); + if (announce_locators) + { + // Clear all locators + participant_data->metatraffic_locators.unicast.clear(); + participant_data->metatraffic_locators.multicast.clear(); + participant_data->default_locators.unicast.clear(); + participant_data->default_locators.multicast.clear(); + + // Update default locators + for (const Locator_t& loc : new_atts.defaultUnicastLocatorList) + { + participant_data->default_locators.add_unicast_locator(loc); + } + for (const Locator_t& loc : new_atts.defaultMulticastLocatorList) + { + participant_data->default_locators.add_multicast_locator(loc); + } + + // Update metatraffic locators + for (const auto& locator : new_atts.builtin.metatrafficUnicastLocatorList) + { + participant_data->metatraffic_locators.add_unicast_locator(locator); + } + if (!new_atts.builtin.avoid_builtin_multicast || + participant_data->metatraffic_locators.unicast.empty()) + { + for (const auto& locator : new_atts.builtin.metatrafficMulticastLocatorList) + { + participant_data->metatraffic_locators.add_multicast_locator(locator); + } + } + + namespace external_locators = fastdds::rtps::ExternalLocatorsProcessor; + external_locators::add_external_locators(*participant_data, + new_atts.builtin.metatraffic_external_unicast_locators, + new_atts.default_external_unicast_locators); + } +} + +// -- The following is a backport from 3.x of a PDP method +static void update_endpoint_locators_if_default_nts( + const std::vector& writers, + const std::vector& readers, + const RTPSParticipantAttributes& old_atts, + const RTPSParticipantAttributes& new_atts, + PDP* pdp) +{ + // Check if default locators have changed + const auto& old_default_unicast = old_atts.defaultUnicastLocatorList; + const auto& old_default_multicast = old_atts.defaultMulticastLocatorList; + const auto& new_default_unicast = new_atts.defaultUnicastLocatorList; + const auto& new_default_multicast = new_atts.defaultMulticastLocatorList; + + // Early return if there is no change in default unicast locators + if ((old_default_unicast == new_default_unicast) && + (old_default_multicast == new_default_multicast)) + { + return; + } + + // Update proxies of endpoints with default configured locators + EDP* edp = pdp->getEDP(); + for (RTPSWriter* writer : writers) + { + if ((old_default_multicast == writer->getAttributes().multicastLocatorList) && + (old_default_unicast == writer->getAttributes().unicastLocatorList)) + { + writer->getAttributes().multicastLocatorList = new_default_multicast; + writer->getAttributes().unicastLocatorList = new_default_unicast; + + WriterProxyData* wdata = nullptr; + GUID_t participant_guid; + wdata = pdp->addWriterProxyData(writer->getGuid(), participant_guid, + [](WriterProxyData* proxy, bool is_update, + const ParticipantProxyData& participant) + { + static_cast(is_update); + assert(is_update); + proxy->set_locators(participant.default_locators); + return true; + }); + assert(wdata != nullptr); + edp->processLocalWriterProxyData(writer, wdata); + } + } + for (RTPSReader* reader : readers) + { + if ((old_default_multicast == reader->getAttributes().multicastLocatorList) && + (old_default_unicast == reader->getAttributes().unicastLocatorList)) + { + reader->getAttributes().multicastLocatorList = new_default_multicast; + reader->getAttributes().unicastLocatorList = new_default_unicast; + + ReaderProxyData* rdata = nullptr; + GUID_t participant_guid; + rdata = pdp->addReaderProxyData(reader->getGuid(), participant_guid, + [](ReaderProxyData* proxy, bool is_update, + const ParticipantProxyData& participant) + { + static_cast(is_update); + assert(is_update); + + proxy->set_locators(participant.default_locators); + return true; + }); + assert(rdata != nullptr); + edp->processLocalReaderProxyData(reader, rdata); + } + } +} + Locator_t& RTPSParticipantImpl::applyLocatorAdaptRule( Locator_t& loc) { @@ -1377,29 +1499,38 @@ bool RTPSParticipantImpl::registerReader( void RTPSParticipantImpl::update_attributes( const RTPSParticipantAttributes& patt) { + // Avoid ABBA with PDP mutex by using a local copy of the attributes + RTPSParticipantAttributes temp_atts; + { + std::lock_guard guard(*mp_mutex); + temp_atts = m_att; + } + bool local_interfaces_changed = false; // Check if new interfaces have been added if (internal_metatraffic_locators_) { - LocatorList_t metatraffic_unicast_locator_list = m_att.builtin.metatrafficUnicastLocatorList; - get_default_metatraffic_locators(); - if (!(metatraffic_unicast_locator_list == m_att.builtin.metatrafficUnicastLocatorList)) + LocatorList_t metatraffic_unicast_locator_list = temp_atts.builtin.metatrafficUnicastLocatorList; + temp_atts.builtin.metatrafficUnicastLocatorList.clear(); + get_default_metatraffic_locators(temp_atts); + if (!(metatraffic_unicast_locator_list == temp_atts.builtin.metatrafficUnicastLocatorList)) { local_interfaces_changed = true; - EPROSIMA_LOG_INFO(RTPS_PARTICIPANT, m_att.getName() << " updated its metatraffic locators"); + EPROSIMA_LOG_INFO(RTPS_PARTICIPANT, temp_atts.getName() << " updated its metatraffic locators"); } } if (internal_default_locators_) { - LocatorList_t default_unicast_locator_list = m_att.defaultUnicastLocatorList; - get_default_unicast_locators(); - if (!(default_unicast_locator_list == m_att.defaultUnicastLocatorList)) + LocatorList_t default_unicast_locator_list = temp_atts.defaultUnicastLocatorList; + temp_atts.defaultUnicastLocatorList.clear(); + get_default_unicast_locators(temp_atts); + if (!(default_unicast_locator_list == temp_atts.defaultUnicastLocatorList)) { local_interfaces_changed = true; EPROSIMA_LOG_INFO(RTPS_PARTICIPANT, - m_att.getName() << " updated default unicast locator list, current locators: " - << m_att.defaultUnicastLocatorList); + temp_atts.getName() << " updated default unicast locator list, current locators: " + << temp_atts.defaultUnicastLocatorList); } } @@ -1414,9 +1545,9 @@ void RTPSParticipantImpl::update_attributes( // Check if discovery servers need to be updated eprosima::fastdds::rtps::RemoteServerList_t converted_discovery_servers = patt.builtin.discovery_config.m_DiscoveryServers; - if (patt.builtin.discovery_config.m_DiscoveryServers != m_att.builtin.discovery_config.m_DiscoveryServers) + if (converted_discovery_servers != temp_atts.builtin.discovery_config.m_DiscoveryServers) { - for (auto& transportDescriptor : m_att.userTransports) + for (auto& transportDescriptor : temp_atts.userTransports) { TCPTransportDescriptor* pT = dynamic_cast(transportDescriptor.get()); if (pT) @@ -1441,8 +1572,8 @@ void RTPSParticipantImpl::update_attributes( } // Check if there are changes - if (converted_discovery_servers != m_att.builtin.discovery_config.m_DiscoveryServers - || patt.userData != m_att.userData + if (converted_discovery_servers != temp_atts.builtin.discovery_config.m_DiscoveryServers + || patt.userData != temp_atts.userData || local_interfaces_changed) { update_pdp = true; @@ -1450,7 +1581,7 @@ void RTPSParticipantImpl::update_attributes( LocatorList_t modified_locators; // Update RTPSParticipantAttributes members - m_att.userData = patt.userData; + temp_atts.userData = patt.userData; // If there's no PDP don't process Discovery-related attributes. if (!pdp) @@ -1460,22 +1591,24 @@ void RTPSParticipantImpl::update_attributes( // Update listening locators on external locators { - namespace ExternalLocatorsProcessor = fastdds::rtps::ExternalLocatorsProcessor; + namespace external_locators = fastdds::rtps::ExternalLocatorsProcessor; if (local_interfaces_changed && internal_metatraffic_locators_) { - ExternalLocatorsProcessor::set_listening_locators(m_att.builtin.metatraffic_external_unicast_locators, - m_att.builtin.metatrafficUnicastLocatorList); + external_locators::set_listening_locators( + temp_atts.builtin.metatraffic_external_unicast_locators, + temp_atts.builtin.metatrafficUnicastLocatorList); } if (local_interfaces_changed && internal_default_locators_) { - ExternalLocatorsProcessor::set_listening_locators(m_att.default_external_unicast_locators, - m_att.defaultUnicastLocatorList); + external_locators::set_listening_locators( + temp_atts.default_external_unicast_locators, + temp_atts.defaultUnicastLocatorList); } } // Check that the remote servers list is consistent: all the already known remote servers must be included in // the list and either new remote servers are added or remote server listening locator is modified. - for (auto existing_server : m_att.builtin.discovery_config.m_DiscoveryServers) + for (auto existing_server : temp_atts.builtin.discovery_config.m_DiscoveryServers) { bool contained = false; for (auto incoming_server : converted_discovery_servers) @@ -1514,34 +1647,22 @@ void RTPSParticipantImpl::update_attributes( } } + // Update discovery information { std::lock_guard lock(*pdp->getMutex()); + local_participant_attributes_update_nts(temp_atts, pdp, this); - // Update user data - auto local_participant_proxy_data = pdp->getLocalParticipantProxyData(); - local_participant_proxy_data->m_userData.data_vec(m_att.userData); - - // Update metatraffic locators - for (auto locator : m_att.builtin.metatrafficMulticastLocatorList) - { - local_participant_proxy_data->metatraffic_locators.add_multicast_locator(locator); - } - for (auto locator : m_att.builtin.metatrafficUnicastLocatorList) - { - local_participant_proxy_data->metatraffic_locators.add_unicast_locator(locator); - } - - // Update default locators - for (auto locator : m_att.defaultUnicastLocatorList) + if (local_interfaces_changed && internal_default_locators_) { - local_participant_proxy_data->default_locators.add_unicast_locator(locator); + std::lock_guard _(endpoints_list_mutex); + update_endpoint_locators_if_default_nts(m_userWriterList, m_userReaderList, m_att, temp_atts, pdp); } if (local_interfaces_changed) { - createSenderResources(m_att.builtin.metatrafficMulticastLocatorList); - createSenderResources(m_att.builtin.metatrafficUnicastLocatorList); - createSenderResources(m_att.defaultUnicastLocatorList); + createSenderResources(temp_atts.builtin.metatrafficMulticastLocatorList); + createSenderResources(temp_atts.builtin.metatrafficUnicastLocatorList); + createSenderResources(temp_atts.defaultUnicastLocatorList); } if (!modified_locators.empty()) { @@ -1549,17 +1670,17 @@ void RTPSParticipantImpl::update_attributes( } // Update remote servers list - if (m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::CLIENT || - m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::SUPER_CLIENT || - m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::SERVER || - m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::BACKUP) + if (temp_atts.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::CLIENT || + temp_atts.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::SUPER_CLIENT || + temp_atts.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::SERVER || + temp_atts.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::BACKUP) { // Add incoming servers if we don't know about them already or the listening locator has been modified for (auto incoming_server : converted_discovery_servers) { eprosima::fastdds::rtps::RemoteServerList_t::iterator server_it; - for (server_it = m_att.builtin.discovery_config.m_DiscoveryServers.begin(); - server_it != m_att.builtin.discovery_config.m_DiscoveryServers.end(); server_it++) + for (server_it = temp_atts.builtin.discovery_config.m_DiscoveryServers.begin(); + server_it != temp_atts.builtin.discovery_config.m_DiscoveryServers.end(); server_it++) { if (server_it->guidPrefix == incoming_server.guidPrefix) { @@ -1576,21 +1697,21 @@ void RTPSParticipantImpl::update_attributes( break; } } - if (server_it == m_att.builtin.discovery_config.m_DiscoveryServers.end()) + if (server_it == temp_atts.builtin.discovery_config.m_DiscoveryServers.end()) { - m_att.builtin.discovery_config.m_DiscoveryServers.push_back(incoming_server); + temp_atts.builtin.discovery_config.m_DiscoveryServers.push_back(incoming_server); } } // Update the servers list in builtin protocols { std::unique_lock disc_lock(mp_builtinProtocols->getDiscoveryMutex()); - mp_builtinProtocols->m_DiscoveryServers = m_att.builtin.discovery_config.m_DiscoveryServers; + mp_builtinProtocols->m_DiscoveryServers = temp_atts.builtin.discovery_config.m_DiscoveryServers; } // Notify PDPServer - if (m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::SERVER || - m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::BACKUP) + if (temp_atts.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::SERVER || + temp_atts.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::BACKUP) { fastdds::rtps::PDPServer* pdp_server = static_cast(pdp); pdp_server->update_remote_servers_list(); @@ -1601,8 +1722,8 @@ void RTPSParticipantImpl::update_attributes( } } // Notify PDPClient - else if (m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::CLIENT || - m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::SUPER_CLIENT) + else if (temp_atts.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::CLIENT || + temp_atts.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol::SUPER_CLIENT) { fastdds::rtps::PDPClient* pdp_client = static_cast(pdp); pdp_client->update_remote_servers_list(); @@ -1614,6 +1735,12 @@ void RTPSParticipantImpl::update_attributes( } } } + + // Update the attributes data member + { + std::lock_guard guard(*mp_mutex); + m_att = temp_atts; + } } if (update_pdp) @@ -1757,8 +1884,8 @@ bool RTPSParticipantImpl::createAndAssociateReceiverswithEndpoint( createReceiverResources(attributes.multicastLocatorList, false, true, true); } - fastdds::rtps::ExternalLocatorsProcessor::set_listening_locators(attributes.external_unicast_locators, - attributes.unicastLocatorList); + namespace external_locators = fastdds::rtps::ExternalLocatorsProcessor; + external_locators::set_listening_locators(attributes.external_unicast_locators, attributes.unicastLocatorList); // Associate the Endpoint with ReceiverControlBlock assignEndpointListenResources(pend); @@ -2713,22 +2840,34 @@ void RTPSParticipantImpl::environment_file_has_changed() void RTPSParticipantImpl::get_default_metatraffic_locators() { - uint32_t metatraffic_multicast_port = m_att.port.getMulticastPort(domain_id_); + get_default_metatraffic_locators(m_att); +} + +void RTPSParticipantImpl::get_default_metatraffic_locators( + RTPSParticipantAttributes& att) +{ + uint32_t metatraffic_multicast_port = att.port.getMulticastPort(domain_id_); - m_network_Factory.getDefaultMetatrafficMulticastLocators(m_att.builtin.metatrafficMulticastLocatorList, + m_network_Factory.getDefaultMetatrafficMulticastLocators(att.builtin.metatrafficMulticastLocatorList, metatraffic_multicast_port); - m_network_Factory.NormalizeLocators(m_att.builtin.metatrafficMulticastLocatorList); + m_network_Factory.NormalizeLocators(att.builtin.metatrafficMulticastLocatorList); - m_network_Factory.getDefaultMetatrafficUnicastLocators(m_att.builtin.metatrafficUnicastLocatorList, + m_network_Factory.getDefaultMetatrafficUnicastLocators(att.builtin.metatrafficUnicastLocatorList, metatraffic_unicast_port_); - m_network_Factory.NormalizeLocators(m_att.builtin.metatrafficUnicastLocatorList); + m_network_Factory.NormalizeLocators(att.builtin.metatrafficUnicastLocatorList); } void RTPSParticipantImpl::get_default_unicast_locators() { - uint32_t unicast_port = metatraffic_unicast_port_ + m_att.port.offsetd3 - m_att.port.offsetd1; - m_network_Factory.getDefaultUnicastLocators(m_att.defaultUnicastLocatorList, unicast_port); - m_network_Factory.NormalizeLocators(m_att.defaultUnicastLocatorList); + get_default_unicast_locators(m_att); +} + +void RTPSParticipantImpl::get_default_unicast_locators( + RTPSParticipantAttributes& att) +{ + uint32_t unicast_port = metatraffic_unicast_port_ + att.port.offsetd3 - att.port.offsetd1; + m_network_Factory.getDefaultUnicastLocators(att.defaultUnicastLocatorList, unicast_port); + m_network_Factory.NormalizeLocators(att.defaultUnicastLocatorList); } bool RTPSParticipantImpl::is_participant_ignored( diff --git a/src/cpp/rtps/participant/RTPSParticipantImpl.h b/src/cpp/rtps/participant/RTPSParticipantImpl.h index 1bb84058698..e8e61b91386 100644 --- a/src/cpp/rtps/participant/RTPSParticipantImpl.h +++ b/src/cpp/rtps/participant/RTPSParticipantImpl.h @@ -762,11 +762,17 @@ class RTPSParticipantImpl */ void get_default_metatraffic_locators(); + void get_default_metatraffic_locators( + RTPSParticipantAttributes& att); + /** * Get default unicast locators when not provided by the user. */ void get_default_unicast_locators(); + void get_default_unicast_locators( + RTPSParticipantAttributes& att); + bool match_local_endpoints_ = true; bool should_match_local_endpoints( diff --git a/test/dds/communication/CMakeLists.txt b/test/dds/communication/CMakeLists.txt index d099dac93ce..5c02f1db47b 100644 --- a/test/dds/communication/CMakeLists.txt +++ b/test/dds/communication/CMakeLists.txt @@ -206,3 +206,7 @@ if(Python3_Interpreter_FOUND) endif() endforeach() endif() + +if(UNIX AND NOT(APPLE) AND NOT(QNXNTO) AND NOT(ANDROID)) + add_subdirectory(dyn_network) +endif() diff --git a/test/dds/communication/PubSubMain.cpp b/test/dds/communication/PubSubMain.cpp index 46553c9f64b..b713b92f002 100644 --- a/test/dds/communication/PubSubMain.cpp +++ b/test/dds/communication/PubSubMain.cpp @@ -52,7 +52,7 @@ void publisher_run( publisher->wait_discovery(wait); } - publisher->run(samples, loops, interval); + publisher->run(samples, 0, loops, interval); } int main( @@ -197,7 +197,7 @@ int main( DomainParticipantFactory::get_instance()->load_XML_profiles_file(xml_file); } - SubscriberModule subscriber(publishers, samples, fixed_type, zero_copy); + SubscriberModule subscriber(publishers, samples, fixed_type, zero_copy, false); PublisherModule publisher(exit_on_lost_liveliness, fixed_type, zero_copy); uint32_t result = 1; @@ -208,7 +208,7 @@ int main( if (subscriber.init(seed, magic)) { - result = subscriber.run(notexit, timeout) ? 0 : -1; + result = subscriber.run(notexit, 0, timeout) ? 0 : -1; } publisher_thread.join(); diff --git a/test/dds/communication/PublisherMain.cpp b/test/dds/communication/PublisherMain.cpp index 0d72b20ae5b..3124e1c592d 100644 --- a/test/dds/communication/PublisherMain.cpp +++ b/test/dds/communication/PublisherMain.cpp @@ -29,9 +29,11 @@ using namespace eprosima::fastdds::dds; * --seed * --wait * --samples + * --loops + * --interval * --magic * --xmlfile - * --interval + * --rescan */ int main( @@ -46,7 +48,9 @@ int main( uint32_t wait = 0; char* xml_file = nullptr; uint32_t samples = 4; + uint32_t loops = 0; uint32_t interval = 250; + uint32_t rescan_interval_seconds = 0; std::string magic; while (arg_count < argc) @@ -93,6 +97,16 @@ int main( samples = strtol(argv[arg_count], nullptr, 10); } + else if (strcmp(argv[arg_count], "--loops") == 0) + { + if (++arg_count >= argc) + { + std::cout << "--loops expects a parameter" << std::endl; + return -1; + } + + loops = strtol(argv[arg_count], nullptr, 10); + } else if (strcmp(argv[arg_count], "--interval") == 0) { if (++arg_count >= argc) @@ -123,6 +137,16 @@ int main( xml_file = argv[arg_count]; } + else if (strcmp(argv[arg_count], "--rescan") == 0) + { + if (++arg_count >= argc) + { + std::cout << "--rescan expects a parameter" << std::endl; + return -1; + } + + rescan_interval_seconds = strtol(argv[arg_count], nullptr, 10); + } else { std::cout << "Wrong argument " << argv[arg_count] << std::endl; @@ -146,7 +170,7 @@ int main( publisher.wait_discovery(wait); } - publisher.run(samples, 0, interval); + publisher.run(samples, rescan_interval_seconds, loops, interval); return 0; } diff --git a/test/dds/communication/PublisherModule.cpp b/test/dds/communication/PublisherModule.cpp index 35debd61a9c..e664090fbd1 100644 --- a/test/dds/communication/PublisherModule.cpp +++ b/test/dds/communication/PublisherModule.cpp @@ -131,6 +131,7 @@ void PublisherModule::wait_discovery( void PublisherModule::run( uint32_t samples, + const uint32_t rescan_interval, const uint32_t loops, uint32_t interval) { @@ -138,6 +139,22 @@ void PublisherModule::run( uint16_t index = 1; void* sample = nullptr; + std::thread net_rescan_thread([this, rescan_interval]() + { + if (rescan_interval > 0) + { + auto interval = std::chrono::seconds(rescan_interval); + while (run_) + { + std::this_thread::sleep_for(interval); + if (run_) + { + participant_->set_qos(participant_->get_qos()); + } + } + } + }); + while (run_ && (loops == 0 || loops > current_loop)) { if (zero_copy_) @@ -184,6 +201,9 @@ void PublisherModule::run( std::this_thread::sleep_for(std::chrono::milliseconds(interval)); } + + run_ = false; + net_rescan_thread.join(); } void PublisherModule::on_publication_matched( diff --git a/test/dds/communication/PublisherModule.hpp b/test/dds/communication/PublisherModule.hpp index fb144821a0a..593d224c6d6 100644 --- a/test/dds/communication/PublisherModule.hpp +++ b/test/dds/communication/PublisherModule.hpp @@ -19,6 +19,11 @@ #ifndef TEST_DDS_COMMUNICATION_PUBLISHERMODULE_HPP #define TEST_DDS_COMMUNICATION_PUBLISHERMODULE_HPP +#include +#include +#include +#include + #include #include #include @@ -27,9 +32,6 @@ #include "types/FixedSizedPubSubTypes.h" #include "types/HelloWorldPubSubTypes.h" -#include -#include - namespace eprosima { namespace fastdds { namespace dds { @@ -41,8 +43,8 @@ class PublisherModule PublisherModule( bool exit_on_lost_liveliness, - bool fixed_type = false, - bool zero_copy = false) + bool fixed_type, + bool zero_copy) : exit_on_lost_liveliness_(exit_on_lost_liveliness) , fixed_type_(zero_copy || fixed_type) // If zero copy active, fixed type is required , zero_copy_(zero_copy) @@ -80,8 +82,9 @@ class PublisherModule void run( uint32_t samples, - uint32_t loops = 0, - uint32_t interval = 250); + const uint32_t rescan_interval, + uint32_t loops, + uint32_t interval); private: @@ -93,7 +96,7 @@ class PublisherModule bool exit_on_lost_liveliness_ = false; bool fixed_type_ = false; bool zero_copy_ = false; - bool run_ = true; + std::atomic_bool run_{true}; DomainParticipant* participant_ = nullptr; TypeSupport type_; Publisher* publisher_ = nullptr; diff --git a/test/dds/communication/SubscriberMain.cpp b/test/dds/communication/SubscriberMain.cpp index e1e578c64c7..61ff6ddfe1e 100644 --- a/test/dds/communication/SubscriberMain.cpp +++ b/test/dds/communication/SubscriberMain.cpp @@ -26,13 +26,14 @@ using namespace eprosima::fastdds::dds; * --notexit * --fixed_type * --zero_copy + * --succeed_on_timeout * --seed * --samples * --magic + * --timeout * --xmlfile * --publishers - * --succeed_on_timeout - * --timeout + * --rescan */ int main( @@ -48,6 +49,7 @@ int main( uint32_t samples = 4; uint32_t publishers = 1; uint32_t timeout = 86400000; // 24 h in ms + uint32_t rescan_interval_seconds = 0; char* xml_file = nullptr; std::string magic; @@ -129,6 +131,16 @@ int main( publishers = strtol(argv[arg_count], nullptr, 10); } + else if (strcmp(argv[arg_count], "--rescan") == 0) + { + if (++arg_count >= argc) + { + std::cout << "--rescan expects a parameter" << std::endl; + return -1; + } + + rescan_interval_seconds = strtol(argv[arg_count], nullptr, 10); + } else { std::cout << "Wrong argument " << argv[arg_count] << std::endl; @@ -147,7 +159,7 @@ int main( if (subscriber.init(seed, magic)) { - return subscriber.run(notexit, timeout) ? 0 : -1; + return subscriber.run(notexit, rescan_interval_seconds, timeout) ? 0 : -1; } return -1; diff --git a/test/dds/communication/SubscriberModule.cpp b/test/dds/communication/SubscriberModule.cpp index a1d2b7b2a24..3e9d543472a 100644 --- a/test/dds/communication/SubscriberModule.cpp +++ b/test/dds/communication/SubscriberModule.cpp @@ -130,17 +130,35 @@ bool SubscriberModule::init( bool SubscriberModule::run( bool notexit, + const uint32_t rescan_interval, uint32_t timeout) { - return run_for(notexit, std::chrono::milliseconds(timeout)); + return run_for(notexit, rescan_interval, std::chrono::milliseconds(timeout)); } bool SubscriberModule::run_for( bool notexit, + const uint32_t rescan_interval, const std::chrono::milliseconds& timeout) { bool returned_value = false; + std::thread net_rescan_thread([this, rescan_interval]() + { + if (rescan_interval > 0) + { + auto interval = std::chrono::seconds(rescan_interval); + while (run_) + { + std::this_thread::sleep_for(interval); + if (run_) + { + participant_->set_qos(participant_->get_qos()); + } + } + } + }); + while (notexit && run_) { std::this_thread::sleep_for(std::chrono::milliseconds(250)); @@ -152,7 +170,7 @@ bool SubscriberModule::run_for( std::unique_lock lock(mutex_); returned_value = cv_.wait_for(lock, timeout, [&] { - if (succeeed_on_timeout_ && (std::chrono::steady_clock::now() - t0) > timeout) + if (succeed_on_timeout_ && (std::chrono::steady_clock::now() - t0) > timeout) { return true; } @@ -190,6 +208,9 @@ bool SubscriberModule::run_for( returned_value = false; } + run_ = false; + net_rescan_thread.join(); + return returned_value; } diff --git a/test/dds/communication/SubscriberModule.hpp b/test/dds/communication/SubscriberModule.hpp index e8559128440..6c50841ed07 100644 --- a/test/dds/communication/SubscriberModule.hpp +++ b/test/dds/communication/SubscriberModule.hpp @@ -19,6 +19,12 @@ #ifndef TEST_COMMUNICATION_SUBSCRIBER_HPP #define TEST_COMMUNICATION_SUBSCRIBER_HPP +#include +#include +#include +#include +#include + #include #include #include @@ -27,11 +33,6 @@ #include "types/FixedSizedPubSubTypes.h" #include "types/HelloWorldPubSubTypes.h" -#include -#include -#include -#include - namespace eprosima { namespace fastdds { namespace dds { @@ -44,14 +45,14 @@ class SubscriberModule SubscriberModule( const uint32_t publishers, const uint32_t max_number_samples, - bool fixed_type = false, - bool zero_copy = false, - bool succeed_on_timeout = false) + bool fixed_type, + bool zero_copy, + bool succeed_on_timeout) : publishers_(publishers) , max_number_samples_(max_number_samples) , fixed_type_(zero_copy || fixed_type) // If zero copy active, fixed type is required , zero_copy_(zero_copy) - , succeeed_on_timeout_(succeed_on_timeout) + , succeed_on_timeout_(succeed_on_timeout) { } @@ -84,10 +85,12 @@ class SubscriberModule bool run( bool notexit, - uint32_t timeout = 86400000); + const uint32_t rescan_interval, + uint32_t timeout); bool run_for( bool notexit, + const uint32_t rescan_interval, const std::chrono::milliseconds& timeout); private: @@ -101,8 +104,8 @@ class SubscriberModule std::map number_samples_; bool fixed_type_ = false; bool zero_copy_ = false; - bool run_ = true; - bool succeeed_on_timeout_ = false; + std::atomic_bool run_{true}; + bool succeed_on_timeout_ = false; DomainParticipant* participant_ = nullptr; TypeSupport type_; Subscriber* subscriber_ = nullptr; diff --git a/test/dds/communication/dyn_network/CMakeLists.txt b/test/dds/communication/dyn_network/CMakeLists.txt new file mode 100644 index 00000000000..0d95ac665e0 --- /dev/null +++ b/test/dds/communication/dyn_network/CMakeLists.txt @@ -0,0 +1,61 @@ +# Copyright 2024 Proyectos y Sistemas de Mantenimiento SL (eProsima). +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +message(STATUS "Configuring dynamic network interfaces tests") + +# Find docker +find_program(DOCKER_EXECUTABLE docker) +if(NOT DOCKER_EXECUTABLE) + message(FATAL_ERROR "Docker not found") +endif() + +set(SHELL_EXECUTABLE "") +set(TINYXML2_LIB_DIR_COMPOSE_VOLUME "") +set(TINYXML2_LIB_DIR_COMPOSE_LD_LIBRARY_PATH "") + +# Linux configurations +if(UNIX AND NOT(APPLE) AND NOT(QNXNTO) AND NOT(ANDROID)) + # Find bash + find_program(BASH_EXECUTABLE bash) + if(NOT BASH_EXECUTABLE) + message(FATAL_ERROR "bash not found") + endif() + + set(SHELL_EXECUTABLE ${BASH_EXECUTABLE}) + +# Windows configurations +elseif(WIN32) + # We don't know which docker image to use for Windows yet + message(FATAL_ERROR "Windows not supported yet") + +# Unsupported platform +else() + message(FATAL_ERROR "Unsupported platform") +endif() + +# Configure TinyXML2 library path if installed in user library path +if(NOT (TINYXML2_FROM_SOURCE OR TINYXML2_FROM_THIRDPARTY)) + get_filename_component(TINYXML2_LIB_DIR ${TINYXML2_LIBRARY} DIRECTORY) + set(TINYXML2_LIB_DIR_COMPOSE_VOLUME "- ${TINYXML2_LIB_DIR}:${CMAKE_INSTALL_PREFIX}/${DATA_INSTALL_DIR}/fastdds:ro") + set(TINYXML2_LIB_DIR_COMPOSE_LD_LIBRARY_PATH ":${CMAKE_INSTALL_PREFIX}/${DATA_INSTALL_DIR}/fastdds") +endif() + +configure_file(Dockerfile + ${CMAKE_CURRENT_BINARY_DIR}/Dockerfile @ONLY) +configure_file(dynamic_interfaces.compose.yml + ${CMAKE_CURRENT_BINARY_DIR}/dynamic_interfaces.compose.yml @ONLY) +configure_file(launch_subscriber.bash + ${CMAKE_CURRENT_BINARY_DIR}/launch_subscriber.bash @ONLY) +add_test(NAME dds.communication.dynamic_interfaces + COMMAND ${DOCKER_EXECUTABLE} compose -f ${CMAKE_CURRENT_BINARY_DIR}/dynamic_interfaces.compose.yml up + WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}) diff --git a/test/dds/communication/dyn_network/Dockerfile b/test/dds/communication/dyn_network/Dockerfile new file mode 100644 index 00000000000..ebdcae10bba --- /dev/null +++ b/test/dds/communication/dyn_network/Dockerfile @@ -0,0 +1,27 @@ +# Copyright 2024 Proyectos y Sistemas de Mantenimiento SL (eProsima). +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Tag, branch, or commit in github.com/eProsima/DDS-Suite +ARG ubuntu_version=22.04 +FROM ubuntu:$ubuntu_version AS ubuntu-net-tools + +# Needed for a dependency that forces to set timezone +ENV TZ=Europe/Madrid +RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone + +# Avoids using interactions during building +ENV DEBIAN_FRONTEND=noninteractive + +# Install apt dependencies +RUN apt-get update && apt-get install --yes net-tools && rm -rf /var/lib/apt/lists/* diff --git a/test/dds/communication/dyn_network/dynamic_interfaces.compose.yml b/test/dds/communication/dyn_network/dynamic_interfaces.compose.yml new file mode 100644 index 00000000000..9d45abbbc48 --- /dev/null +++ b/test/dds/communication/dyn_network/dynamic_interfaces.compose.yml @@ -0,0 +1,42 @@ +# Copyright 2024 Proyectos y Sistemas de Mantenimiento SL (eProsima). +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +version: "3" + +services: + publisher: + image: ubuntu:22.04 + volumes: + - @PROJECT_BINARY_DIR@:@PROJECT_BINARY_DIR@ + - @fastcdr_LIB_DIR@:@fastcdr_LIB_DIR@ + @TINYXML2_LIB_DIR_COMPOSE_VOLUME@ + environment: + LD_LIBRARY_PATH: @PROJECT_BINARY_DIR@/src/cpp:@fastcdr_LIB_DIR@@TINYXML2_LIB_DIR_COMPOSE_LD_LIBRARY_PATH@ + EXAMPLE_DIR: @PROJECT_BINARY_DIR@/test/dds/communication + command: @SHELL_EXECUTABLE@ -c "$${EXAMPLE_DIR}/DDSCommunicationPublisher --xmlfile $${EXAMPLE_DIR}/simple_reliable_profile.xml --wait 1 --samples 10 --loops 1 --seed 0 --magic T" + + subscriber: + build: . + image: ubuntu-net-tools:22.04 + privileged: true + volumes: + - @PROJECT_BINARY_DIR@:@PROJECT_BINARY_DIR@ + - @fastcdr_LIB_DIR@:@fastcdr_LIB_DIR@ + @TINYXML2_LIB_DIR_COMPOSE_VOLUME@ + environment: + LD_LIBRARY_PATH: @PROJECT_BINARY_DIR@/src/cpp:@fastcdr_LIB_DIR@@TINYXML2_LIB_DIR_COMPOSE_LD_LIBRARY_PATH@ + EXAMPLE_DIR: @PROJECT_BINARY_DIR@/test/dds/communication + working_dir: @PROJECT_BINARY_DIR@/test/dds/communication + command: @SHELL_EXECUTABLE@ "dyn_network/launch_subscriber.bash" + depends_on: + - publisher diff --git a/test/dds/communication/dyn_network/launch_subscriber.bash b/test/dds/communication/dyn_network/launch_subscriber.bash new file mode 100755 index 00000000000..9ee62e1338b --- /dev/null +++ b/test/dds/communication/dyn_network/launch_subscriber.bash @@ -0,0 +1,33 @@ +# Copyright 2019 Proyectos y Sistemas de Mantenimiento SL (eProsima). +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +#!/bin/bash + +# Note: This script is intended to be used in a privileged container, since it requires to bring down and up the eth0 interface. + +echo "Putting down eth0 interface..." +ifconfig eth0 down + +echo "Launching subscriber..." +${EXAMPLE_DIR}/DDSCommunicationSubscriber --xmlfile ${EXAMPLE_DIR}/simple_reliable_profile.xml --samples 10 --seed 0 --magic T --rescan 2 & +subs_pid=$! +echo "Subscriber launched." + +echo "Waiting 2 seconds and bring up eth0 interface..." +sleep 2s +ifconfig eth0 up +echo "eth0 interface is up." + +echo "Waiting 3s for the subscriber (process id $subs_pid) to finish..." +wait $subs_pid