Skip to content

Commit 3aad708

Browse files
MiguelCompanymergify[bot]
MiguelCompany
authored and
mergify[bot]
committed
Fix issues in Dynamic Network Interfaces (#5282)
* Refs #21690. Parse `--rescan` argument on communication applications. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Implement rescan mechanism. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Add docker infrastructure. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Add CMake infrastructure. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Ensure same domain and topic name are used. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Add `--loops` argument to publisher. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Publisher exits after publishing all samples. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Improve subscriber script. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Add test. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Make publisher wait subscriber. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Possible fix. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Clear locators before recalculating them. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Move local participant proxy update to PDP. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Improve new method's logic. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Include what you use. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Add empty method to update endpoint locators. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Add implementation for `update_endpoint_locators_if_default_nts`. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Compare against old default locators. Signed-off-by: Miguel Company <[email protected]> * Refs #21690. Update locators in attributes. Signed-off-by: Miguel Company <[email protected]> * Refs #17283. Avoid early return on `PDP::local_participant_attributes_update_nts`. Signed-off-by: Miguel Company <[email protected]> * Refs #17283. Apply suggestions. Signed-off-by: Miguel Company <[email protected]> --------- Signed-off-by: Miguel Company <[email protected]> (cherry picked from commit 91bd7c8)
1 parent 3be34b7 commit 3aad708

21 files changed

+513
-69
lines changed

‎src/cpp/rtps/builtin/discovery/participant/PDP.cpp

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,8 @@
6262
#endif // if HAVE_SECURITY
6363
#include <utils/shared_mutex.hpp>
6464
#include <utils/TimeConversion.hpp>
65+
#include <rtps/writer/BaseWriter.hpp>
66+
#include <rtps/reader/BaseReader.hpp>
6567

6668
namespace eprosima {
6769
namespace fastdds {
@@ -1698,6 +1700,121 @@ void PDP::add_builtin_security_attributes(
16981700

16991701
#endif // HAVE_SECURITY
17001702

1703+
void PDP::local_participant_attributes_update_nts(
1704+
const RTPSParticipantAttributes& new_atts)
1705+
{
1706+
// Update user data
1707+
auto participant_data = getLocalParticipantProxyData();
1708+
participant_data->m_userData.data_vec(new_atts.userData);
1709+
1710+
// If we are intraprocess only, we do not need to update locators
1711+
bool announce_locators = !mp_RTPSParticipant->is_intraprocess_only();
1712+
if (announce_locators)
1713+
{
1714+
// Clear all locators
1715+
participant_data->metatraffic_locators.unicast.clear();
1716+
participant_data->metatraffic_locators.multicast.clear();
1717+
participant_data->default_locators.unicast.clear();
1718+
participant_data->default_locators.multicast.clear();
1719+
1720+
// Update default locators
1721+
for (const Locator_t& loc : new_atts.defaultUnicastLocatorList)
1722+
{
1723+
participant_data->default_locators.add_unicast_locator(loc);
1724+
}
1725+
for (const Locator_t& loc : new_atts.defaultMulticastLocatorList)
1726+
{
1727+
participant_data->default_locators.add_multicast_locator(loc);
1728+
}
1729+
1730+
// Update metatraffic locators
1731+
for (const auto& locator : new_atts.builtin.metatrafficUnicastLocatorList)
1732+
{
1733+
participant_data->metatraffic_locators.add_unicast_locator(locator);
1734+
}
1735+
if (!new_atts.builtin.avoid_builtin_multicast || participant_data->metatraffic_locators.unicast.empty())
1736+
{
1737+
for (const auto& locator : new_atts.builtin.metatrafficMulticastLocatorList)
1738+
{
1739+
participant_data->metatraffic_locators.add_multicast_locator(locator);
1740+
}
1741+
}
1742+
1743+
fastdds::rtps::network::external_locators::add_external_locators(*participant_data,
1744+
new_atts.builtin.metatraffic_external_unicast_locators,
1745+
new_atts.default_external_unicast_locators);
1746+
}
1747+
}
1748+
1749+
void PDP::update_endpoint_locators_if_default_nts(
1750+
const std::vector<BaseWriter*>& writers,
1751+
const std::vector<BaseReader*>& readers,
1752+
const RTPSParticipantAttributes& old_atts,
1753+
const RTPSParticipantAttributes& new_atts)
1754+
{
1755+
// Check if default locators have changed
1756+
const auto& old_default_unicast = old_atts.defaultUnicastLocatorList;
1757+
const auto& old_default_multicast = old_atts.defaultMulticastLocatorList;
1758+
const auto& new_default_unicast = new_atts.defaultUnicastLocatorList;
1759+
const auto& new_default_multicast = new_atts.defaultMulticastLocatorList;
1760+
1761+
// Early return if there is no change in default unicast locators
1762+
if ((old_default_unicast == new_default_unicast) &&
1763+
(old_default_multicast == new_default_multicast))
1764+
{
1765+
return;
1766+
}
1767+
1768+
// Update proxies of endpoints with default configured locators
1769+
EDP* edp = get_edp();
1770+
for (BaseWriter* writer : writers)
1771+
{
1772+
if ((old_default_multicast == writer->getAttributes().multicastLocatorList) &&
1773+
(old_default_unicast == writer->getAttributes().unicastLocatorList))
1774+
{
1775+
writer->getAttributes().multicastLocatorList = new_default_multicast;
1776+
writer->getAttributes().unicastLocatorList = new_default_unicast;
1777+
1778+
WriterProxyData* wdata = nullptr;
1779+
GUID_t participant_guid;
1780+
wdata = addWriterProxyData(writer->getGuid(), participant_guid,
1781+
[](WriterProxyData* proxy, bool is_update, const ParticipantProxyData& participant)
1782+
{
1783+
static_cast<void>(is_update);
1784+
assert(is_update);
1785+
1786+
proxy->set_locators(participant.default_locators);
1787+
return true;
1788+
});
1789+
assert(wdata != nullptr);
1790+
edp->process_writer_proxy_data(writer, wdata);
1791+
}
1792+
}
1793+
for (BaseReader* reader : readers)
1794+
{
1795+
if ((old_default_multicast == reader->getAttributes().multicastLocatorList) &&
1796+
(old_default_unicast == reader->getAttributes().unicastLocatorList))
1797+
{
1798+
reader->getAttributes().multicastLocatorList = new_default_multicast;
1799+
reader->getAttributes().unicastLocatorList = new_default_unicast;
1800+
1801+
ReaderProxyData* rdata = nullptr;
1802+
GUID_t participant_guid;
1803+
rdata = addReaderProxyData(reader->getGuid(), participant_guid,
1804+
[](ReaderProxyData* proxy, bool is_update, const ParticipantProxyData& participant)
1805+
{
1806+
static_cast<void>(is_update);
1807+
assert(is_update);
1808+
1809+
proxy->set_locators(participant.default_locators);
1810+
return true;
1811+
});
1812+
assert(rdata != nullptr);
1813+
edp->process_reader_proxy_data(reader, rdata);
1814+
}
1815+
}
1816+
}
1817+
17011818
} /* namespace rtps */
17021819
} /* namespace fastdds */
17031820
} /* namespace eprosima */

‎src/cpp/rtps/builtin/discovery/participant/PDP.h

Lines changed: 27 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -22,13 +22,27 @@
2222
#ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
2323

2424
#include <atomic>
25+
#include <cstdint>
2526
#include <functional>
2627
#include <memory>
2728
#include <mutex>
29+
#include <string>
30+
#include <vector>
2831

32+
#include <fastcdr/cdr/fixed_size_string.hpp>
33+
34+
#include <fastdds/dds/core/Time_t.hpp>
35+
#include <fastdds/dds/core/policy/ParameterTypes.hpp>
2936
#include <fastdds/dds/core/policy/QosPolicies.hpp>
37+
#include <fastdds/rtps/attributes/ReaderAttributes.hpp>
3038
#include <fastdds/rtps/attributes/RTPSParticipantAttributes.hpp>
39+
#include <fastdds/rtps/attributes/WriterAttributes.hpp>
40+
#include <fastdds/rtps/common/CDRMessage_t.hpp>
3141
#include <fastdds/rtps/common/Guid.hpp>
42+
#include <fastdds/rtps/common/GuidPrefix_t.hpp>
43+
#include <fastdds/rtps/common/InstanceHandle.hpp>
44+
#include <fastdds/rtps/common/LocatorList.hpp>
45+
#include <fastdds/rtps/common/Types.hpp>
3246
#include <fastdds/rtps/common/WriteParams.hpp>
3347
#include <fastdds/rtps/history/IPayloadPool.hpp>
3448
#include <fastdds/rtps/participant/ParticipantDiscoveryInfo.hpp>
@@ -65,19 +79,11 @@ class TypeIdentifier;
6579

6680
namespace rtps {
6781

68-
class PDPServerListener;
69-
class PDPEndpoints;
70-
71-
} // namespace rtps
72-
} // namespace fastdds
73-
74-
namespace fastdds {
75-
namespace rtps {
76-
77-
class RTPSWriter;
78-
class RTPSReader;
82+
class BaseWriter;
83+
class BaseReader;
7984
class WriterHistory;
8085
class ReaderHistory;
86+
struct RTPSParticipantAllocationAttributes;
8187
class RTPSParticipantImpl;
8288
class RTPSParticipantListener;
8389
class BuiltinProtocols;
@@ -87,6 +93,7 @@ class ReaderProxyData;
8793
class WriterProxyData;
8894
class ParticipantProxyData;
8995
class ReaderListener;
96+
class PDPEndpoints;
9097
class PDPListener;
9198
class PDPServerListener;
9299
class ITopicPayloadPool;
@@ -484,6 +491,15 @@ class PDP : public fastdds::statistics::rtps::IProxyQueryable
484491

485492
#endif // FASTDDS_STATISTICS
486493

494+
virtual void local_participant_attributes_update_nts(
495+
const RTPSParticipantAttributes& new_atts);
496+
497+
virtual void update_endpoint_locators_if_default_nts(
498+
const std::vector<BaseWriter*>& writers,
499+
const std::vector<BaseReader*>& readers,
500+
const RTPSParticipantAttributes& old_atts,
501+
const RTPSParticipantAttributes& new_atts);
502+
487503
protected:
488504

489505
//!Pointer to the builtin protocols object.

‎src/cpp/rtps/participant/RTPSParticipantImpl.cpp

Lines changed: 6 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1445,6 +1445,7 @@ void RTPSParticipantImpl::update_attributes(
14451445
if (internal_metatraffic_locators_)
14461446
{
14471447
LocatorList_t metatraffic_unicast_locator_list = temp_atts.builtin.metatrafficUnicastLocatorList;
1448+
temp_atts.builtin.metatrafficUnicastLocatorList.clear();
14481449
get_default_metatraffic_locators(temp_atts);
14491450
if (!(metatraffic_unicast_locator_list == temp_atts.builtin.metatrafficUnicastLocatorList))
14501451
{
@@ -1455,6 +1456,7 @@ void RTPSParticipantImpl::update_attributes(
14551456
if (internal_default_locators_)
14561457
{
14571458
LocatorList_t default_unicast_locator_list = temp_atts.defaultUnicastLocatorList;
1459+
temp_atts.defaultUnicastLocatorList.clear();
14581460
get_default_unicast_locators(temp_atts);
14591461
if (!(default_unicast_locator_list == temp_atts.defaultUnicastLocatorList))
14601462
{
@@ -1529,25 +1531,12 @@ void RTPSParticipantImpl::update_attributes(
15291531

15301532
{
15311533
std::lock_guard<std::recursive_mutex> lock(*pdp->getMutex());
1534+
pdp->local_participant_attributes_update_nts(temp_atts);
15321535

1533-
// Update user data
1534-
auto local_participant_proxy_data = pdp->getLocalParticipantProxyData();
1535-
local_participant_proxy_data->m_userData.data_vec(temp_atts.userData);
1536-
1537-
// Update metatraffic locators
1538-
for (auto locator : temp_atts.builtin.metatrafficMulticastLocatorList)
1539-
{
1540-
local_participant_proxy_data->metatraffic_locators.add_multicast_locator(locator);
1541-
}
1542-
for (auto locator : temp_atts.builtin.metatrafficUnicastLocatorList)
1543-
{
1544-
local_participant_proxy_data->metatraffic_locators.add_unicast_locator(locator);
1545-
}
1546-
1547-
// Update default locators
1548-
for (auto locator : temp_atts.defaultUnicastLocatorList)
1536+
if (local_interfaces_changed && internal_default_locators_)
15491537
{
1550-
local_participant_proxy_data->default_locators.add_unicast_locator(locator);
1538+
std::lock_guard<shared_mutex> _(endpoints_list_mutex);
1539+
pdp->update_endpoint_locators_if_default_nts(m_userWriterList, m_userReaderList, m_att, temp_atts);
15511540
}
15521541

15531542
if (local_interfaces_changed)

‎test/dds/communication/CMakeLists.txt

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -315,3 +315,7 @@ if(Python3_Interpreter_FOUND)
315315
endif()
316316

317317
endif()
318+
319+
if(UNIX AND NOT(APPLE) AND NOT(QNXNTO) AND NOT(ANDROID))
320+
add_subdirectory(dyn_network)
321+
endif()

‎test/dds/communication/PubSubMain.cpp

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ void publisher_run(
5151
publisher->wait_discovery(wait);
5252
}
5353

54-
publisher->run(samples, loops, interval);
54+
publisher->run(samples, 0, loops, interval);
5555
}
5656

5757
int main(
@@ -196,7 +196,7 @@ int main(
196196
DomainParticipantFactory::get_instance()->load_XML_profiles_file(xml_file);
197197
}
198198

199-
SubscriberModule subscriber(publishers, samples, fixed_type, zero_copy);
199+
SubscriberModule subscriber(publishers, samples, fixed_type, zero_copy, false, false);
200200
PublisherModule publisher(exit_on_lost_liveliness, fixed_type, zero_copy);
201201

202202
uint32_t result = 1;
@@ -207,7 +207,7 @@ int main(
207207

208208
if (subscriber.init(seed, magic))
209209
{
210-
result = subscriber.run(notexit, timeout) ? 0 : -1;
210+
result = subscriber.run(notexit, 0, timeout) ? 0 : -1;
211211
}
212212

213213
publisher_thread.join();

‎test/dds/communication/PublisherMain.cpp

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,11 @@ using namespace eprosima::fastdds::dds;
2929
* --seed <int>
3030
* --wait <int>
3131
* --samples <int>
32+
* --loops <int>
33+
* --interval <int>
3234
* --magic <str>
3335
* --xmlfile <path>
34-
* --interval <int>
36+
* --rescan <int>
3537
*/
3638

3739
int main(
@@ -46,7 +48,9 @@ int main(
4648
uint32_t wait = 0;
4749
char* xml_file = nullptr;
4850
uint32_t samples = 4;
51+
uint32_t loops = 0;
4952
uint32_t interval = 250;
53+
uint32_t rescan_interval_seconds = 0;
5054
std::string magic;
5155

5256
while (arg_count < argc)
@@ -93,6 +97,16 @@ int main(
9397

9498
samples = strtol(argv[arg_count], nullptr, 10);
9599
}
100+
else if (strcmp(argv[arg_count], "--loops") == 0)
101+
{
102+
if (++arg_count >= argc)
103+
{
104+
std::cout << "--loops expects a parameter" << std::endl;
105+
return -1;
106+
}
107+
108+
loops = strtol(argv[arg_count], nullptr, 10);
109+
}
96110
else if (strcmp(argv[arg_count], "--interval") == 0)
97111
{
98112
if (++arg_count >= argc)
@@ -123,6 +137,16 @@ int main(
123137

124138
xml_file = argv[arg_count];
125139
}
140+
else if (strcmp(argv[arg_count], "--rescan") == 0)
141+
{
142+
if (++arg_count >= argc)
143+
{
144+
std::cout << "--rescan expects a parameter" << std::endl;
145+
return -1;
146+
}
147+
148+
rescan_interval_seconds = strtol(argv[arg_count], nullptr, 10);
149+
}
126150
else
127151
{
128152
std::cout << "Wrong argument " << argv[arg_count] << std::endl;
@@ -146,7 +170,7 @@ int main(
146170
publisher.wait_discovery(wait);
147171
}
148172

149-
publisher.run(samples, 0, interval);
173+
publisher.run(samples, rescan_interval_seconds, loops, interval);
150174
return 0;
151175
}
152176

‎test/dds/communication/PublisherModule.cpp

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,13 +134,30 @@ void PublisherModule::wait_discovery(
134134

135135
void PublisherModule::run(
136136
uint32_t samples,
137+
const uint32_t rescan_interval,
137138
const uint32_t loops,
138139
uint32_t interval)
139140
{
140141
uint32_t current_loop = 0;
141142
uint16_t index = 1;
142143
void* sample = nullptr;
143144

145+
std::thread net_rescan_thread([this, rescan_interval]()
146+
{
147+
if (rescan_interval > 0)
148+
{
149+
auto interval = std::chrono::seconds(rescan_interval);
150+
while (run_)
151+
{
152+
std::this_thread::sleep_for(interval);
153+
if (run_)
154+
{
155+
participant_->set_qos(participant_->get_qos());
156+
}
157+
}
158+
}
159+
});
160+
144161
while (run_ && (loops == 0 || loops > current_loop))
145162
{
146163
if (zero_copy_)
@@ -187,6 +204,9 @@ void PublisherModule::run(
187204

188205
std::this_thread::sleep_for(std::chrono::milliseconds(interval));
189206
}
207+
208+
run_ = false;
209+
net_rescan_thread.join();
190210
}
191211

192212
void PublisherModule::on_publication_matched(

0 commit comments

Comments
 (0)