diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.cpp b/src/cpp/fastdds/publisher/DataWriterImpl.cpp index 579e8d799b0..00d1699411c 100644 --- a/src/cpp/fastdds/publisher/DataWriterImpl.cpp +++ b/src/cpp/fastdds/publisher/DataWriterImpl.cpp @@ -1286,6 +1286,8 @@ void DataWriterImpl::InnerDataWriterListener::onWriterMatched( RTPSWriter* /*writer*/, const PublicationMatchedStatus& 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 568aa771094..fbfb4483372 100644 --- a/src/cpp/fastdds/publisher/DataWriterImpl.hpp +++ b/src/cpp/fastdds/publisher/DataWriterImpl.hpp @@ -459,11 +459,11 @@ class DataWriterImpl : protected rtps::IReaderDataFilter const uint32_t& status_id); #endif //FASTDDS_STATISTICS - DataWriterImpl* data_writer_; - private: using fastrtps::rtps::WriterListener::onWriterMatched; + std::mutex matching_info_mutex_; + DataWriterImpl* data_writer_; } writer_listener_; diff --git a/src/cpp/fastdds/subscriber/DataReaderImpl.cpp b/src/cpp/fastdds/subscriber/DataReaderImpl.cpp index a60a48bd09e..b421672aaff 100644 --- a/src/cpp/fastdds/subscriber/DataReaderImpl.cpp +++ b/src/cpp/fastdds/subscriber/DataReaderImpl.cpp @@ -941,6 +941,8 @@ void DataReaderImpl::InnerDataReaderListener::onReaderMatched( RTPSReader* /*reader*/, const SubscriptionMatchedStatus& info) { + std::lock_guard scoped_lock(matching_info_mutex_); + data_reader_->update_subscription_matched_status(info); StatusMask notify_status = StatusMask::subscription_matched(); diff --git a/src/cpp/fastdds/subscriber/DataReaderImpl.hpp b/src/cpp/fastdds/subscriber/DataReaderImpl.hpp index 49be7c0cd58..933b6b71edb 100644 --- a/src/cpp/fastdds/subscriber/DataReaderImpl.hpp +++ b/src/cpp/fastdds/subscriber/DataReaderImpl.hpp @@ -448,7 +448,10 @@ class DataReaderImpl const uint32_t& status_id); #endif //FASTDDS_STATISTICS + private: + DataReaderImpl* data_reader_; + std::mutex matching_info_mutex_; private: diff --git a/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp b/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp index fb6c4ed2747..8e30feb5645 100644 --- a/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp +++ b/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp @@ -446,7 +446,8 @@ bool EDP::updatedLocalReader( if (!rdata->type_information().assigned()) { const types::TypeInformation* type_info = - types::TypeObjectFactory::get_instance()->get_type_information(rdata->typeName().c_str()); + types::TypeObjectFactory::get_instance()->get_type_information( + rdata->typeName().c_str()); if (type_info != nullptr) { rdata->type_information() = *type_info; @@ -538,7 +539,8 @@ bool EDP::updatedLocalWriter( if (!wdata->type_information().assigned()) { const types::TypeInformation* type_info = - types::TypeObjectFactory::get_instance()->get_type_information(wdata->typeName().c_str()); + types::TypeObjectFactory::get_instance()->get_type_information( + wdata->typeName().c_str()); if (type_info != nullptr) { wdata->type_information() = *type_info; @@ -619,8 +621,8 @@ bool EDP::unpairWriterProxy( #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - ReaderListener* listener = nullptr; - if (nullptr != (listener = r.getListener())) + ReaderListener* listener = r.getListener(); + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; @@ -658,8 +660,8 @@ bool EDP::unpairReaderProxy( participant_guid, reader_guid); #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - WriterListener* listener = nullptr; - if (nullptr != (listener = w.getListener())) + WriterListener* listener = w.getListener(); + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; @@ -713,9 +715,11 @@ bool EDP::valid_matching( if (wdata->topicKind() != rdata->topicKind()) { - EPROSIMA_LOG_WARNING(RTPS_EDP, "INCOMPATIBLE QOS:Remote Reader " << rdata->guid() << " is publishing in topic " - << rdata->topicName() << "(keyed:" << rdata->topicKind() << - "), local writer publishes as keyed: " << wdata->topicKind()); + EPROSIMA_LOG_WARNING(RTPS_EDP, + "INCOMPATIBLE QOS:Remote Reader " << rdata->guid() << " is publishing in topic " + << rdata->topicName() << "(keyed:" + << rdata->topicKind() << "), local writer publishes as keyed: " + << wdata->topicKind()); reason.set(MatchingFailureMask::inconsistent_topic); return false; @@ -732,18 +736,18 @@ bool EDP::valid_matching( && rdata->m_qos.m_reliability.kind == RELIABLE_RELIABILITY_QOS) //Means our writer is BE but the reader wants RE { - EPROSIMA_LOG_WARNING(RTPS_EDP, "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "):Remote Reader " - << rdata->guid() << - " is Reliable and local writer is BE "); + EPROSIMA_LOG_WARNING(RTPS_EDP, + "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "):Remote Reader " << rdata->guid() + << " is Reliable and local writer is BE "); incompatible_qos.set(fastdds::dds::RELIABILITY_QOS_POLICY_ID); } if (wdata->m_qos.m_durability.kind < rdata->m_qos.m_durability.kind) { // TODO (MCC) Change log message - EPROSIMA_LOG_WARNING(RTPS_EDP, "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "):RemoteReader " - << rdata->guid() << - " has TRANSIENT_LOCAL DURABILITY and we offer VOLATILE"); + EPROSIMA_LOG_WARNING(RTPS_EDP, + "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "):RemoteReader " << rdata->guid() + << " has TRANSIENT_LOCAL DURABILITY and we offer VOLATILE"); incompatible_qos.set(fastdds::dds::DURABILITY_QOS_POLICY_ID); } @@ -994,28 +998,30 @@ bool EDP::pairingReader( if (R->matched_writer_add(*wdatait)) { EPROSIMA_LOG_INFO(RTPS_EDP_MATCH, - "WP:" << wdatait->guid() << " match R:" << R->getGuid() << ". RLoc:" << - wdatait->remote_locators()); + "WP:" << wdatait->guid() << " match R:" << R->getGuid() << ". RLoc:" + << wdatait->remote_locators()); //MATCHED AND ADDED CORRECTLY: - if (R->getListener() != nullptr) + ReaderListener* listener = R->getListener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = writer_guid; - R->getListener()->onReaderMatched(R, info); + listener->onReaderMatched(R, info); const SubscriptionMatchedStatus& sub_info = update_subscription_matched_status(reader_guid, writer_guid, 1); - R->getListener()->onReaderMatched(R, sub_info); + listener->onReaderMatched(R, sub_info); } } #endif // if HAVE_SECURITY } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && R->getListener() != nullptr) + ReaderListener* listener = R->getListener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && nullptr != listener) { - R->getListener()->on_requested_incompatible_qos(R, incompatible_qos); + listener->on_requested_incompatible_qos(R, incompatible_qos); } //EPROSIMA_LOG_INFO(RTPS_EDP,RTPS_CYAN<<"Valid Matching to writerProxy: "<m_guid<getListener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = writer_guid; - R->getListener()->onReaderMatched(R, info); + listener->onReaderMatched(R, info); const SubscriptionMatchedStatus& sub_info = update_subscription_matched_status(reader_guid, writer_guid, -1); - R->getListener()->onReaderMatched(R, sub_info); + listener->onReaderMatched(R, sub_info); } } } @@ -1090,29 +1096,31 @@ bool EDP::pairingWriter( if (W->matched_reader_add(*rdatait)) { EPROSIMA_LOG_INFO(RTPS_EDP_MATCH, - "RP:" << rdatait->guid() << " match W:" << W->getGuid() << ". WLoc:" << - rdatait->remote_locators()); + "RP:" << rdatait->guid() << " match W:" << W->getGuid() << ". WLoc:" + << rdatait->remote_locators()); //MATCHED AND ADDED CORRECTLY: - if (W->getListener() != nullptr) + WriterListener* listener = W->getListener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = reader_guid; - W->getListener()->onWriterMatched(W, info); + listener->onWriterMatched(W, info); const GUID_t& writer_guid = W->getGuid(); const PublicationMatchedStatus& pub_info = update_publication_matched_status(reader_guid, writer_guid, 1); - W->getListener()->onWriterMatched(W, pub_info); + listener->onWriterMatched(W, pub_info); } } #endif // if HAVE_SECURITY } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && W->getListener() != nullptr) + WriterListener* listener = W->getListener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - W->getListener()->on_offered_incompatible_qos(W, incompatible_qos); + listener->on_offered_incompatible_qos(W, incompatible_qos); } //EPROSIMA_LOG_INFO(RTPS_EDP,RTPS_CYAN<<"Valid Matching to writerProxy: "<m_guid<security_manager().remove_reader(W->getGuid(), participant_guid, reader_guid); #endif // if HAVE_SECURITY //MATCHED AND ADDED CORRECTLY: - if (W->getListener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = reader_guid; - W->getListener()->onWriterMatched(W, info); + listener->onWriterMatched(W, info); const GUID_t& writer_guid = W->getGuid(); const PublicationMatchedStatus& pub_info = update_publication_matched_status(reader_guid, writer_guid, -1); - W->getListener()->onWriterMatched(W, pub_info); + listener->onWriterMatched(W, pub_info); } } @@ -1176,48 +1184,49 @@ bool EDP::pairing_reader_proxy_with_any_local_writer( if (w.matched_reader_add(*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.getListener() != nullptr) + WriterListener* listener = w.getListener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = reader_guid; - w.getListener()->onWriterMatched(&w, info); + listener->onWriterMatched(&w, info); const PublicationMatchedStatus& pub_info = update_publication_matched_status(reader_guid, writerGUID, 1); - w.getListener()->onWriterMatched(&w, pub_info); + listener->onWriterMatched(&w, pub_info); } } #endif // if HAVE_SECURITY } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && w.getListener() != nullptr) + WriterListener* listener = w.getListener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - w.getListener()->on_offered_incompatible_qos(&w, incompatible_qos); + listener->on_offered_incompatible_qos(&w, 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.getListener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = reader_guid; - w.getListener()->onWriterMatched(&w, info); + listener->onWriterMatched(&w, info); const PublicationMatchedStatus& pub_info = update_publication_matched_status(reader_guid, writerGUID, -1); - w.getListener()->onWriterMatched(&w, pub_info); + listener->onWriterMatched(&w, pub_info); } } } @@ -1266,28 +1275,27 @@ bool EDP::pairing_reader_proxy_with_local_writer( } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && - w.getListener() != nullptr) + WriterListener* listener = w.getListener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - w.getListener()->on_offered_incompatible_qos(&w, incompatible_qos); + listener->on_offered_incompatible_qos(&w, 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.getListener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = reader_guid; - w.getListener()->onWriterMatched(&w, info); + listener->onWriterMatched(&w, info); const PublicationMatchedStatus& pub_info = update_publication_matched_status(reader_guid, writerGUID, -1); - w.getListener()->onWriterMatched(&w, pub_info); + listener->onWriterMatched(&w, pub_info); } } } @@ -1324,16 +1332,17 @@ bool EDP::pairing_remote_reader_with_local_writer_after_security( matched = true; //MATCHED AND ADDED CORRECTLY: - if (w.getListener() != nullptr) + WriterListener* listener = w.getListener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = reader_guid; - w.getListener()->onWriterMatched(&w, info); + listener->onWriterMatched(&w, info); const PublicationMatchedStatus& pub_info = update_publication_matched_status(reader_guid, writerGUID, 1); - w.getListener()->onWriterMatched(&w, pub_info); + listener->onWriterMatched(&w, pub_info); } } // don't look anymore @@ -1383,48 +1392,49 @@ bool EDP::pairing_writer_proxy_with_any_local_reader( if (r.matched_writer_add(*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.getListener() != nullptr) + ReaderListener* listener = r.getListener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = writer_guid; - r.getListener()->onReaderMatched(&r, info); + listener->onReaderMatched(&r, info); const SubscriptionMatchedStatus& sub_info = update_subscription_matched_status(readerGUID, writer_guid, 1); - r.getListener()->onReaderMatched(&r, sub_info); + listener->onReaderMatched(&r, sub_info); } } #endif // if HAVE_SECURITY } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && r.getListener() != nullptr) + ReaderListener* listener = r.getListener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - r.getListener()->on_requested_incompatible_qos(&r, incompatible_qos); + listener->on_requested_incompatible_qos(&r, 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.getListener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = writer_guid; - r.getListener()->onReaderMatched(&r, info); + listener->onReaderMatched(&r, info); const SubscriptionMatchedStatus& sub_info = update_subscription_matched_status(readerGUID, writer_guid, -1); - r.getListener()->onReaderMatched(&r, sub_info); + listener->onReaderMatched(&r, sub_info); } } } @@ -1473,28 +1483,27 @@ bool EDP::pairing_writer_proxy_with_local_reader( } else { - if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && - r.getListener() != nullptr) + ReaderListener* listener = r.getListener(); + if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener)) { - r.getListener()->on_requested_incompatible_qos(&r, incompatible_qos); + listener->on_requested_incompatible_qos(&r, 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.getListener() != nullptr) + if (nullptr != listener) { MatchingInfo info; info.status = REMOVED_MATCHING; info.remoteEndpointGuid = writer_guid; - r.getListener()->onReaderMatched(&r, info); + listener->onReaderMatched(&r, info); const SubscriptionMatchedStatus& sub_info = update_subscription_matched_status(readerGUID, writer_guid, -1); - r.getListener()->onReaderMatched(&r, sub_info); + listener->onReaderMatched(&r, sub_info); } } } @@ -1534,16 +1543,17 @@ bool EDP::pairing_remote_writer_with_local_reader_after_security( matched = true; //MATCHED AND ADDED CORRECTLY: - if (r.getListener() != nullptr) + ReaderListener* listener = r.getListener(); + if (nullptr != listener) { MatchingInfo info; info.status = MATCHED_MATCHING; info.remoteEndpointGuid = writer_guid; - r.getListener()->onReaderMatched(&r, info); + listener->onReaderMatched(&r, info); const SubscriptionMatchedStatus& sub_info = update_subscription_matched_status(readerGUID, writer_guid, 1); - r.getListener()->onReaderMatched(&r, sub_info); + listener->onReaderMatched(&r, sub_info); } } diff --git a/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp b/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp index 502fbebfae0..7249b322b47 100644 --- a/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp +++ b/test/blackbox/common/DDSBlackboxTestsDiscovery.cpp @@ -604,6 +604,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.