Included new mutex for class protection

zesk1999 · zesk1999 · commit febaf254b55a · 2025-09-30T11:32:30.000+02:00
Signed-off-by: zesk1999 &lt;zesk1999@gmail.com&gt;
diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.cpp b/src/cpp/fastdds/publisher/DataWriterImpl.cpp
@@ -248,6 +248,8 @@ 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(
@@ -431,15 +433,14 @@ ReturnCode_t DataWriterImpl::enable()
     }
 
     writer_ = BaseWriter::downcast(writer);
-    
+
     // Set DataWriterImpl as the implementer of the
     // IReaderDataFilter interface
     writer_->reader_data_filter(this);
 
     // In case it has been loaded from the persistence DB, rebuild instances on history
     history_->rebuild_instances();
 
-    std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
     configure_deadline_timer_();
 
     lifespan_timer_ = new TimedEvent(publisher_->rtps_participant()->get_resource_event(),
@@ -518,6 +519,8 @@ 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));
@@ -819,6 +822,8 @@ 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);
@@ -866,6 +871,8 @@ 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);
@@ -891,6 +898,8 @@ 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;
@@ -1069,7 +1078,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<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,
@@ -1178,6 +1187,8 @@ 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
@@ -1210,6 +1221,8 @@ 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;
@@ -1222,7 +1235,6 @@ 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();
@@ -1275,6 +1287,7 @@ ReturnCode_t DataWriterImpl::set_qos(
 
 const DataWriterQos& DataWriterImpl::get_qos() const
 {
+    std::unique_lock<RecursiveTimedMutex> il(impl_mtx_);
     return qos_;
 }
 
@@ -1555,30 +1568,21 @@ ReturnCode_t DataWriterImpl::set_related_datareader(
     return ret;
 }
 
-// deadline_timer_reschedule returns true if it could compute and set a new interval; false if there’s no pending deadline.
 bool DataWriterImpl::deadline_timer_reschedule()
 {
-    assert(qos_.deadline().period != dds::c_TimeInfinite);
-    if (deadline_timer_ == nullptr || deadline_missed_status_.total_count >= std::numeric_limits<uint32_t>::max())
-    {
-        return false;
-    }
-
     std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
 
+    assert(qos_.deadline().period != dds::c_TimeInfinite);
+    assert(deadline_timer_ != nullptr);
+    assert(deadline_missed_status_.total_count < std::numeric_limits<uint32_t>::max());
+
     steady_clock::time_point next_deadline_us;
     if (!history_->get_next_deadline(timer_owner_, next_deadline_us))
     {
         EPROSIMA_LOG_ERROR(DATA_WRITER, "Could not get the next deadline from the history");
         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;
@@ -1588,27 +1592,24 @@ void DataWriterImpl::configure_deadline_timer_()
 {
     std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
 
-    // Create the timer once 
+    // Create the timer once
     if (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
+            // 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
     if (qos_.deadline().period == dds::c_TimeInfinite)
     {
+        deadline_duration_us_ = std::chrono::duration<double, std::micro>::max();
         deadline_timer_->cancel_timer();
         return;
     }
@@ -1617,52 +1618,48 @@ void DataWriterImpl::configure_deadline_timer_()
 
     if (qos_.deadline().period.to_ns() == 0)
     {
+        deadline_timer_->cancel_timer();
+
+        deadline_missed_status_.total_count = std::numeric_limits<uint32_t>::max();
+        deadline_missed_status_.total_count_change = std::numeric_limits<uint32_t>::max();
         EPROSIMA_LOG_WARNING(
             DATA_WRITER,
             "Deadline period is 0, it will be ignored from now on. Timer is going to be canceled.");
 
         // Bump once and notify listener exactly once.
-        notify_deadline_missed_no_increment_();
-
-        deadline_missed_status_.total_count = std::numeric_limits<uint32_t>::max();
-        deadline_timer_->cancel_timer();
+        notify_deadline_missed_nts_();
         return;
     }
 
-    if (deadline_timer_reschedule())
-    {
-        deadline_timer_->restart_timer();
-    }
-    else
-    {
-        // Keep the timer object around but idle if there's no pending deadline
-        deadline_timer_->cancel_timer();
-    }
+    deadline_timer_->update_interval_millisec(qos_.deadline().period.to_ns() * 1e-6);
 }
 
-void DataWriterImpl::notify_deadline_missed_no_increment_()
+void DataWriterImpl::notify_deadline_missed_nts_()
 {
-    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_);
     }
+
+#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);
 }
 
 bool DataWriterImpl::deadline_missed()
 {
-    assert(qos_.deadline().period != dds::c_TimeInfinite);
-
     std::unique_lock<RecursiveTimedMutex> lock(writer_->getMutex());
 
+    assert(qos_.deadline().period != dds::c_TimeInfinite);
+
     deadline_missed_status_.total_count++;
     deadline_missed_status_.total_count_change++;
     deadline_missed_status_.last_instance_handle = timer_owner_;
 
-    notify_deadline_missed_no_increment_();
+    notify_deadline_missed_nts_();
 
     // If we just reached the max -> log ONCE, stop timer, and bail.
     if (deadline_missed_status_.total_count == std::numeric_limits<uint32_t>::max())
@@ -1672,10 +1669,6 @@ bool DataWriterImpl::deadline_missed()
         return false; // do not reschedule
     }
 
-#ifdef FASTDDS_STATISTICS
-    writer_listener_.notify_status_observer(statistics::StatusKind::DEADLINE_MISSED);
-#endif // FASTDDS_STATISTICS
-
     if (!history_->set_next_deadline(
                 timer_owner_,
                 steady_clock::now() + duration_cast<steady_clock::duration>(deadline_duration_us_)))
@@ -1804,6 +1797,8 @@ 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;
@@ -2532,4 +2527,4 @@ bool DataWriterImpl::is_relevant(
 
 } // namespace dds
 } // namespace fastdds
-} // namespace eprosima
+} // namespace eprosima
diff --git a/src/cpp/fastdds/publisher/DataWriterImpl.hpp b/src/cpp/fastdds/publisher/DataWriterImpl.hpp
@@ -20,6 +20,7 @@
 #define _FASTDDS_DATAWRITERIMPL_HPP_
 
 #include <memory>
+#include <mutex>
 
 #include <fastdds/dds/builtin/topic/PublicationBuiltinTopicData.hpp>
 #include <fastdds/dds/core/ReturnCode.hpp>
@@ -643,6 +644,7 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
 
     /**
      * @brief A method to reschedule the deadline timer
+     * @return true if it could compute and set a new interval, if there’s a pending deadline
      */
     bool deadline_timer_reschedule();
 
@@ -772,16 +774,22 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
 private:
 
     /**
-    * 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.
+    * 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.
     */
     void configure_deadline_timer_();
 
     /**
     * Notifies listeners that a deadline has been missed without touching the counters.
     */
-    void notify_deadline_missed_no_increment_();
+    void notify_deadline_missed_nts_();
 
     void create_history(
             const std::shared_ptr<IPayloadPool>& payload_pool,
diff --git a/test/blackbox/api/dds-pim/PubSubWriter.hpp b/test/blackbox/api/dds-pim/PubSubWriter.hpp
@@ -183,7 +183,6 @@ class PubSubWriter
         Listener(
                 PubSubWriter& writer)
             : writer_(writer)
-            , times_deadline_missed_(0)
             , times_liveliness_lost_(0)
             , times_unack_sample_removed_(0)
         {
@@ -214,7 +213,8 @@ class PubSubWriter
                 const eprosima::fastdds::dds::OfferedDeadlineMissedStatus& status) override
         {
             static_cast<void>(datawriter);
-            times_deadline_missed_ = status.total_count;
+            std::lock_guard<std::mutex> lk(mutex_);
+            offered_deadline_status_ = status;
         }
 
         void on_offered_incompatible_qos(
@@ -245,7 +245,14 @@ class PubSubWriter
 
         unsigned int missed_deadlines() const
         {
-            return times_deadline_missed_;
+            std::lock_guard<std::mutex> lk(mutex_);
+            return offered_deadline_status_.total_count;
+        }
+
+        unsigned int missed_deadlines_change() const
+        {
+            std::lock_guard<std::mutex> lk(mutex_);
+            return offered_deadline_status_.total_count_change;
         }
 
         unsigned int times_liveliness_lost() const
@@ -269,9 +276,10 @@ class PubSubWriter
                 const Listener&) = delete;
 
         PubSubWriter& writer_;
+        mutable std::mutex mutex_;
+
+        eprosima::fastdds::dds::OfferedDeadlineMissedStatus offered_deadline_status_{};
 
-        //! The number of times deadline was missed
-        unsigned int times_deadline_missed_;
         //! The number of times liveliness was lost
         unsigned int times_liveliness_lost_;
         //! The number of times a sample has been removed unacknowledged
@@ -287,12 +295,6 @@ class PubSubWriter
     typedef TypeSupport type_support;
     typedef typename type_support::type type;
 
-    eprosima::fastdds::dds::ReturnCode_t get_offered_deadline_missed_status(
-            eprosima::fastdds::dds::OfferedDeadlineMissedStatus& st)
-    {
-        return datawriter_->get_offered_deadline_missed_status(st);
-    }
-
     PubSubWriter(
             const std::string& topic_name)
         : participant_listener_(*this)
@@ -1761,6 +1763,11 @@ class PubSubWriter
         return listener_.missed_deadlines();
     }
 
+    unsigned int missed_deadlines_change() const
+    {
+        return listener_.missed_deadlines_change();
+    }
+
     unsigned int times_liveliness_lost() const
     {
         return listener_.times_liveliness_lost();
diff --git a/test/blackbox/common/BlackboxTestsDeadlineQos.cpp b/test/blackbox/common/BlackboxTestsDeadlineQos.cpp
