Refs #23289. Handle maximum deadline misses case.

Signed-off-by: zesk1999 <[email protected]>

Author: zesk1999

src/cpp/fastdds/publisher/DataWriterImpl.cpp

@@ -248,8 +248,6 @@ void DataWriterImpl::create_history(
 
 ReturnCode_t DataWriterImpl::enable()
 {
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
-
     assert(writer_ == nullptr);
 
     auto history_att = DataWriterHistory::to_history_attributes(
@@ -519,8 +517,6 @@ ReturnCode_t DataWriterImpl::loan_sample(
         void*& sample,
         LoanInitializationKind initialization)
 {
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
-
     // Block lowlevel writer
     auto max_blocking_time = steady_clock::now() +
             microseconds(rtps::TimeConv::Time_t2MicroSecondsInt64(qos_.reliability().max_blocking_time));
@@ -822,8 +818,6 @@ InstanceHandle_t DataWriterImpl::do_register_instance(
         const InstanceHandle_t instance_handle,
         WriteParams& wparams)
 {
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
-
     // TODO(MiguelCompany): wparams should be used when propagating the register_instance operation to the DataReader.
     // See redmine issue #14494
     static_cast<void>(wparams);
@@ -871,8 +865,6 @@ ReturnCode_t DataWriterImpl::unregister_instance(
         const InstanceHandle_t& handle,
         bool dispose)
 {
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
-
     // Preconditions
     InstanceHandle_t ih;
     ReturnCode_t returned_value = check_instance_preconditions(instance, handle, ih);
@@ -898,8 +890,6 @@ ReturnCode_t DataWriterImpl::unregister_instance_w_timestamp(
         const fastdds::dds::Time_t& timestamp,
         bool dispose)
 {
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
-
     // Preconditions
     InstanceHandle_t instance_handle;
     ReturnCode_t ret = RETCODE_OK;
@@ -1078,7 +1068,8 @@ ReturnCode_t DataWriterImpl::perform_create_new_change(
             return RETCODE_TIMEOUT;
         }
 
-        if (qos_.deadline().period.to_ns() > 0 && qos_.deadline().period != dds::c_TimeInfinite && deadline_missed_status_.total_count < std::numeric_limits<uint32_t>::max())
+        if (qos_.deadline().period.to_ns() > 0 && qos_.deadline().period != dds::c_TimeInfinite &&
+                deadline_missed_status_.total_count < std::numeric_limits<uint32_t>::max())
         {
             if (!history_->set_next_deadline(
                         handle,
@@ -1187,8 +1178,6 @@ InstanceHandle_t DataWriterImpl::get_instance_handle() const
 
 void DataWriterImpl::publisher_qos_updated()
 {
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
-
     if (writer_ != nullptr)
     {
         // NOTIFY THE BUILTIN PROTOCOLS THAT THE WRITER HAS CHANGED
@@ -1221,8 +1210,6 @@ ReturnCode_t DataWriterImpl::set_qos(
         }
     }
 
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
-
     if (enabled && !can_qos_be_updated(qos_, qos_to_set))
     {
         return RETCODE_IMMUTABLE_POLICY;
@@ -1235,6 +1222,7 @@ ReturnCode_t DataWriterImpl::set_qos(
 
     if (enabled)
     {
+        std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
         // Locks after we checked that writer exists
 
         int32_t transport_priority = writer_->get_transport_priority();
@@ -1268,7 +1256,8 @@ ReturnCode_t DataWriterImpl::set_qos(
         }
 
         // Lifespan
-        if (old_qos.lifespan().duration != qos_.lifespan().duration) {
+        if (old_qos.lifespan().duration != qos_.lifespan().duration)
+        {
             if (qos_.lifespan().duration != dds::c_TimeInfinite)
             {
                 lifespan_duration_us_ =
@@ -1287,7 +1276,6 @@ ReturnCode_t DataWriterImpl::set_qos(
 
 const DataWriterQos& DataWriterImpl::get_qos() const
 {
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
     return qos_;
 }
 
@@ -1603,7 +1591,7 @@ void DataWriterImpl::configure_deadline_timer_()
             },
             // Park timer with a huge interval (prevents spurious callbacks); we'll arm/cancel explicitly
             std::numeric_limits<double>::max()
-        );
+            );
     }
 
     // Handle "infinite" and "zero" outside the callback
@@ -1614,7 +1602,8 @@ void DataWriterImpl::configure_deadline_timer_()
         return;
     }
 
-    deadline_duration_us_ = std::chrono::duration<double, std::ratio<1, 1000000>>(qos_.deadline().period.to_ns() * 1e-3);
+    deadline_duration_us_ =
+            std::chrono::duration<double, std::ratio<1, 1000000>>(qos_.deadline().period.to_ns() * 1e-3);
 
     if (qos_.deadline().period.to_ns() == 0)
     {
@@ -1664,7 +1653,8 @@ bool DataWriterImpl::deadline_missed()
     // If we just reached the max -> log ONCE, stop timer, and bail.
     if (deadline_missed_status_.total_count == std::numeric_limits<uint32_t>::max())
     {
-        EPROSIMA_LOG_WARNING(DATA_WRITER, "Maximum number of deadline missed messages reached. Stopping deadline timer.");
+        EPROSIMA_LOG_WARNING(DATA_WRITER,
+                "Maximum number of deadline missed messages reached. Stopping deadline timer.");
         deadline_timer_->cancel_timer();
         return false; // do not reschedule
     }
@@ -1797,8 +1787,6 @@ ReturnCode_t DataWriterImpl::assert_liveliness()
 ReturnCode_t DataWriterImpl::get_publication_builtin_topic_data(
         PublicationBuiltinTopicData& publication_data) const
 {
-    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
-
     if (nullptr == writer_)
     {
         return RETCODE_NOT_ENABLED;

src/cpp/fastdds/publisher/DataWriterImpl.hpp

@@ -774,21 +774,16 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
 private:
 
     /**
-    * Protects implementation class
-    */
-    mutable RecursiveTimedMutex impl_mtx_;
-
-    /**
-    * (Re)configures the deadline timer:
-    *  Create once, parked with a huge interval (idle).
-    *  In case of deadline period ∞ cancel it, for 0 warn and notify once; set counts to max and
-    *  for values >0 store period.
-    */
+     * (Re)configures the deadline timer:
+     *  Create once, parked with a huge interval (idle).
+     *  In case of deadline period ∞ cancel it, for 0 warn and notify once; set counts to max and
+     *  for values >0 store period.
+     */
     void configure_deadline_timer_();
 
     /**
-    * Notifies listeners that a deadline has been missed without touching the counters.
-    */
+     * Notifies listeners that a deadline has been missed without touching the counters.
+     */
     void notify_deadline_missed_nts_();
 
     void create_history(

test/blackbox/common/BlackboxTestsDeadlineQos.cpp

@@ -317,22 +317,34 @@ TEST_P(DeadlineQos, KeyedTopicBestEffortReaderVolatileWriterSetDeadline)
  * Creating a DataWriter with a deadline of 0.
  * Checking if a warning is logged exactly once, the timer is cancelled without missed deadline
  * messages and a total count and count change set to max integer.
+ * Checking warnings, total count and count change when changing the deadline at runtime
  */
 TEST_P(DeadlineQos, ZeroDeadlinePeriod)
 {
     // Local helper used only by this test
     struct LocalWarningCounter : fastlog::LogConsumer
     {
-        explicit LocalWarningCounter(std::string needle) : needle_(std::move(needle)) {}
-        void Consume(const fastlog::Log::Entry& e) override
+        explicit LocalWarningCounter(
+                std::string needle)
+            : needle_(std::move(needle))
+        {
+        }
+
+        void Consume(
+                const fastlog::Log::Entry& e) override
         {
             if (e.kind == fastlog::Log::Kind::Warning &&
-                e.message.find(needle_) != std::string::npos)
+                    e.message.find(needle_) != std::string::npos)
             {
                 count_.fetch_add(1, std::memory_order_relaxed);
             }
         }
-        size_t count() const { return count_.load(std::memory_order_relaxed); }
+
+        size_t count() const
+        {
+            return count_.load(std::memory_order_relaxed);
+        }
+
         std::string needle_;
         std::atomic<size_t> count_{0};
     };
@@ -362,7 +374,8 @@ TEST_P(DeadlineQos, ZeroDeadlinePeriod)
 
     std::this_thread::sleep_for(std::chrono::milliseconds(150));
 
-    EXPECT_EQ(writer.missed_deadlines(), std::numeric_limits<uint32_t>::max()) << "Expected the max value after a zero-deadline warning.";
+    EXPECT_EQ(writer.missed_deadlines(),
+            std::numeric_limits<uint32_t>::max()) << "Expected the max value after a zero-deadline warning.";
     EXPECT_EQ(writer.missed_deadlines_change(), std::numeric_limits<uint32_t>::max());
 
     EXPECT_EQ(consumer_ptr->count(), 1u) << "Expected exactly one 'deadline=0' warning\n";
@@ -383,6 +396,55 @@ TEST_P(DeadlineQos, ZeroDeadlinePeriod)
     std::this_thread::sleep_for(std::chrono::milliseconds(100));
     EXPECT_EQ(consumer_ptr->count(), prev) << "Timer should be canceled; no more warnings";
 
+    eprosima::fastdds::dds::DataWriter& native = writer.get_native_writer();
+
+    // Helper: double seconds -> Duration_t
+    auto to_duration = [](double s) -> eprosima::fastdds::dds::Duration_t
+            {
+                eprosima::fastdds::dds::Duration_t d{};
+                d.seconds = static_cast<int32_t>(s);
+                d.nanosec = static_cast<uint32_t>((s - d.seconds) * 1e9);
+                return d;
+            };
+
+    // Helper: set writer deadline at runtime via standard DDS API
+    auto set_deadline = [&](double seconds)
+            {
+                eprosima::fastdds::dds::DataWriterQos q;
+                ASSERT_EQ(native.get_qos(q), eprosima::fastdds::dds::RETCODE_OK);
+                q.deadline().period = to_duration(seconds);
+                ASSERT_EQ(native.set_qos(q), eprosima::fastdds::dds::RETCODE_OK);
+            };
+
+    set_deadline(0.1); // Update 0 -> finite: counters must reset, no new warning
+
+    eprosima::fastdds::dds::OfferedDeadlineMissedStatus st{};
+    std::this_thread::sleep_for(std::chrono::milliseconds(150));
+    ASSERT_EQ(native.get_offered_deadline_missed_status(st), eprosima::fastdds::dds::RETCODE_OK);
+    EXPECT_EQ(st.total_count, 0u);
+    EXPECT_EQ(st.total_count_change, 0u);
+    EXPECT_EQ(consumer_ptr->count(), prev) << "No new warning when moving from 0 -> finite";
+
+    set_deadline(0.0); // Update finite -> 0: exactly one new warning, counters jump to max
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    ASSERT_EQ(native.get_offered_deadline_missed_status(st), eprosima::fastdds::dds::RETCODE_OK);
+    EXPECT_EQ(consumer_ptr->count(), prev + 1) << "Exactly one extra warning on finite -> 0";
+    EXPECT_EQ(st.total_count, std::numeric_limits<uint32_t>::max());
+    EXPECT_EQ(st.total_count_change, std::numeric_limits<uint32_t>::max());
+
+    set_deadline(0.001); // Update 0 -> finite small: missed deadlines registered
+    KeyedHelloWorld sample;
+    sample.key(1);
+    ASSERT_EQ(native.write(&sample), eprosima::fastdds::dds::RETCODE_OK);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    ASSERT_EQ(native.get_offered_deadline_missed_status(st), eprosima::fastdds::dds::RETCODE_OK);
+    EXPECT_GT(st.total_count, 0u);
+    EXPECT_GT(st.total_count_change, 0u);
+    EXPECT_LT(st.total_count, std::numeric_limits<uint32_t>::max());
+    EXPECT_LT(st.total_count_change, std::numeric_limits<uint32_t>::max());
+
     fastlog::Log::ClearConsumers();
 }