[24604] Fix minor dataraces

src/cpp/fastdds/core/condition/StatusConditionImpl.cpp

@@ -67,12 +67,16 @@ ReturnCode_t StatusConditionImpl::set_enabled_statuses(
 const StatusMask& StatusConditionImpl::get_enabled_statuses() const
 {
     std::lock_guard<std::mutex> guard(mutex_);
+    FASTDDS_TODO_BEFORE(4, 0, "Same issue as get_raw_status, reference escapes the lock.");
     return mask_;
 }
 
 const StatusMask& StatusConditionImpl::get_raw_status() const
 {
     std::lock_guard<std::mutex> guard(mutex_);
+    FASTDDS_TODO_BEFORE(4, 0,
+            "Returning a reference to a member protected by a mutex that escapes the lock is a data race. "
+            "Fix requires changing the return type to StatusMask (by value), which is an API break.");
     return status_;
 }
 

src/cpp/rtps/writer/StatefulWriter.cpp

@@ -445,11 +445,14 @@ bool StatefulWriter::intraprocess_heartbeat(
         bool liveliness)
 {
     bool returned_value = false;
+    std::unique_lock<RecursiveTimedMutex> lockW(mp_mutex);
     LocalReaderPointer::Instance local_reader = reader_proxy->local_reader();
+    lockW.unlock();
 
     if (local_reader)
     {
-        std::unique_lock<RecursiveTimedMutex> lockW(mp_mutex);
+        // std::unique_lock<RecursiveTimedMutex> lockW(mp_mutex);
+        lockW.lock();
         SequenceNumber_t first_seq = get_seq_num_min();
         SequenceNumber_t last_seq = get_seq_num_max();
 

src/cpp/rtps/writer/StatelessWriter.cpp

@@ -317,6 +317,7 @@ bool StatelessWriter::intraprocess_delivery(
         CacheChange_t* change,
         ReaderLocator& reader_locator)
 {
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
     LocalReaderPointer::Instance local_reader = reader_locator.local_reader();
 
     if (local_reader &&

src/cpp/security/accesscontrol/PermissionsParser.cpp

@@ -17,10 +17,16 @@
 
 #include <cstring>
 #include <cassert>
+#include <ctime>
 #include <iostream>
+#include <mutex>
 #include <sstream>
 #include <iomanip>
 
+namespace {
+std::mutex mktime_mutex;
+} // namespace
+
 #if TIXML2_MAJOR_VERSION >= 6
 #define PRINTLINE(node) node->GetLineNum()
 #define PRINTLINEPLUSONE(node) node->GetLineNum() + 1
@@ -339,7 +345,10 @@ bool PermissionsParser::parse_validity(
 
                 if (!stream.fail())
                 {
-                    validity.not_before = std::mktime(&time);
+                    {
+                        std::lock_guard<std::mutex> lock(mktime_mutex);
+                        validity.not_before = std::mktime(&time);
+                    }
 #endif // if _MSC_VER != 1800
 
                 tinyxml2::XMLElement* old_node = node;
@@ -360,7 +369,10 @@ bool PermissionsParser::parse_validity(
 
                             if (!stream.fail())
                             {
-                                validity.not_after = std::mktime(&time);
+                                {
+                                    std::lock_guard<std::mutex> lock(mktime_mutex);
+                                    validity.not_after = std::mktime(&time);
+                                }
 #endif // if _MSC_VER != 1800
                             returned_value = true;
                         }

src/cpp/utils/SystemInfo.cpp

@@ -139,10 +139,12 @@ fastdds::dds::ReturnCode_t SystemInfo::get_username(
     return fastdds::dds::RETCODE_OK;
 #else
     uid_t user_id = geteuid();
-    struct passwd* pwd = getpwuid(user_id);
-    if (pwd != nullptr)
+    struct passwd pwd {};
+    struct passwd* result = nullptr;
+    char buf[1024];
+    if (getpwuid_r(user_id, &pwd, buf, sizeof(buf), &result) == 0 && result != nullptr)
     {
-        username = pwd->pw_name;
+        username = pwd.pw_name;
         if (!username.empty())
         {
             return fastdds::dds::RETCODE_OK;

test/blackbox/api/dds-pim/CustomPayloadPool.hpp

@@ -20,6 +20,7 @@
 #define DDS_CUSTOM_PAYLOAD_POOL_HPP
 
 #include <assert.h>
+#include <atomic>
 #include <cstdint>
 #include <stdio.h>
 #include <string.h>
@@ -95,8 +96,8 @@ class CustomPayloadPool : public eprosima::fastdds::rtps::IPayloadPool
         return true;
     }
 
-    uint32_t requested_payload_count = 0;
-    uint32_t returned_payload_count = 0;
+    std::atomic<uint32_t> requested_payload_count {0};
+    std::atomic<uint32_t> returned_payload_count {0};
 
 };
 

test/blackbox/api/dds-pim/PubSubReader.hpp

@@ -609,7 +609,7 @@ class PubSubReader
 
     std::list<type> data_not_received()
     {
-        std::unique_lock<std::mutex> lock(mutex_);
+        std::lock_guard<std::mutex> lock(mutex_);
         return total_msgs_;
     }
 
@@ -639,7 +639,7 @@ class PubSubReader
             size_t expected_samples)
     {
         {
-            std::unique_lock<std::mutex> lock(mutex_);
+            std::lock_guard<std::mutex> lock(mutex_);
             current_processed_count_ = 0;
             number_samples_expected_ = expected_samples;
             last_seq.clear();
@@ -1965,7 +1965,7 @@ class PubSubReader
         return status;
     }
 
-    const eprosima::fastdds::dds::LivelinessChangedStatus& liveliness_changed_status()
+    eprosima::fastdds::dds::LivelinessChangedStatus liveliness_changed_status()
     {
         std::unique_lock<std::mutex> lock(liveliness_mutex_);
 
@@ -2014,6 +2014,7 @@ class PubSubReader
 
     unsigned int get_participants_matched() const
     {
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         return participant_matched_;
     }
 
@@ -2138,7 +2139,7 @@ class PubSubReader
         {
             returnedValue = true;
 
-            std::unique_lock<std::mutex> lock(mutex_);
+            std::lock_guard<std::mutex> lock(mutex_);
 
             // Check order of changes.
             LastSeqInfo seq_info{ info.instance_handle, info.sample_identity.writer_guid() };
@@ -2182,7 +2183,7 @@ class PubSubReader
         }
 
         // Traverse the collection
-        std::unique_lock<std::mutex> lock(mutex_);
+        std::lock_guard<std::mutex> lock(mutex_);
         for (int32_t i = 0; i < datas.length(); ++i)
         {
             type& data = datas[i];
@@ -2248,28 +2249,28 @@ class PubSubReader
 
     void participant_matched()
     {
-        std::unique_lock<std::mutex> lock(mutexDiscovery_);
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         ++participant_matched_;
         cvDiscovery_.notify_one();
     }
 
     void participant_unmatched()
     {
-        std::unique_lock<std::mutex> lock(mutexDiscovery_);
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         --participant_matched_;
         cvDiscovery_.notify_one();
     }
 
     void matched()
     {
-        std::unique_lock<std::mutex> lock(mutexDiscovery_);
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         ++matched_;
         cvDiscovery_.notify_one();
     }
 
     void unmatched()
     {
-        std::unique_lock<std::mutex> lock(mutexDiscovery_);
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         --matched_;
         cvDiscovery_.notify_one();
     }
@@ -2314,7 +2315,7 @@ class PubSubReader
     std::list<type> total_msgs_;
     std::mutex mutex_;
     std::condition_variable cv_;
-    std::mutex mutexDiscovery_;
+    mutable std::mutex mutexDiscovery_;
     std::condition_variable cvDiscovery_;
     std::atomic<unsigned int> matched_;
     unsigned int participant_matched_;
@@ -2426,7 +2427,7 @@ class PubSubReaderWithWaitsets : public PubSubReader<TypeSupport>
             waitset_.attach_condition(reader_.subscriber_->get_statuscondition());
             waitset_.attach_condition(guard_condition_);
 
-            std::unique_lock<std::mutex> lock(mutex_);
+            std::lock_guard<std::mutex> lock(mutex_);
             if (nullptr == thread_)
             {
                 running_ = true;
@@ -2618,7 +2619,7 @@ class PubSubReaderWithWaitsets : public PubSubReader<TypeSupport>
         eprosima::fastdds::dds::GuardCondition guard_condition_;
 
         //! Number of times deadline was missed
-        unsigned int times_deadline_missed_ = 0;
+        std::atomic<unsigned int> times_deadline_missed_ {0};
 
         //! The timeout for the wait operation
         eprosima::fastdds::dds::Duration_t timeout_;
@@ -2643,6 +2644,7 @@ class PubSubReaderWithWaitsets : public PubSubReader<TypeSupport>
 
     ~PubSubReaderWithWaitsets() override
     {
+        waitset_thread_.stop();
     }
 
     void createSubscriber() override

test/blackbox/api/dds-pim/PubSubWriter.hpp

@@ -20,6 +20,7 @@
 #ifndef _TEST_BLACKBOX_PUBSUBWRITER_HPP_
 #define _TEST_BLACKBOX_PUBSUBWRITER_HPP_
 
+#include <atomic>
 #include <condition_variable>
 #include <list>
 #include <map>
@@ -1820,6 +1821,7 @@ class PubSubWriter
 
     unsigned int get_participants_matched() const
     {
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         return participant_matched_;
     }
 
@@ -1983,28 +1985,28 @@ class PubSubWriter
 
     void participant_matched()
     {
-        std::unique_lock<std::mutex> lock(mutexDiscovery_);
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         ++participant_matched_;
         cv_.notify_one();
     }
 
     void participant_unmatched()
     {
-        std::unique_lock<std::mutex> lock(mutexDiscovery_);
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         --participant_matched_;
         cv_.notify_one();
     }
 
     void matched()
     {
-        std::unique_lock<std::mutex> lock(mutexDiscovery_);
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         ++matched_;
         cv_.notify_one();
     }
 
     void unmatched()
     {
-        std::unique_lock<std::mutex> lock(mutexDiscovery_);
+        std::lock_guard<std::mutex> lock(mutexDiscovery_);
         --matched_;
         cv_.notify_one();
     }
@@ -2242,7 +2244,7 @@ class PubSubWriter
     eprosima::fastdds::rtps::GUID_t datawriter_guid_;
     bool initialized_;
     bool use_domain_id_from_profile_;
-    std::mutex mutexDiscovery_;
+    mutable std::mutex mutexDiscovery_;
     std::condition_variable cv_;
     std::atomic<unsigned int> matched_;
     unsigned int participant_matched_;
@@ -2322,7 +2324,7 @@ class PubSubWriterWithWaitsets : public PubSubWriter<TypeSupport>
             waitset_.attach_condition(writer_.datawriter_->get_statuscondition());
             waitset_.attach_condition(guard_condition_);
 
-            std::unique_lock<std::mutex> lock(mutex_);
+            std::lock_guard<std::mutex> lock(mutex_);
             if (nullptr == thread_)
             {
                 running_ = true;
@@ -2454,10 +2456,10 @@ class PubSubWriterWithWaitsets : public PubSubWriter<TypeSupport>
         eprosima::fastdds::dds::GuardCondition guard_condition_;
 
         //! The number of times deadline was missed
-        unsigned int times_deadline_missed_ = 0;
+        std::atomic<unsigned int> times_deadline_missed_ {0};
 
         //! The number of times liveliness was lost
-        unsigned int times_liveliness_lost_ = 0;
+        std::atomic<unsigned int> times_liveliness_lost_ {0};
 
         //! The timeout for the wait operation
         eprosima::fastdds::dds::Duration_t timeout_;

test/blackbox/common/BlackboxTestsTransportCustom.cpp

@@ -14,6 +14,7 @@
 
 #include "BlackboxTests.hpp"
 
+#include <atomic>
 #include <string>
 
 #include <gtest/gtest.h>
@@ -371,12 +372,12 @@ const std::string BuiltinTransportsTest::env_var_name_ = "FASTDDS_BUILTIN_TRANSP
 
 TEST(ChainingTransportTests, basic_test)
 {
-    bool writer_init_function_called = false;
-    bool writer_receive_function_called = false;
-    bool writer_send_function_called = false;
-    bool reader_init_function_called = false;
-    bool reader_receive_function_called = false;
-    bool reader_send_function_called = false;
+    std::atomic<bool> writer_init_function_called {false};
+    std::atomic<bool> writer_receive_function_called {false};
+    std::atomic<bool> writer_send_function_called {false};
+    std::atomic<bool> reader_init_function_called {false};
+    std::atomic<bool> reader_receive_function_called {false};
+    std::atomic<bool> reader_send_function_called {false};
     eprosima::fastdds::rtps::PropertyPolicy test_property_policy;
     test_property_policy.properties().push_back({test_property_name, test_property_value});
     std::shared_ptr<UDPv4TransportDescriptor> udp_transport = std::make_shared<UDPv4TransportDescriptor>();

test/blackbox/common/DDSBlackboxTestsDataReader.cpp

@@ -1421,6 +1421,9 @@ TEST(DDSDataReader, reception_timestamp_for_resent_samples)
     // Wait for the reception of all samples
     ASSERT_EQ(reader.block_for_all(std::chrono::seconds(5)), 3u);
 
+    // Avoid data race between the destructor and on_data_available
+    reader.destroy();
+
     // Check timestamps. reception_timestamps map is accesed by index of HelloWorld data
     ASSERT_EQ(reader.reception_timestamps.size(), 3u);
     auto reception_ts_1 = reader.reception_timestamps[1];