Address review: deadline timer (∞/0 handling), count_change reset, lifespan/deadline tests (#23289)

Author: zesk1999

src/cpp/fastdds/publisher/DataWriterImpl.cpp

@@ -431,6 +431,7 @@ ReturnCode_t DataWriterImpl::enable()
     }
 
     writer_ = BaseWriter::downcast(writer);
+    
     // Set DataWriterImpl as the implementer of the
     // IReaderDataFilter interface
     writer_->reader_data_filter(this);
@@ -439,7 +440,7 @@ ReturnCode_t DataWriterImpl::enable()
     history_->rebuild_instances();
 
     std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
-    configure_deadline_timer_locked_();
+    configure_deadline_timer_();
 
     lifespan_timer_ = new TimedEvent(publisher_->rtps_participant()->get_resource_event(),
                     [&]() -> bool
@@ -1068,7 +1069,7 @@ ReturnCode_t DataWriterImpl::perform_create_new_change(
             return RETCODE_TIMEOUT;
         }
 
-        if (deadline_timer_ != nullptr && qos_.deadline().period.to_ns() > 0 && qos_.deadline().period != dds::c_TimeInfinite && deadline_missed_status_.total_count < std::numeric_limits<int32_t>::max())
+        if (deadline_timer_ != nullptr && 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,
@@ -1214,12 +1215,16 @@ ReturnCode_t DataWriterImpl::set_qos(
         return RETCODE_IMMUTABLE_POLICY;
     }
 
-    const dds::Duration_t prev_deadline = qos_.deadline().period;
+    // Take a snapshot of the current QoS before mutating it
+    const DataWriterQos old_qos = qos_;
 
     set_qos(qos_, qos_to_set, !enabled);
 
     if (enabled)
     {
+        std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
+        // Locks after we checked that writer exists
+
         int32_t transport_priority = writer_->get_transport_priority();
 
         if ((qos_.reliability().kind == ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS &&
@@ -1239,30 +1244,29 @@ ReturnCode_t DataWriterImpl::set_qos(
         WriterQos wqos = qos_.get_writerqos(get_publisher()->get_qos(), topic_->get_qos());
         publisher_->rtps_participant()->update_writer(writer_, wqos);
 
-
-        std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
-
-        // If the deadline period actually changed, (re) the timer.
-        if (qos_.deadline().period != prev_deadline)
+        // If the deadline period actually changed, (re)configure the timer.
+        if (old_qos.deadline().period != qos_.deadline().period)
         {
             // Resetting total count value whenever the deadline period changes
             deadline_missed_status_.total_count = 0;
             deadline_missed_status_.total_count_change = 0;
             deadline_missed_status_.last_instance_handle = InstanceHandle_t();
 
-            configure_deadline_timer_locked_();
+            configure_deadline_timer_();
         }
 
         // Lifespan
-        if (qos_.lifespan().duration != dds::c_TimeInfinite)
-        {
-            lifespan_duration_us_ =
-                    duration<double, std::ratio<1, 1000000>>(qos_.lifespan().duration.to_ns() * 1e-3);
-            lifespan_timer_->update_interval_millisec(qos_.lifespan().duration.to_ns() * 1e-6);
-        }
-        else
-        {
-            lifespan_timer_->cancel_timer();
+        if (old_qos.lifespan().duration != qos_.lifespan().duration) {
+            if (qos_.lifespan().duration != dds::c_TimeInfinite)
+            {
+                lifespan_duration_us_ =
+                        duration<double, std::ratio<1, 1000000>>(qos_.lifespan().duration.to_ns() * 1e-3);
+                lifespan_timer_->update_interval_millisec(qos_.lifespan().duration.to_ns() * 1e-6);
+            }
+            else
+            {
+                lifespan_timer_->cancel_timer();
+            }
         }
     }
 
@@ -1555,7 +1559,7 @@ ReturnCode_t DataWriterImpl::set_related_datareader(
 bool DataWriterImpl::deadline_timer_reschedule()
 {
     assert(qos_.deadline().period != dds::c_TimeInfinite);
-    if (deadline_timer_ == nullptr || deadline_missed_status_.total_count >= std::numeric_limits<int32_t>::max())
+    if (deadline_timer_ == nullptr || deadline_missed_status_.total_count >= std::numeric_limits<uint32_t>::max())
     {
         return false;
     }
@@ -1569,128 +1573,109 @@ bool DataWriterImpl::deadline_timer_reschedule()
         return false;
     }
 
+    if (next_deadline_us <= std::chrono::steady_clock::now())
+    {
+        // Ignore uninitialized or stale deadlines; don't arm to avoid a spurious first miss
+        return false;
+    }
+
     auto interval_ms = duration_cast<milliseconds>(next_deadline_us - steady_clock::now());
     deadline_timer_->update_interval_millisec(static_cast<double>(interval_ms.count()));
     return true;
 }
 
-/**
- * This function tears down any existing deadline timer and creates a new one
- * configured for the current deadline period. If the period is 0 or infinite,
- * the timer is created to fire only once to log a warning and then cancel itself.
- */
-void DataWriterImpl::configure_deadline_timer_locked_()
+void DataWriterImpl::configure_deadline_timer_()
 {
-    if (deadline_timer_ != nullptr)
+    std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
+
+    // Create the timer once 
+    if (deadline_timer_ == nullptr)
     {
-        deadline_timer_->cancel_timer();
-        delete deadline_timer_;
-        deadline_timer_ = nullptr;
+        deadline_timer_ = new TimedEvent(
+            publisher_->rtps_participant()->get_resource_event(),
+            [this]() -> bool
+            {
+                std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
+
+                // Normal path: process a deadline miss. This function will
+                // update counters, notify, and ask for reschedule as needed.
+                return deadline_missed();
+            },
+            (qos_.deadline().period == dds::c_TimeInfinite) ? std::numeric_limits<double>::max() : qos_.deadline().period.to_ns() * 1e-6
+            // In case of deadline period = infinite, multiplying it by 1e-6 could cause a problem
+        );
     }
 
-    // If we had previously hit the max, keep timer disabled until QoS changes reset the count.
-    if (deadline_missed_status_.total_count >= std::numeric_limits<int32_t>::max())
+    // Handle "infinite" and "zero" outside the callback
+    if (qos_.deadline().period == dds::c_TimeInfinite)
     {
-        return; // no new timer
+        deadline_timer_->cancel_timer();
+        return;
     }
 
-    // Refresh cached duration and guards
-    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);
 
-    // Create the timer with a safe callback
-    deadline_timer_ = new TimedEvent(
-        publisher_->rtps_participant()->get_resource_event(),
-        [this]() -> bool
-        {
-            std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
+    if (qos_.deadline().period.to_ns() == 0)
+    {
+        EPROSIMA_LOG_WARNING(
+            DATA_WRITER,
+            "Deadline period is 0, it will be ignored from now on. Timer is going to be canceled.");
 
-            if ((qos_.deadline().period.to_ns() == 0) || (qos_.deadline().period == dds::c_TimeInfinite))
-            {
-                
-                EPROSIMA_LOG_WARNING(
-                    DATA_WRITER,
-                    "Deadline period is "
-                        << (qos_.deadline().period.to_ns() == 0 ? "0" : "infinite")
-                        << ", it will be ignored from now on. Timer is going to be canceled."); 
-                deadline_missed_status_.total_count = std::numeric_limits<int32_t>::max();
-
-                if (deadline_timer_ != nullptr)
-                {
-                    deadline_timer_->cancel_timer();
-                }
-                return false;
-            }
+        // Bump once and notify listener exactly once.
+        notify_deadline_missed_no_increment_();
 
-            // Normal path
-            return deadline_missed();
-        },
-        0.0);
+        deadline_missed_status_.total_count = std::numeric_limits<uint32_t>::max();
+        deadline_timer_->cancel_timer();
+        return;
+    }
 
-    if ((qos_.deadline().period.to_ns() > 0) && (qos_.deadline().period != dds::c_TimeInfinite))
+    if (deadline_timer_reschedule())
     {
-        // Only arm if a next deadline is already computed; otherwise keep disabled to avoid spurious fire.
-        std::chrono::steady_clock::time_point next_tp;
-        if (history_->get_next_deadline(timer_owner_, next_tp))
-        {
-            auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(
-                          next_tp - std::chrono::steady_clock::now()).count();
-            if (ms < 0) ms = 0;
-
-            deadline_timer_->update_interval_millisec(static_cast<double>(ms));
-            deadline_timer_->restart_timer();
-        }
-        else
-        {
-            deadline_timer_->cancel_timer(); // keep it created but idle
-        }
+        deadline_timer_->restart_timer();
     }
     else
     {
-        // 0 or infinite -> fire once to log and self-disable
-        deadline_timer_->restart_timer();
+        // Keep the timer object around but idle if there's no pending deadline
+        deadline_timer_->cancel_timer();
     }
 }
 
+void DataWriterImpl::notify_deadline_missed_no_increment_()
+{
+    std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
+
+    StatusMask notify_status = StatusMask::offered_deadline_missed();
+    if (auto* listener = get_listener_for(notify_status))
+    {
+        listener->on_offered_deadline_missed(user_datawriter_, deadline_missed_status_);
+    }
+    user_datawriter_->get_statuscondition().get_impl()->set_status(notify_status, true);
+}
 
 bool DataWriterImpl::deadline_missed()
 {
     assert(qos_.deadline().period != dds::c_TimeInfinite);
 
     std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
 
-    if (deadline_missed_status_.total_count >= std::numeric_limits<int32_t>::max() && deadline_timer_ != nullptr)
-    {
-        deadline_timer_->cancel_timer();
-        return false; 
-    }
-
     deadline_missed_status_.total_count++;
     deadline_missed_status_.total_count_change++;
     deadline_missed_status_.last_instance_handle = timer_owner_;
 
+    notify_deadline_missed_no_increment_();
+
     // If we just reached the max -> log ONCE, stop timer, and bail.
-    if (deadline_missed_status_.total_count == std::numeric_limits<int32_t>::max())
+    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.");
         deadline_timer_->cancel_timer();
         return false; // do not reschedule
     }
 
-    StatusMask notify_status = StatusMask::offered_deadline_missed();
-    auto listener = get_listener_for(notify_status);
-    if (nullptr != listener)
-    {
-        listener->on_offered_deadline_missed(user_datawriter_, deadline_missed_status_);
-        deadline_missed_status_.total_count_change = 0;
-    }
-
 #ifdef FASTDDS_STATISTICS
     writer_listener_.notify_status_observer(statistics::StatusKind::DEADLINE_MISSED);
 #endif // FASTDDS_STATISTICS
 
-    user_datawriter_->get_statuscondition().get_impl()->set_status(notify_status, true);
-
     if (!history_->set_next_deadline(
                 timer_owner_,
                 steady_clock::now() + duration_cast<steady_clock::duration>(deadline_duration_us_)))
@@ -2547,4 +2532,4 @@ bool DataWriterImpl::is_relevant(
 
 } // namespace dds
 } // namespace fastdds
-} // namespace eprosima
+} // namespace eprosima

src/cpp/fastdds/publisher/DataWriterImpl.hpp

@@ -771,7 +771,17 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
 
 private:
 
-    void configure_deadline_timer_locked_();
+    /**
+    * This function tears down any existing deadline timer and creates a new one
+    * configured for the current deadline period. If the period is 0 or infinite,
+    * the timer is created to fire only once to log a warning and then cancel itself.
+    */
+    void configure_deadline_timer_();
+
+    /**
+    * Notifies listeners that a deadline has been missed without touching the counters.
+    */
+    void notify_deadline_missed_no_increment_();
 
     void create_history(
             const std::shared_ptr<IPayloadPool>& payload_pool,