diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.cpp b/src/cpp/fastdds/publisher/DataWriterImpl.cpp index 7e03b504306..4aa28eb27fc 100644 --- a/src/cpp/fastdds/publisher/DataWriterImpl.cpp +++ b/src/cpp/fastdds/publisher/DataWriterImpl.cpp @@ -1320,6 +1320,8 @@ void DataWriterImpl::InnerDataWriterListener::on_writer_matched( RTPSWriter* /*writer*/, const MatchingInfo& info) { + std::lock_guard scoped_lock(matching_info_mutex_); + data_writer_->update_publication_matched_status(info); StatusMask notify_status = StatusMask::publication_matched(); diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.hpp b/src/cpp/fastdds/publisher/DataWriterImpl.hpp index 81b05d43202..816047a7d89 100644 --- a/src/cpp/fastdds/publisher/DataWriterImpl.hpp +++ b/src/cpp/fastdds/publisher/DataWriterImpl.hpp @@ -487,7 +487,10 @@ class DataWriterImpl : protected rtps::IReaderDataFilter const uint32_t& status_id); #endif //FASTDDS_STATISTICS + private: + DataWriterImpl* data_writer_; + std::mutex matching_info_mutex_; } writer_listener_; diff --git a/src/cpp/fastdds/subscriber/DataReaderImpl.cpp b/src/cpp/fastdds/subscriber/DataReaderImpl.cpp index d0aa76fb1dc..a74e04e4db1 100644 --- a/src/cpp/fastdds/subscriber/DataReaderImpl.cpp +++ b/src/cpp/fastdds/subscriber/DataReaderImpl.cpp @@ -941,6 +941,8 @@ void DataReaderImpl::InnerDataReaderListener::on_reader_matched( RTPSReader* /*reader*/, const MatchingInfo& info) { + std::lock_guard scoped_lock(matching_info_mutex_); + data_reader_->update_subscription_matched_status(info); StatusMask notify_status = StatusMask::subscription_matched(); @@ -1615,11 +1617,11 @@ ReturnCode_t DataReaderImpl::check_qos( qos.history().depth > qos.resource_limits().max_samples_per_instance) { EPROSIMA_LOG_WARNING(RTPS_QOS_CHECK, - "HISTORY DEPTH '" << qos.history().depth << - "' is inconsistent with max_samples_per_instance: '" << - qos.resource_limits().max_samples_per_instance << - "'. Consistency rule: depth <= max_samples_per_instance." << - " Effectively using max_samples_per_instance as depth."); + "HISTORY DEPTH '" << qos.history().depth + << "' is inconsistent with max_samples_per_instance: '" + << qos.resource_limits().max_samples_per_instance + << "'. Consistency rule: depth <= max_samples_per_instance." + << " Effectively using max_samples_per_instance as depth."); } // Check for nanoseconds in all duration policies if (!utils::is_duration_consistent(qos.deadline().period)) diff --git a/src/cpp/fastdds/subscriber/DataReaderImpl.hpp b/src/cpp/fastdds/subscriber/DataReaderImpl.hpp index 275363ea08a..93fd7b894f4 100644 --- a/src/cpp/fastdds/subscriber/DataReaderImpl.hpp +++ b/src/cpp/fastdds/subscriber/DataReaderImpl.hpp @@ -481,7 +481,10 @@ class DataReaderImpl const uint32_t& status_id); #endif //FASTDDS_STATISTICS + private: + DataReaderImpl* data_reader_; + std::mutex matching_info_mutex_; } reader_listener_; diff --git a/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp b/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp index 3a557239fae..d8f5e468d2c 100644 --- a/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp +++ b/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp @@ -97,8 +97,8 @@ bool EDP::new_reader_proxy_data( const fastdds::rtps::ContentFilterProperty* content_filter) { EPROSIMA_LOG_INFO(RTPS_EDP, - "Adding " << rtps_reader->getGuid().entityId << " in topic " << - topic.topic_name.to_string()); + "Adding " << rtps_reader->getGuid().entityId << " in topic " + << topic.topic_name.to_string()); auto init_fun = [this, rtps_reader, &topic, &qos, content_filter]( ReaderProxyData* rpd, @@ -243,8 +243,8 @@ dds::ReturnCode_t EDP::new_reader_proxy_data( const fastdds::rtps::ContentFilterProperty* content_filter) { EPROSIMA_LOG_INFO(RTPS_EDP, - "Adding " << rtps_reader->getGuid().entityId << " in topic " << - topic.topic_name.to_string()); + "Adding " << rtps_reader->getGuid().entityId << " in topic " + << topic.topic_name.to_string()); auto init_fun = [this, rtps_reader, &topic, &sub_builtin_topic_data, content_filter]( ReaderProxyData* rpd, @@ -388,8 +388,8 @@ bool EDP::new_writer_proxy_data( const fastdds::dds::WriterQos& qos) { EPROSIMA_LOG_INFO(RTPS_EDP, - "Adding " << rtps_writer->getGuid().entityId << " in topic " << - topic.topic_name.to_string()); + "Adding " << rtps_writer->getGuid().entityId << " in topic " + << topic.topic_name.to_string()); auto init_fun = [this, rtps_writer, &topic, &qos]( WriterProxyData* wpd, @@ -521,8 +521,8 @@ dds::ReturnCode_t EDP::new_writer_proxy_data( bool should_send_opt_qos) { EPROSIMA_LOG_INFO(RTPS_EDP, - "Adding " << rtps_writer->getGuid().entityId << " in topic " << - topic.topic_name.to_string()); + "Adding " << rtps_writer->getGuid().entityId << " in topic " + << topic.topic_name.to_string()); auto init_fun = [this, rtps_writer, &topic, &pub_builtin_topic_data]( WriterProxyData* wpd, @@ -798,8 +798,8 @@ bool EDP::unpairWriterProxy( #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - ReaderListener* listener = nullptr; - if (nullptr != (listener = r.get_listener())) + ReaderListener* listener = r.get_listener(); + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; @@ -833,8 +833,8 @@ bool EDP::unpairReaderProxy( participant_guid, reader_guid); #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - WriterListener* listener = nullptr; - if (nullptr != (listener = w.get_listener())) + WriterListener* listener = w.get_listener(); + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; @@ -886,8 +886,10 @@ bool EDP::valid_matching( if (wdata->topic_kind != rdata->topic_kind) { EPROSIMA_LOG_WARNING(RTPS_EDP, "INCOMPATIBLE QOS:Remote Reader " << rdata->guid << " is publishing in topic " - << rdata->topic_name << "(keyed:" << rdata->topic_kind << - "), local writer publishes as keyed: " << wdata->topic_kind); + << rdata->topic_name << "(keyed:" + << rdata->topic_kind + << "), local writer publishes as keyed: " + << wdata->topic_kind); reason.set(MatchingFailureMask::inconsistent_topic); return false; } @@ -904,8 +906,8 @@ bool EDP::valid_matching( //Means our writer is BE but the reader wants RE { EPROSIMA_LOG_WARNING(RTPS_EDP, "INCOMPATIBLE QOS (topic: " << rdata->topic_name << "):Remote Reader " - << rdata->guid << - " is Reliable and local writer is BE "); + << rdata->guid + << " is Reliable and local writer is BE "); incompatible_qos.set(fastdds::dds::RELIABILITY_QOS_POLICY_ID); } @@ -913,7 +915,8 @@ bool EDP::valid_matching( { // TODO (MCC) Change log message EPROSIMA_LOG_WARNING(RTPS_EDP, "INCOMPATIBLE QOS (topic: " << rdata->topic_name << "):RemoteReader " - << rdata->guid << + << rdata->guid + << " has TRANSIENT_LOCAL DURABILITY and we offer VOLATILE"); incompatible_qos.set(fastdds::dds::DURABILITY_QOS_POLICY_ID); } @@ -1140,43 +1143,42 @@ bool EDP::pairingReader( { static_cast(reader_guid); // Void cast to force usage if we don't have LOG_INFOs EPROSIMA_LOG_INFO(RTPS_EDP_MATCH, - "WP:" << wdatait->guid << " match R:" << reader_guid << ". RLoc:" << - wdatait->remote_locators); + "WP:" << wdatait->guid << " match R:" << reader_guid << ". RLoc:" + << wdatait->remote_locators); //MATCHED AND ADDED CORRECTLY: - if (reader->get_listener() != nullptr) + ReaderListener* listener = reader->get_listener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = writer_guid; - reader->get_listener()->on_reader_matched(reader, info); + listener->on_reader_matched(reader, info); } } #endif // if HAVE_SECURITY } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && reader->get_listener() != nullptr) + ReaderListener* listener = reader->get_listener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - reader->get_listener()->on_requested_incompatible_qos(reader, incompatible_qos); + listener->on_requested_incompatible_qos(reader, incompatible_qos); mp_PDP->notify_incompatible_qos_matching(R->getGuid(), wdatait->guid, incompatible_qos); } - //EPROSIMA_LOG_INFO(RTPS_EDP,RTPS_CYAN<<"Valid Matching to writerProxy: "<guid<matched_writer_is_matched(wdatait->guid) - && reader->matched_writer_remove(wdatait->guid)) + if (reader->matched_writer_is_matched(wdatait->guid) && reader->matched_writer_remove(wdatait->guid)) { #if HAVE_SECURITY - mp_RTPSParticipant->security_manager().remove_writer(reader_guid, participant_guid, - wdatait->guid); + mp_RTPSParticipant->security_manager().remove_writer(reader_guid, participant_guid, wdatait->guid); #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - if (reader->get_listener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = writer_guid; - reader->get_listener()->on_reader_matched(reader, info); + listener->on_reader_matched(reader, info); } } } @@ -1233,24 +1235,26 @@ bool EDP::pairingWriter( if (writer->matched_reader_add_edp(*rdatait)) { EPROSIMA_LOG_INFO(RTPS_EDP_MATCH, - "RP:" << rdatait->guid << " match W:" << writer_guid << ". WLoc:" << - rdatait->remote_locators); + "RP:" << rdatait->guid << " match W:" << writer_guid << ". WLoc:" + << rdatait->remote_locators); //MATCHED AND ADDED CORRECTLY: - if (writer->get_listener() != nullptr) + WriterListener* listener = writer->get_listener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = reader_guid; - writer->get_listener()->on_writer_matched(writer, info); + listener->on_writer_matched(writer, info); } } #endif // if HAVE_SECURITY } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && writer->get_listener() != nullptr) + WriterListener* listener = writer->get_listener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - writer->get_listener()->on_offered_incompatible_qos(writer, incompatible_qos); + listener->on_offered_incompatible_qos(writer, incompatible_qos); mp_PDP->notify_incompatible_qos_matching(W->getGuid(), rdatait->guid, incompatible_qos); } @@ -1261,12 +1265,12 @@ bool EDP::pairingWriter( mp_RTPSParticipant->security_manager().remove_reader(writer_guid, participant_guid, reader_guid); #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - if (writer->get_listener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = reader_guid; - writer->get_listener()->on_writer_matched(writer, info); + listener->on_writer_matched(writer, info); } } } @@ -1309,41 +1313,42 @@ bool EDP::pairing_reader_proxy_with_any_local_writer( if (w.matched_reader_add_edp(*rdata)) { EPROSIMA_LOG_INFO(RTPS_EDP_MATCH, - "RP:" << rdata->guid << " match W:" << w.getGuid() << ". RLoc:" << - rdata->remote_locators); + "RP:" << rdata->guid << " match W:" << w.getGuid() << ". RLoc:" + << rdata->remote_locators); //MATCHED AND ADDED CORRECTLY: - if (w.get_listener() != nullptr) + WriterListener* listener = w.get_listener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = reader_guid; - w.get_listener()->on_writer_matched(&w, info); + listener->on_writer_matched(&w, info); } } #endif // if HAVE_SECURITY } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && w.get_listener() != nullptr) + WriterListener* listener = w.get_listener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - w.get_listener()->on_offered_incompatible_qos(&w, incompatible_qos); + listener->on_offered_incompatible_qos(&w, incompatible_qos); mp_PDP->notify_incompatible_qos_matching(w.getGuid(), rdata->guid, incompatible_qos); } - if (w.matched_reader_is_matched(reader_guid) - && w.matched_reader_remove(reader_guid)) + if (w.matched_reader_is_matched(reader_guid) && w.matched_reader_remove(reader_guid)) { #if HAVE_SECURITY mp_RTPSParticipant->security_manager().remove_reader( w.getGuid(), participant_guid, reader_guid); #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - if (w.get_listener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = reader_guid; - w.get_listener()->on_writer_matched(&w, info); + listener->on_writer_matched(&w, info); } } } @@ -1392,25 +1397,24 @@ bool EDP::pairing_reader_proxy_with_local_writer( } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && - w.get_listener() != nullptr) + WriterListener* listener = w.get_listener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - w.get_listener()->on_offered_incompatible_qos(&w, incompatible_qos); + listener->on_offered_incompatible_qos(&w, incompatible_qos); mp_PDP->notify_incompatible_qos_matching(local_writer, rdata.guid, incompatible_qos); } - if (w.matched_reader_is_matched(reader_guid) - && w.matched_reader_remove(reader_guid)) + if (w.matched_reader_is_matched(reader_guid) && w.matched_reader_remove(reader_guid)) { mp_RTPSParticipant->security_manager().remove_reader(w.getGuid(), remote_participant_guid, reader_guid); //MATCHED AND ADDED CORRECTLY: - if (w.get_listener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = reader_guid; - w.get_listener()->on_writer_matched(&w, info); + listener->on_writer_matched(&w, info); } } } @@ -1447,12 +1451,13 @@ bool EDP::pairing_remote_reader_with_local_writer_after_security( matched = true; //MATCHED AND ADDED CORRECTLY: - if (w.get_listener() != nullptr) + WriterListener* listener = w.get_listener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = reader_guid; - w.get_listener()->on_writer_matched(&w, info); + listener->on_writer_matched(&w, info); } } // don't look anymore @@ -1502,41 +1507,42 @@ bool EDP::pairing_writer_proxy_with_any_local_reader( if (r.matched_writer_add_edp(*wdata)) { EPROSIMA_LOG_INFO(RTPS_EDP_MATCH, - "WP:" << wdata->guid << " match R:" << r.getGuid() << ". WLoc:" << - wdata->remote_locators); + "WP:" << wdata->guid << " match R:" << r.getGuid() << ". WLoc:" + << wdata->remote_locators); //MATCHED AND ADDED CORRECTLY: - if (r.get_listener() != nullptr) + ReaderListener* listener = r.get_listener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = writer_guid; - r.get_listener()->on_reader_matched(&r, info); + listener->on_reader_matched(&r, info); } } #endif // if HAVE_SECURITY } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && r.get_listener() != nullptr) + ReaderListener* listener = r.get_listener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - r.get_listener()->on_requested_incompatible_qos(&r, incompatible_qos); + listener->on_requested_incompatible_qos(&r, incompatible_qos); mp_PDP->notify_incompatible_qos_matching(r.getGuid(), wdata->guid, incompatible_qos); } - if (r.matched_writer_is_matched(writer_guid) - && r.matched_writer_remove(writer_guid)) + if (r.matched_writer_is_matched(writer_guid) && r.matched_writer_remove(writer_guid)) { #if HAVE_SECURITY mp_RTPSParticipant->security_manager().remove_writer(readerGUID, participant_guid, writer_guid); #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - if (r.get_listener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = writer_guid; - r.get_listener()->on_reader_matched(&r, info); + listener->on_reader_matched(&r, info); } } } @@ -1585,25 +1591,24 @@ bool EDP::pairing_writer_proxy_with_local_reader( } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && - r.get_listener() != nullptr) + ReaderListener* listener = r.get_listener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - r.get_listener()->on_requested_incompatible_qos(&r, incompatible_qos); + listener->on_requested_incompatible_qos(&r, incompatible_qos); mp_PDP->notify_incompatible_qos_matching(local_reader, wdata.guid, incompatible_qos); } - if (r.matched_writer_is_matched(writer_guid) - && r.matched_writer_remove(writer_guid)) + if (r.matched_writer_is_matched(writer_guid) && r.matched_writer_remove(writer_guid)) { mp_RTPSParticipant->security_manager().remove_writer(readerGUID, remote_participant_guid, writer_guid); //MATCHED AND ADDED CORRECTLY: - if (r.get_listener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = writer_guid; - r.get_listener()->on_reader_matched(&r, info); + listener->on_reader_matched(&r, info); } } } @@ -1642,13 +1647,13 @@ bool EDP::pairing_remote_writer_with_local_reader_after_security( matched = true; //MATCHED AND ADDED CORRECTLY: - if (r.get_listener() != nullptr) + ReaderListener* listener = r.get_listener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = writer_guid; - r.get_listener()->on_reader_matched(&r, info); - + listener->on_reader_matched(&r, info); } } // dont' look anymore diff --git a/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp b/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp index 9c1bd7ae1bf..d0fd39a68ce 100644 --- a/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp +++ b/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp @@ -548,6 +548,125 @@ TEST(DDSDiscovery, UpdateMatchedStatus) datawriter_2.destroy(); } +/* + * This is a regression test for redmine issue 21355. + * + * In order to check that the on_publication_matched() and on_subscription_matched() callbacks are always called with + * either +1 or -1 as the current_count_change, this test creates and destroys multiple DataReaders and DataWriters in + * different threads, with a listener that counts the number of times on_publication_matched() and + * on_subscription_matched() are called with current_count_change different than 1 or -1. + * + * The test fails if any of the calls to on_publication_matched() or on_subscription_matched() is made with a + * current_count_change different than 1 or -1. + */ +TEST(DDSDiscovery, MatchedCallbackListenerMultithread) +{ + constexpr size_t NUM_THREADS = 25; + + struct TestListener + : public eprosima::fastdds::dds::DataReaderListener + , public eprosima::fastdds::dds::DataWriterListener + { + std::atomic reader_other_calls{ 0 }; + std::atomic writer_other_calls{ 0 }; + + void on_publication_matched( + eprosima::fastdds::dds::DataWriter*, + const eprosima::fastdds::dds::PublicationMatchedStatus& info) override + { + if ((info.current_count_change != 1) && (info.current_count_change != -1)) + { + ++writer_other_calls; + } + } + + void on_subscription_matched( + eprosima::fastdds::dds::DataReader*, + const eprosima::fastdds::dds::SubscriptionMatchedStatus& info) override + { + if ((info.current_count_change != 1) && (info.current_count_change != -1)) + { + ++reader_other_calls; + } + } + + }; + + std::ostringstream t; + t << TEST_TOPIC_NAME << "_" << asio::ip::host_name() << "_" << GET_PID(); + std::string topic_name = t.str(); + TestListener listener; + eprosima::fastdds::dds::TypeSupport type(new HelloWorldPubSubType()); + + auto create_entities = [&listener, &topic_name, &type]() + { + using namespace eprosima::fastdds::dds; + + uint32_t domain_id = static_cast(GET_PID()) % 100; + auto factory = DomainParticipantFactory::get_shared_instance(); + DomainParticipantQos participant_qos; + factory->get_default_participant_qos(participant_qos); + participant_qos.setup_transports(eprosima::fastdds::rtps::BuiltinTransports::UDPv4); + participant_qos.wire_protocol().builtin.discovery_config.leaseDuration.seconds = 0; + participant_qos.wire_protocol().builtin.discovery_config.leaseDuration.nanosec = 100000000; + participant_qos.wire_protocol().builtin.discovery_config.leaseDuration_announcementperiod.seconds = 0; + participant_qos.wire_protocol().builtin.discovery_config.leaseDuration_announcementperiod.nanosec = + 50000000; + + auto participant = factory->create_participant(domain_id, participant_qos); + ASSERT_NE(participant, nullptr); + + type.register_type(participant); + auto topic = participant->create_topic(topic_name, type.get_type_name(), TOPIC_QOS_DEFAULT); + ASSERT_NE(topic, nullptr); + + auto publisher = participant->create_publisher(PUBLISHER_QOS_DEFAULT); + ASSERT_NE(publisher, nullptr); + + DataWriterQos datawriter_qos; + publisher->get_default_datawriter_qos(datawriter_qos); + datawriter_qos.data_sharing().off(); + auto datawriter = publisher->create_datawriter(topic, datawriter_qos, &listener); + ASSERT_NE(datawriter, nullptr); + + auto subscriber = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT); + ASSERT_NE(subscriber, nullptr); + + DataReaderQos datareader_qos; + subscriber->get_default_datareader_qos(datareader_qos); + datareader_qos.data_sharing().off(); + auto datareader = subscriber->create_datareader(topic, datareader_qos, &listener); + ASSERT_NE(datareader, nullptr); + + HelloWorld msg; + msg.index(1); + datawriter->write(&msg); + + SampleInfo info; + datareader->take_next_sample(&msg, &info); + + participant->delete_contained_entities(); + factory->delete_participant(participant); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + }; + + std::vector threads; + threads.reserve(NUM_THREADS); + for (size_t i = 0; i < NUM_THREADS; ++i) + { + threads.emplace_back(create_entities); + } + + for (auto& thread : threads) + { + thread.join(); + } + + EXPECT_EQ(listener.reader_other_calls.load(), 0u); + EXPECT_EQ(listener.writer_other_calls.load(), 0u); +} + /** * This test checks that the physical properties are correctly sent on the DATA[p], and that the * ParticipantProxyData on the receiver side has the correct values.