Protect PDPStatelessWriter attributes (#6100)

* Refs #23822. Protect PDPStatelessWriter attributes

Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com>

* Refs #23822. Atomize matched_ attribute

Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com>

* Refs #23822. Protect reader_data_filter_ attribute

Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com>

---------

Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com>
(cherry picked from commit 6ca678b)

Author: juanlofer-eprosima

src/cpp/rtps/builtin/discovery/participant/simple/PDPStatelessWriter.cpp

@@ -55,6 +55,7 @@ PDPStatelessWriter::PDPStatelessWriter(
 bool PDPStatelessWriter::matched_reader_add_edp(
         const ReaderProxyData& data)
 {
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
     bool ret = StatelessWriter::matched_reader_add_edp(data);
     if (ret)
     {
@@ -69,6 +70,7 @@ bool PDPStatelessWriter::matched_reader_add_edp(
 bool PDPStatelessWriter::matched_reader_remove(
         const GUID_t& reader_guid)
 {
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
     bool ret = StatelessWriter::matched_reader_remove(reader_guid);
     if (ret)
     {
@@ -82,14 +84,15 @@ void PDPStatelessWriter::unsent_change_added_to_history(
         CacheChange_t* change,
         const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time)
 {
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
     mark_all_readers_interested();
     StatelessWriter::unsent_change_added_to_history(change, max_blocking_time);
 }
 
 void PDPStatelessWriter::set_initial_peers(
         const LocatorList& locator_list)
 {
-    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+    std::lock_guard<std::mutex> guard(interested_readers_mutex_);
 
     initial_peers_.push_back(locator_list);
     mp_RTPSParticipant->createSenderResources(initial_peers_);
@@ -106,6 +109,8 @@ bool PDPStatelessWriter::send_to_fixed_locators(
         const uint32_t& total_bytes,
         std::chrono::steady_clock::time_point& max_blocking_time_point) const
 {
+    std::lock_guard<std::mutex> guard(interested_readers_mutex_);
+
     bool ret = true;
 
     if (should_reach_all_destinations_)
@@ -133,13 +138,15 @@ bool PDPStatelessWriter::is_relevant(
         const fastdds::rtps::CacheChange_t& /* change */,
         const fastdds::rtps::GUID_t& reader_guid) const
 {
+    std::lock_guard<std::mutex> guard(interested_readers_mutex_);
     return interested_readers_.end() !=
            std::find(interested_readers_.begin(), interested_readers_.end(), reader_guid);
 }
 
 void PDPStatelessWriter::mark_all_readers_interested()
 {
     std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+    std::lock_guard<std::mutex> _guard(interested_readers_mutex_);
     should_reach_all_destinations_ = true;
     interested_readers_.clear();
     fixed_locators_.clear();
@@ -151,6 +158,7 @@ void PDPStatelessWriter::add_interested_reader(
         const GUID_t& reader_guid)
 {
     std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+    std::lock_guard<std::mutex> _guard(interested_readers_mutex_);
     if (!should_reach_all_destinations_)
     {
         auto it = std::find(interested_readers_.begin(), interested_readers_.end(), reader_guid);
@@ -165,7 +173,7 @@ void PDPStatelessWriter::add_interested_reader(
 void PDPStatelessWriter::remove_interested_reader(
         const GUID_t& reader_guid)
 {
-    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+    std::lock_guard<std::mutex> guard(interested_readers_mutex_);
     interested_readers_.remove(reader_guid);
 }
 

src/cpp/rtps/builtin/discovery/participant/simple/PDPStatelessWriter.hpp

@@ -20,6 +20,7 @@
 #define FASTDDS_RTPS_BUILTIN_DISCOVERY_PARTICIPANT_SIMPLE__PDPSTATELESSWRITER_HPP
 
 #include <chrono>
+#include <mutex>
 
 #include <fastdds/rtps/common/LocatorList.hpp>
 #include <fastdds/rtps/interfaces/IReaderDataFilter.hpp>
@@ -144,6 +145,8 @@ class PDPStatelessWriter : public StatelessWriter, private IReaderDataFilter
     mutable ResourceLimitedVector<GUID_t> interested_readers_;
     //! Whether we have set that all destinations are interested
     mutable bool should_reach_all_destinations_ = false;
+    //! Mutex to protect attributes of this class
+    mutable std::mutex interested_readers_mutex_;
 
 };
 

test/blackbox/common/RTPSWithRegistrationWriter.hpp

@@ -673,7 +673,7 @@ class RTPSWithRegistrationWriter
     bool initialized_;
     std::mutex mutex_;
     std::condition_variable cv_;
-    uint32_t matched_;
+    std::atomic<uint32_t> matched_;
     eprosima::fastdds::rtps::EntityId_t custom_entity_id_ = eprosima::fastdds::rtps::c_EntityId_Unknown;
     type_support type_;
     std::shared_ptr<eprosima::fastdds::rtps::IPayloadPool> payload_pool_;