[#24061] Solved TSAN race conditions + removed 'bridges_mutex_'

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

Author: Danipiza

ddspipe_core/include/ddspipe_core/core/DdsPipe.hpp

@@ -379,6 +379,14 @@ class DdsPipe
      */
     void deactivate_all_topics_nts_() noexcept;
 
+    /**
+     * @brief Update partitions for all bridges.
+     *
+     * This method must be called with \c mutex_ locked.
+     */
+    void update_partitions_nts_(
+            const std::set<std::string>& partitions_set);
+
     //////////////////////////
     // CONFIGURATION VARIABLES
     //////////////////////////
@@ -459,11 +467,6 @@ class DdsPipe
      * @brief Internal mutex for concurrent calls
      */
     mutable std::mutex mutex_;
-
-    /**
-     * @brief Internal mutex for bridges
-     */
-    mutable std::mutex bridges_mutex_;
 };
 
 } /* namespace core */

ddspipe_core/src/cpp/communication/dds/DdsBridge.cpp

@@ -285,6 +285,8 @@ void DdsBridge::add_writers_to_tracks_nts_(
 void DdsBridge::update_partitions(
         const std::set<std::string>& partitions_set)
 {
+    std::lock_guard<std::mutex> lock(mutex_);
+
     for (const auto& track: tracks_)
     {
         track.second->update_writers_topic_partitions(topic_->partition_name);
@@ -295,6 +297,8 @@ void DdsBridge::update_partitions(
 void DdsBridge::update_topic_filter(
         const std::string& expression)
 {
+    std::lock_guard<std::mutex> lock(mutex_);
+
     for (const auto& track: tracks_)
     {
         track.second->update_reader_content_filter(expression);
@@ -347,6 +351,8 @@ void DdsBridge::add_partition_to_topic(
         std::string guid,
         std::string partition)
 {
+    std::lock_guard<std::mutex> lock(mutex_);
+
     topic_->partition_name[guid] = partition;
 }
 

ddspipe_core/src/cpp/core/DdsPipe.cpp

@@ -303,29 +303,24 @@ void DdsPipe::discovered_endpoint_nts_(
         // update the track of the topic to
         // add the partition in the reader if it is in the filter
 
-        // update partitions under bridges_mutex_
+        const auto bridge_it = bridges_.find(utils::Heritable<DdsTopic>::make_heritable(endpoint.topic));
+        // add the specific partition of the endpoint in the bridge topic.
+        if (bridge_it != bridges_.end())
         {
-            std::lock_guard<std::mutex> lock(bridges_mutex_);
+            std::ostringstream guid_ss;
+            guid_ss << endpoint.guid;
 
-            const auto bridge_it = bridges_.find(utils::Heritable<DdsTopic>::make_heritable(endpoint.topic));
-            // add the specific partition of the endpoint in the bridges topic.
-            if (bridge_it != bridges_.end())
+            const auto part_it = endpoint.specific_partitions.find(guid_ss.str());
+            if (part_it != endpoint.specific_partitions.end())
             {
-                std::ostringstream guid_ss;
-                guid_ss << endpoint.guid;
-
-                const auto part_it = endpoint.specific_partitions.find(guid_ss.str());
-                if (part_it != endpoint.specific_partitions.end())
-                {
-                    bridge_it->second->add_partition_to_topic(guid_ss.str(), part_it->second);
-                }
+                bridge_it->second->add_partition_to_topic(guid_ss.str(), part_it->second);
             }
         }
 
         // update readers outside the lock
         if (!filter_partition_.empty())
         {
-            update_partitions(filter_partition_);
+            update_partitions_nts_(filter_partition_);
         }
     }
 }
@@ -626,6 +621,14 @@ void DdsPipe::deactivate_all_topics_nts_() noexcept
 
 void DdsPipe::update_partitions(
         const std::set<std::string>& partitions_set)
+{
+    std::lock_guard<std::mutex> lock(mutex_);
+
+    update_partitions_nts_(partitions_set);
+}
+
+void DdsPipe::update_partitions_nts_(
+        const std::set<std::string>& partitions_set)
 {
     // TODO.
     // In the future it could be interesting to update the partitions
@@ -641,6 +644,8 @@ void DdsPipe::update_content_filter(
         const std::string& topic_name,
         const std::string& expression)
 {
+    std::lock_guard<std::mutex> lock(mutex_);
+
     for (const auto& pair : bridges_)
     {
         if (pair.first->m_topic_name == topic_name)
@@ -653,8 +658,7 @@ void DdsPipe::update_content_filter(
 void DdsPipe::update_filter(
         const std::set<std::string> filter_partition_set)
 {
-    // Avoid possible datarace with partitions filter
-    std::lock_guard<std::mutex> lock(bridges_mutex_);
+    std::lock_guard<std::mutex> lock(mutex_);
 
     filter_partition_ = std::move(filter_partition_set);
 }

ddspipe_participants/src/cpp/writer/dynamic_types/SchemaWriter.cpp

@@ -59,6 +59,8 @@ void SchemaWriter::update_content_topic_filter(
 void SchemaWriter::update_topic_partitions(
         const std::map<std::string, std::string>& partition_name)
 {
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+
     topic_.partition_name = partition_name;
 }