diff --git a/src/cpp/fastdds/publisher/DataWriter.cpp b/src/cpp/fastdds/publisher/DataWriter.cpp index 92e10412200..e0330ada38d 100644 --- a/src/cpp/fastdds/publisher/DataWriter.cpp +++ b/src/cpp/fastdds/publisher/DataWriter.cpp @@ -225,8 +225,7 @@ const DataWriterQos& DataWriter::get_qos() const ReturnCode_t DataWriter::get_qos( DataWriterQos& qos) const { - qos = impl_->get_qos(); - return ReturnCode_t::RETCODE_OK; + return impl_->get_qos(qos); } ReturnCode_t DataWriter::set_listener( diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.cpp b/src/cpp/fastdds/publisher/DataWriterImpl.cpp index 00d1699411c..8e0e556b5fb 100644 --- a/src/cpp/fastdds/publisher/DataWriterImpl.cpp +++ b/src/cpp/fastdds/publisher/DataWriterImpl.cpp @@ -237,124 +237,143 @@ ReturnCode_t DataWriterImpl::enable() { assert(writer_ == nullptr); - pool_config_ = PoolConfig::from_history_attributes(history_.m_att); - // When the user requested PREALLOCATED_WITH_REALLOC, but we know the type cannot - // grow, we translate the policy into bare PREALLOCATED - if (PREALLOCATED_WITH_REALLOC_MEMORY_MODE == pool_config_.memory_policy && - (type_->is_bounded() || type_->is_plain(data_representation_))) - { - pool_config_.memory_policy = PREALLOCATED_MEMORY_MODE; - } - WriterAttributes w_att; - w_att.throughputController = qos_.throughput_controller(); - w_att.endpoint.durabilityKind = qos_.durability().durabilityKind(); - w_att.endpoint.endpointKind = WRITER; - w_att.endpoint.reliabilityKind = qos_.reliability().kind == RELIABLE_RELIABILITY_QOS ? RELIABLE : BEST_EFFORT; - w_att.endpoint.topicKind = type_->m_isGetKeyDefined ? WITH_KEY : NO_KEY; - w_att.endpoint.multicastLocatorList = qos_.endpoint().multicast_locator_list; - w_att.endpoint.unicastLocatorList = qos_.endpoint().unicast_locator_list; - w_att.endpoint.remoteLocatorList = qos_.endpoint().remote_locator_list; - w_att.endpoint.external_unicast_locators = qos_.endpoint().external_unicast_locators; - w_att.endpoint.ignore_non_matching_locators = qos_.endpoint().ignore_non_matching_locators; - w_att.mode = qos_.publish_mode().kind == SYNCHRONOUS_PUBLISH_MODE ? SYNCHRONOUS_WRITER : ASYNCHRONOUS_WRITER; - w_att.flow_controller_name = qos_.publish_mode().flow_controller_name; - w_att.endpoint.properties = qos_.properties(); - w_att.endpoint.ownershipKind = qos_.ownership().kind; - w_att.endpoint.setEntityID(qos_.endpoint().entity_id); - w_att.endpoint.setUserDefinedID(qos_.endpoint().user_defined_id); - w_att.times = qos_.reliable_writer_qos().times; - w_att.liveliness_kind = qos_.liveliness().kind; - w_att.liveliness_lease_duration = qos_.liveliness().lease_duration; - w_att.liveliness_announcement_period = qos_.liveliness().announcement_period; - w_att.matched_readers_allocation = qos_.writer_resource_limits().matched_subscriber_allocation; - w_att.disable_heartbeat_piggyback = qos_.reliable_writer_qos().disable_heartbeat_piggyback; - - // TODO(Ricardo) Remove in future - // Insert topic_name and partitions - Property property; - property.name("topic_name"); - property.value(topic_->get_name().c_str()); - w_att.endpoint.properties.properties().push_back(std::move(property)); - - std::string* endpoint_partitions = PropertyPolicyHelper::find_property(qos_.properties(), "partitions"); - - if (endpoint_partitions) - { - property.name("partitions"); - property.value(*endpoint_partitions); - w_att.endpoint.properties.properties().push_back(std::move(property)); - } - else if (publisher_->get_qos().partition().names().size() > 0) + bool filtering_enabled = false; + fastrtps::ResourceLimitedContainerConfig reader_filters_alloc{}; + fastrtps::Duration_t lifespan_duration{}; + std::string* endpoint_partitions; + { - property.name("partitions"); - std::string partitions; - bool is_first_partition = true; - for (auto partition : publisher_->get_qos().partition().names()) + std::lock_guard qos_guard(qos_mutex_); + + pool_config_ = PoolConfig::from_history_attributes(history_.m_att); + + // When the user requested PREALLOCATED_WITH_REALLOC, but we know the type cannot + // grow, we translate the policy into bare PREALLOCATED + if (PREALLOCATED_WITH_REALLOC_MEMORY_MODE == pool_config_.memory_policy && + (type_->is_bounded() || type_->is_plain(data_representation_))) { - partitions += (is_first_partition ? "" : ";") + partition; - is_first_partition = false; + pool_config_.memory_policy = PREALLOCATED_MEMORY_MODE; } - property.value(std::move(partitions)); + + w_att.throughputController = qos_.throughput_controller(); + w_att.endpoint.durabilityKind = qos_.durability().durabilityKind(); + w_att.endpoint.endpointKind = WRITER; + w_att.endpoint.reliabilityKind = qos_.reliability().kind == RELIABLE_RELIABILITY_QOS ? RELIABLE : BEST_EFFORT; + w_att.endpoint.topicKind = type_->m_isGetKeyDefined ? WITH_KEY : NO_KEY; + w_att.endpoint.multicastLocatorList = qos_.endpoint().multicast_locator_list; + w_att.endpoint.unicastLocatorList = qos_.endpoint().unicast_locator_list; + w_att.endpoint.remoteLocatorList = qos_.endpoint().remote_locator_list; + w_att.endpoint.external_unicast_locators = qos_.endpoint().external_unicast_locators; + w_att.endpoint.ignore_non_matching_locators = qos_.endpoint().ignore_non_matching_locators; + w_att.mode = qos_.publish_mode().kind == SYNCHRONOUS_PUBLISH_MODE ? SYNCHRONOUS_WRITER : ASYNCHRONOUS_WRITER; + w_att.flow_controller_name = qos_.publish_mode().flow_controller_name; + w_att.endpoint.properties = qos_.properties(); + w_att.endpoint.ownershipKind = qos_.ownership().kind; + w_att.endpoint.setEntityID(qos_.endpoint().entity_id); + w_att.endpoint.setUserDefinedID(qos_.endpoint().user_defined_id); + w_att.times = qos_.reliable_writer_qos().times; + w_att.liveliness_kind = qos_.liveliness().kind; + w_att.liveliness_lease_duration = qos_.liveliness().lease_duration; + w_att.liveliness_announcement_period = qos_.liveliness().announcement_period; + w_att.matched_readers_allocation = qos_.writer_resource_limits().matched_subscriber_allocation; + w_att.disable_heartbeat_piggyback = qos_.reliable_writer_qos().disable_heartbeat_piggyback; + + // TODO(Ricardo) Remove in future + // Insert topic_name and partitions + Property property; + property.name("topic_name"); + property.value(topic_->get_name().c_str()); w_att.endpoint.properties.properties().push_back(std::move(property)); - } - if (qos_.reliable_writer_qos().disable_positive_acks.enabled && - qos_.reliable_writer_qos().disable_positive_acks.duration != c_TimeInfinite) - { - w_att.disable_positive_acks = true; - w_att.keep_duration = qos_.reliable_writer_qos().disable_positive_acks.duration; - } + endpoint_partitions = PropertyPolicyHelper::find_property(qos_.properties(), "partitions"); - ReturnCode_t ret_code = check_datasharing_compatible(w_att, is_data_sharing_compatible_); - if (ret_code != ReturnCode_t::RETCODE_OK) - { - return ret_code; - } + if (endpoint_partitions) + { + property.name("partitions"); + property.value(*endpoint_partitions); + w_att.endpoint.properties.properties().push_back(std::move(property)); + } + else if (publisher_->get_qos().partition().names().size() > 0) + { + property.name("partitions"); + std::string partitions; + bool is_first_partition = true; + for (auto partition : publisher_->get_qos().partition().names()) + { + partitions += (is_first_partition ? "" : ";") + partition; + is_first_partition = false; + } + property.value(std::move(partitions)); + w_att.endpoint.properties.properties().push_back(std::move(property)); + } - if (is_data_sharing_compatible_) - { - DataSharingQosPolicy datasharing(qos_.data_sharing()); - if (datasharing.domain_ids().empty()) + if (qos_.reliable_writer_qos().disable_positive_acks.enabled && + qos_.reliable_writer_qos().disable_positive_acks.duration != c_TimeInfinite) { - datasharing.add_domain_id(utils::default_domain_id()); + w_att.disable_positive_acks = true; + w_att.keep_duration = qos_.reliable_writer_qos().disable_positive_acks.duration; + } + + ReturnCode_t ret_code = check_datasharing_compatible(w_att, is_data_sharing_compatible_); + if (ret_code != ReturnCode_t::RETCODE_OK) + { + return ret_code; } - w_att.endpoint.set_data_sharing_configuration(datasharing); - // Update pool config for KEEP_ALL when max_samples is infinite - if ((0 >= pool_config_.maximum_size) && (KEEP_ALL_HISTORY_QOS == qos_.history().kind)) + if (is_data_sharing_compatible_) { - // Override infinite with old default value for max_samples + extra samples - pool_config_.maximum_size = 5000; - if (0 < qos_.resource_limits().extra_samples) + DataSharingQosPolicy datasharing(qos_.data_sharing()); + if (datasharing.domain_ids().empty()) { - pool_config_.maximum_size += static_cast(qos_.resource_limits().extra_samples); + datasharing.add_domain_id(utils::default_domain_id()); + } + w_att.endpoint.set_data_sharing_configuration(datasharing); + + // Update pool config for KEEP_ALL when max_samples is infinite + if ((0 == pool_config_.maximum_size) && (KEEP_ALL_HISTORY_QOS == qos_.history().kind)) + { + // Override infinite with old default value for max_samples + extra samples + pool_config_.maximum_size = 5000; + if (0 < qos_.resource_limits().extra_samples) + { + pool_config_.maximum_size += static_cast(qos_.resource_limits().extra_samples); + } + EPROSIMA_LOG_ERROR(DATA_WRITER, + "DataWriter with KEEP_ALL history and infinite max_samples is not compatible with DataSharing. " + "Setting max_samples to " << pool_config_.maximum_size); } - EPROSIMA_LOG_ERROR(DATA_WRITER, - "DataWriter with KEEP_ALL history and infinite max_samples is not compatible with DataSharing. " - "Setting max_samples to " << pool_config_.maximum_size); } - } - else - { - DataSharingQosPolicy datasharing; - datasharing.off(); - w_att.endpoint.set_data_sharing_configuration(datasharing); + else + { + DataSharingQosPolicy datasharing; + datasharing.off(); + w_att.endpoint.set_data_sharing_configuration(datasharing); + } + + filtering_enabled = + qos_.liveliness().lease_duration.is_infinite() && + (0 < qos_.writer_resource_limits().reader_filters_allocation.maximum); + + if (filtering_enabled) + { + reader_filters_alloc = qos_.writer_resource_limits().reader_filters_allocation; + } + + // Set Datawriter's DataRepresentationId taking into account the QoS. + data_representation_ = qos_.representation().m_value.empty() + || XCDR_DATA_REPRESENTATION == qos_.representation().m_value.at(0) + ? XCDR_DATA_REPRESENTATION : XCDR2_DATA_REPRESENTATION; + + lifespan_duration = qos_.lifespan().duration; } - bool filtering_enabled = - qos_.liveliness().lease_duration.is_infinite() && - (0 < qos_.writer_resource_limits().reader_filters_allocation.maximum); if (filtering_enabled) { - reader_filters_.reset(new ReaderFilterCollection(qos_.writer_resource_limits().reader_filters_allocation)); + std::lock_guard lock(filters_mtx_); + reader_filters_.reset(new ReaderFilterCollection(reader_filters_alloc)); } - // Set Datawriter's DataRepresentationId taking into account the QoS. - data_representation_ = qos_.representation().m_value.empty() - || XCDR_DATA_REPRESENTATION == qos_.representation().m_value.at(0) - ? XCDR_DATA_REPRESENTATION : XCDR2_DATA_REPRESENTATION; - auto change_pool = get_change_pool(); if (!change_pool) { @@ -411,7 +430,11 @@ ReturnCode_t DataWriterImpl::enable() return ReturnCode_t::RETCODE_ERROR; } - writer_ = writer; + { + std::lock_guard writer_assign_lock(qos_mutex_); + writer_ = writer; + } + if (filtering_enabled) { writer_->reader_data_filter(this); @@ -427,10 +450,10 @@ ReturnCode_t DataWriterImpl::enable() { return lifespan_expired(); }, - qos_.lifespan().duration.to_ns() * 1e-6); + lifespan_duration.to_ns() * 1e-6); // In case it has been loaded from the persistence DB, expire old samples. - if (qos_.lifespan().duration != c_TimeInfinite) + if (lifespan_duration != c_TimeInfinite) { if (lifespan_expired()) { @@ -439,23 +462,29 @@ ReturnCode_t DataWriterImpl::enable() } // REGISTER THE WRITER - WriterQos wqos = qos_.get_writerqos(get_publisher()->get_qos(), topic_->get_qos()); - if (!is_data_sharing_compatible_) - { - wqos.data_sharing.off(); - } - if (endpoint_partitions) + WriterQos wqos; + DataWriterQos qos_snapshot; { - std::istringstream partition_string(*endpoint_partitions); - std::string partition_name; - wqos.m_partition.clear(); - - while (std::getline(partition_string, partition_name, ';')) + std::lock_guard qos_guard(qos_mutex_); + wqos = qos_.get_writerqos(get_publisher()->get_qos(), topic_->get_qos()); + if (!is_data_sharing_compatible_) { - wqos.m_partition.push_back(partition_name.c_str()); + wqos.data_sharing.off(); } + if (endpoint_partitions) + { + std::istringstream partition_string(*endpoint_partitions); + std::string partition_name; + wqos.m_partition.clear(); + + while (std::getline(partition_string, partition_name, ';')) + { + wqos.m_partition.push_back(partition_name.c_str()); + } + } + qos_snapshot = qos_; } - publisher_->rtps_participant()->registerWriter(writer_, get_topic_attributes(qos_, *topic_, type_), wqos); + publisher_->rtps_participant()->registerWriter(writer_, get_topic_attributes(qos_snapshot, *topic_, type_), wqos); return ReturnCode_t::RETCODE_OK; } @@ -1169,18 +1198,31 @@ InstanceHandle_t DataWriterImpl::get_instance_handle() const void DataWriterImpl::publisher_qos_updated() { - if (writer_ != nullptr) + fastrtps::rtps::RTPSWriter* writer_snapshot = nullptr; + WriterQos wqos; + + DataWriterQos qos_snapshot; { - // NOTIFY THE BUILTIN PROTOCOLS THAT THE WRITER HAS CHANGED - WriterQos wqos = qos_.get_writerqos(get_publisher()->get_qos(), topic_->get_qos()); - publisher_->rtps_participant()->updateWriter(writer_, get_topic_attributes(qos_, *topic_, type_), wqos); + std::lock_guard qos_guard(qos_mutex_); + + if (writer_ == nullptr) + { + return; + } + + writer_snapshot = writer_; + wqos = qos_.get_writerqos(get_publisher()->get_qos(), topic_->get_qos()); + qos_snapshot = qos_; } + + // NOTIFY THE BUILTIN PROTOCOLS THAT THE WRITER HAS CHANGED + publisher_->rtps_participant()->updateWriter(writer_snapshot, get_topic_attributes(qos_snapshot, *topic_, type_), + wqos); } ReturnCode_t DataWriterImpl::set_qos( const DataWriterQos& qos) { - bool enabled = writer_ != nullptr; const DataWriterQos& qos_to_set = (&qos == &DATAWRITER_QOS_DEFAULT) ? publisher_->get_default_datawriter_qos() : qos; @@ -1201,53 +1243,73 @@ ReturnCode_t DataWriterImpl::set_qos( } } - if (enabled && !can_qos_be_updated(qos_, qos_to_set)) + DataWriterQos old_qos; + DataWriterQos new_qos; { - return ReturnCode_t::RETCODE_IMMUTABLE_POLICY; + std::lock_guard qos_guard(qos_mutex_); + bool enabled = writer_ != nullptr; + + if (!enabled) + { + set_qos(qos_, qos_to_set, true); + return ReturnCode_t::RETCODE_OK; + } + + if (!can_qos_be_updated(qos_, qos_to_set)) + { + return ReturnCode_t::RETCODE_IMMUTABLE_POLICY; + } + + old_qos = qos_; + set_qos(qos_, qos_to_set, false); + new_qos = qos_; } - // Take a snapshot of the current QoS before mutating it - const DataWriterQos old_qos = qos_; + WriterQos wqos; + bool update_attributes = false; + WriterAttributes w_att; + bool deadline_changed = false; + bool lifespan_changed = false; + + if (new_qos.reliability().kind == eprosima::fastrtps::RELIABLE_RELIABILITY_QOS && + new_qos.reliable_writer_qos() == qos_to_set.reliable_writer_qos()) + { + update_attributes = true; + w_att.times = new_qos.reliable_writer_qos().times; + w_att.disable_positive_acks = new_qos.reliable_writer_qos().disable_positive_acks.enabled; + w_att.keep_duration = new_qos.reliable_writer_qos().disable_positive_acks.duration; + } - set_qos(qos_, qos_to_set, !enabled); + wqos = new_qos.get_writerqos(get_publisher()->get_qos(), topic_->get_qos()); + deadline_changed = old_qos.deadline().period != new_qos.deadline().period; + lifespan_changed = old_qos.lifespan().duration != new_qos.lifespan().duration; - if (enabled) + if (update_attributes) { - if (qos_.reliability().kind == eprosima::fastrtps::RELIABLE_RELIABILITY_QOS && - qos_.reliable_writer_qos() == qos_to_set.reliable_writer_qos()) - { - // Update times and positive_acks attributes on RTPS Layer - WriterAttributes w_att; - w_att.times = qos_.reliable_writer_qos().times; - w_att.disable_positive_acks = qos_.reliable_writer_qos().disable_positive_acks.enabled; - w_att.keep_duration = qos_.reliable_writer_qos().disable_positive_acks.duration; - writer_->updateAttributes(w_att); - } + writer_->updateAttributes(w_att); + } - // Notify the participant that a Writer has changed its QOS - fastrtps::TopicAttributes topic_att = get_topic_attributes(qos_, *topic_, type_); - WriterQos wqos = qos_.get_writerqos(get_publisher()->get_qos(), topic_->get_qos()); - publisher_->rtps_participant()->updateWriter(writer_, topic_att, wqos); + // Notify the participant that a Writer has changed its QOS + fastrtps::TopicAttributes topic_att = get_topic_attributes(new_qos, *topic_, type_); + publisher_->rtps_participant()->updateWriter(writer_, topic_att, wqos); - // If the deadline period actually changed, (re)configure the timer. - if (old_qos.deadline().period != qos_.deadline().period) + if (deadline_changed) + { + configure_deadline_timer_(); + } + + // Lifespan + if (lifespan_changed) + { + if (new_qos.lifespan().duration != c_TimeInfinite) { - configure_deadline_timer_(); + lifespan_duration_us_ = + duration>(new_qos.lifespan().duration.to_ns() * 1e-3); + lifespan_timer_->update_interval_millisec(new_qos.lifespan().duration.to_ns() * 1e-6); } - - // Lifespan - if (old_qos.lifespan().duration != qos_.lifespan().duration) + else { - if (qos_.lifespan().duration != c_TimeInfinite) - { - lifespan_duration_us_ = - duration>(qos_.lifespan().duration.to_ns() * 1e-3); - lifespan_timer_->update_interval_millisec(qos_.lifespan().duration.to_ns() * 1e-6); - } - else - { - lifespan_timer_->cancel_timer(); - } + lifespan_timer_->cancel_timer(); } } @@ -1259,6 +1321,14 @@ const DataWriterQos& DataWriterImpl::get_qos() const return qos_; } +ReturnCode_t DataWriterImpl::get_qos( + DataWriterQos& qos) const +{ + std::lock_guard qos_guard(qos_mutex_); + qos = qos_; + return ReturnCode_t::RETCODE_OK; +} + ReturnCode_t DataWriterImpl::set_listener( DataWriterListener* listener) { @@ -1695,9 +1765,12 @@ ReturnCode_t DataWriterImpl::get_liveliness_lost_status( ReturnCode_t DataWriterImpl::assert_liveliness() { - if (writer_ == nullptr) { - return ReturnCode_t::RETCODE_NOT_ENABLED; + std::lock_guard qos_guard(qos_mutex_); + if (writer_ == nullptr) + { + return ReturnCode_t::RETCODE_NOT_ENABLED; + } } if (!publisher_->rtps_participant()->wlp()->assert_liveliness( @@ -1709,18 +1782,20 @@ ReturnCode_t DataWriterImpl::assert_liveliness() return ReturnCode_t::RETCODE_ERROR; } - if (qos_.liveliness().kind == MANUAL_BY_TOPIC_LIVELINESS_QOS) { - // As described in the RTPS specification, if liveliness kind is manual a heartbeat must be sent - // This only applies to stateful writers, as stateless writers do not send heartbeats - - StatefulWriter* stateful_writer = dynamic_cast(writer_); - - if (stateful_writer != nullptr) + std::lock_guard qos_guard(qos_mutex_); + if (qos_.liveliness().kind != MANUAL_BY_TOPIC_LIVELINESS_QOS) { - stateful_writer->send_periodic_heartbeat(true, true); + return ReturnCode_t::RETCODE_OK; } } + // As described in the RTPS specification, if liveliness kind is manual a heartbeat must be sent + // This only applies to stateful writers, as stateless writers do not send heartbeats + StatefulWriter* stateful_writer = dynamic_cast(writer_); + if (stateful_writer != nullptr) + { + stateful_writer->send_periodic_heartbeat(true, true); + } return ReturnCode_t::RETCODE_OK; } diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.hpp b/src/cpp/fastdds/publisher/DataWriterImpl.hpp index fbfb4483372..c66f01757ce 100644 --- a/src/cpp/fastdds/publisher/DataWriterImpl.hpp +++ b/src/cpp/fastdds/publisher/DataWriterImpl.hpp @@ -328,6 +328,9 @@ class DataWriterImpl : protected rtps::IReaderDataFilter const DataWriterQos& get_qos() const; + ReturnCode_t get_qos( + DataWriterQos& qos) const; + Topic* get_topic() const; const DataWriterListener* get_listener() const; @@ -408,6 +411,9 @@ class DataWriterImpl : protected rtps::IReaderDataFilter DataWriterQos qos_; + //! Mutex to protect qos_ + mutable std::mutex qos_mutex_; + //! DataWriterListener DataWriterListener* listener_ = nullptr; @@ -514,6 +520,8 @@ class DataWriterImpl : protected rtps::IReaderDataFilter DataRepresentationId_t data_representation_ {DEFAULT_DATA_REPRESENTATION}; + mutable std::mutex filters_mtx_; + ReturnCode_t check_write_preconditions( void* data, const InstanceHandle_t& handle, diff --git a/test/unittest/dds/publisher/DataWriterTests.cpp b/test/unittest/dds/publisher/DataWriterTests.cpp index 43f519e5e07..6994ef1c5aa 100644 --- a/test/unittest/dds/publisher/DataWriterTests.cpp +++ b/test/unittest/dds/publisher/DataWriterTests.cpp @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +#include #include #include #include @@ -2615,6 +2616,111 @@ TEST(DataWriterTests, data_type_is_plain_data_representation) DomainParticipantFactory::get_instance()->delete_participant(participant); } +/*! + * This test checks that concurrent set_qos() and get_qos() calls on an enabled DataWriter are race-free. + */ +TEST(DataWriterTests, ConcurrentSetQosVsGetQos) +{ + DomainParticipant* participant = + DomainParticipantFactory::get_instance()->create_participant(0, PARTICIPANT_QOS_DEFAULT); + ASSERT_NE(participant, nullptr); + + Publisher* publisher = participant->create_publisher(PUBLISHER_QOS_DEFAULT); + ASSERT_NE(publisher, nullptr); + + TypeSupport type(new TopicDataTypeMock()); + type.register_type(participant); + + Topic* topic = participant->create_topic("footopic_concurrent", type.get_type_name(), TOPIC_QOS_DEFAULT); + ASSERT_NE(topic, nullptr); + + DataWriter* datawriter = publisher->create_datawriter(topic, DATAWRITER_QOS_DEFAULT); + ASSERT_NE(datawriter, nullptr); + ASSERT_TRUE(datawriter->is_enabled()); + + // Two valid, mutable QoS configurations that differ in deadline period. + DataWriterQos qos_a = DATAWRITER_QOS_DEFAULT; + qos_a.deadline().period = eprosima::fastdds::dds::Duration_t(1, 0); + + DataWriterQos qos_b = DATAWRITER_QOS_DEFAULT; + qos_b.deadline().period = eprosima::fastdds::dds::Duration_t(2, 0); + + constexpr int kIterations = 5000; + + std::mutex barrier_mutex; + std::condition_variable barrier_cv; + bool go = false; + + std::atomic writer_error{false}; + std::atomic reader_error{false}; + + std::thread writer_thread([&]() + { + { + std::unique_lock lk(barrier_mutex); + barrier_cv.wait(lk, [&] + { + return go; + }); + } + for (int i = 0; i < kIterations; ++i) + { + const DataWriterQos& qos_to_set = (i % 2 == 0) ? qos_a : qos_b; + if (datawriter->set_qos(qos_to_set) != ReturnCode_t::RETCODE_OK) + { + writer_error.store(true); + break; + } + } + }); + + std::thread reader_thread([&]() + { + { + std::unique_lock lk(barrier_mutex); + barrier_cv.wait(lk, [&] + { + return go; + }); + } + DataWriterQos observed; + for (int i = 0; i < kIterations; ++i) + { + if (datawriter->get_qos(observed) != ReturnCode_t::RETCODE_OK) + { + reader_error.store(true); + break; + } + const eprosima::fastdds::dds::Duration_t& p = observed.deadline().period; + const bool valid = (p == qos_a.deadline().period) || + (p == qos_b.deadline().period) || + (p == DATAWRITER_QOS_DEFAULT.deadline().period); + if (!valid) + { + reader_error.store(true); + break; + } + } + }); + + { + std::lock_guard lk(barrier_mutex); + go = true; + } + barrier_cv.notify_all(); + + writer_thread.join(); + reader_thread.join(); + + EXPECT_FALSE(writer_error.load()) << "set_qos() returned an unexpected error code during concurrent run"; + EXPECT_FALSE(reader_error.load()) << "get_qos() returned a torn or unexpected QoS value during concurrent run"; + + ASSERT_TRUE(publisher->delete_datawriter(datawriter) == ReturnCode_t::RETCODE_OK); + ASSERT_TRUE(participant->delete_topic(topic) == ReturnCode_t::RETCODE_OK); + ASSERT_TRUE(participant->delete_publisher(publisher) == ReturnCode_t::RETCODE_OK); + ASSERT_TRUE(DomainParticipantFactory::get_instance()->delete_participant(participant) == ReturnCode_t::RETCODE_OK); +} + } // namespace dds } // namespace fastdds } // namespace eprosima diff --git a/test/unittest/statistics/dds/StatisticsDomainParticipantStatusQueryableTests/mock/fastdds/publisher/DataWriterImpl.hpp b/test/unittest/statistics/dds/StatisticsDomainParticipantStatusQueryableTests/mock/fastdds/publisher/DataWriterImpl.hpp index e6036a05b00..f695ac692b4 100644 --- a/test/unittest/statistics/dds/StatisticsDomainParticipantStatusQueryableTests/mock/fastdds/publisher/DataWriterImpl.hpp +++ b/test/unittest/statistics/dds/StatisticsDomainParticipantStatusQueryableTests/mock/fastdds/publisher/DataWriterImpl.hpp @@ -299,6 +299,13 @@ class DataWriterImpl : protected rtps::IReaderDataFilter return qos_; } + ReturnCode_t get_qos( + DataWriterQos& qos) const + { + qos = qos_; + return ReturnCode_t::RETCODE_OK; + } + Topic* get_topic() const { return topic_;