fixes

fix: some data races
and more

Author: JuanCarlosgg

CMakeLists.txt

@@ -6,6 +6,19 @@ project(dsr
 
 include(GNUInstallDirs)
 
+# -DTSAN=ON  enables ThreadSanitizer across the whole build (library + tests/benchmarks).
+# Must be applied before any add_subdirectory so every TU is instrumented.
+# Incompatible with SANITIZER (ASan/UBSan) — enforce that here.
+option(TSAN "Enable ThreadSanitizer" OFF)
+if (TSAN)
+    if (SANITIZER)
+        message(FATAL_ERROR "TSAN and SANITIZER (ASan+UBSan) are mutually exclusive")
+    endif()
+    message(STATUS "ThreadSanitizer enabled")
+    add_compile_options(-fsanitize=thread -fno-omit-frame-pointer)
+    add_link_options(-fsanitize=thread)
+endif()
+
 add_definitions(-I/usr/include/x86_64-linux-gnu/qt6/QtOpenGLWidgets/)
 
 include_directories(/home/robocomp/robocomp/classes)

api/dsr_api.cpp

@@ -1093,11 +1093,20 @@ void DSRGraph::join_delta_node(IDL::MvregNode &&mvreg)
         };
 
         std::optional<std::unordered_set<std::pair<uint64_t, std::string>,hash_tuple>> cache_map_to_edges = {};
+        // Snapshot the data needed for signal emission while the lock is held.
+        // nodes.at(id) must NOT be accessed after the lock is released: a concurrent
+        // insert_node_/update_node call on the same id runs nodes[id].write() which
+        // calls dk.rmv() (clears dk.ds) followed by dk.add(), leaving a window where
+        // read_reg()'s assert(dk.ds.size() >= 1) would fire.
+        std::string node_type_snapshot;
+        std::vector<std::pair<uint64_t, std::string>> from_edges_snapshot;
         {
             std::unique_lock<std::shared_mutex> lock(_mutex);
             if (!deleted.contains(id)) {
                 joined = true;
-                maybe_deleted_node = (nodes[id].empty()) ? std::nullopt : std::make_optional(nodes.at(id).read_reg());
+                if (auto it = nodes.find(id); it != nodes.end() && !it->second.empty()) {
+                    maybe_deleted_node = it->second.read_reg();
+                }
                 nodes[id].join(std::move(crdt_delta));
                 if (nodes.at(id).empty() or d_empty) {
                     nodes.erase(id);
@@ -1106,8 +1115,14 @@ void DSRGraph::join_delta_node(IDL::MvregNode &&mvreg)
                     delete_unprocessed_deltas();
                 } else {
                     signal = true;
-                    update_maps_node_insert(id, nodes.at(id).read_reg());
+                    const auto& reg = nodes.at(id).read_reg();
+                    update_maps_node_insert(id, reg);
                     consume_unprocessed_deltas();
+                    // Snapshot type and outgoing edges before the lock is released.
+                    node_type_snapshot = reg.type();
+                    for (const auto &[k, v] : reg.fano()) {
+                        from_edges_snapshot.emplace_back(k.first, k.second);
+                    }
                 }
             } else {
                 delete_unprocessed_deltas();
@@ -1116,11 +1131,11 @@ void DSRGraph::join_delta_node(IDL::MvregNode &&mvreg)
 
         if (joined) {
             if (signal) {
-                DSR_LOG_DEBUG("[JOIN_NODE] node inserted/updated:", id, nodes.at(id).read_reg().type());
-                emitter.update_node_signal(id, nodes.at(id).read_reg().type(), SignalInfo{ mvreg.agent_id() });
-                for (const auto &[k, v] : nodes.at(id).read_reg().fano()) {
-                    DSR_LOG_DEBUG("[JOIN_NODE] add edge FROM:", id, k.first, k.second);
-                    emitter.update_edge_signal(id, k.first, k.second, SignalInfo{ mvreg.agent_id() });
+                DSR_LOG_DEBUG("[JOIN_NODE] node inserted/updated:", id, node_type_snapshot);
+                emitter.update_node_signal(id, node_type_snapshot, SignalInfo{ mvreg.agent_id() });
+                for (const auto &[to_id, edge_type] : from_edges_snapshot) {
+                    DSR_LOG_DEBUG("[JOIN_NODE] add edge FROM:", id, to_id, edge_type);
+                    emitter.update_edge_signal(id, to_id, edge_type, SignalInfo{ mvreg.agent_id() });
                 }
 
                 for (const auto &[k, v]: map_new_to_edges)
@@ -1443,7 +1458,10 @@ std::optional<std::string> DSRGraph::join_delta_edge_attr(IDL::MvregEdgeAttr &&m
 
 void DSRGraph::join_full_graph(IDL::OrMap &&full_graph)
 {
-    std::vector<std::tuple<bool, uint64_t, std::string, std::optional<CRDTNode>>> updates;
+    // 5th element: post-join node snapshot captured inside the lock, used for
+    // signal emission after the lock is released to avoid racing with
+    // insert_node_/update_node (same pattern as join_delta_node).
+    std::vector<std::tuple<bool, uint64_t, std::string, std::optional<CRDTNode>, std::optional<CRDTNode>>> updates;
 
     uint64_t id{0}, timestamp{0};
     uint32_t agent_id_ch{0};
@@ -1541,24 +1559,26 @@ void DSRGraph::join_full_graph(IDL::OrMap &&full_graph)
                 it->second.join(std::move(mv));
                 if (mv_empty or it->second.empty()) {
                     update_maps_node_delete(k, nd);
-                    updates.emplace_back(false, k, "", std::nullopt);
+                    updates.emplace_back(false, k, "", std::nullopt, std::nullopt);
                     delete_unprocessed_deltas();
                 } else {
-                    update_maps_node_insert(k, it->second.read_reg());
-                    updates.emplace_back(true, k, it->second.read_reg().type(), nd);
+                    const auto& reg = it->second.read_reg();
+                    update_maps_node_insert(k, reg);
+                    updates.emplace_back(true, k, reg.type(), nd, reg);
                     consume_unprocessed_deltas();
                 }
             }
         }
 
     }
-    for (auto &[signal, id, type, nd] : updates)
+    for (auto &[signal, id, type, nd, current_nd] : updates)
         if (signal) {
-            //check what change is joined
-            if (!nd.has_value() || nd->attrs() != nodes[id].read_reg().attrs()) {
-                emitter.update_node_signal(id, nodes[id].read_reg().type(), SignalInfo{ agent_id_ch });
-            } else if (nd.value() != nodes[id].read_reg()) {
-                auto iter = nodes[id].read_reg().fano();
+            //check what change is joined — use the snapshot captured inside the lock,
+            //not nodes[id], which races with concurrent insert_node_/update_node calls.
+            if (!nd.has_value() || nd->attrs() != current_nd->attrs()) {
+                emitter.update_node_signal(id, type, SignalInfo{ agent_id_ch });
+            } else if (nd.value() != *current_nd) {
+                const auto& iter = current_nd->fano();
                 for (const auto &[k, v] : nd->fano()) {
                     if (!iter.contains(k)) {
                         emitter.del_edge_signal(id, k.first, k.second, SignalInfo{ agent_id_ch });
@@ -1908,8 +1928,8 @@ void DSRGraph::fullgraph_server_thread()
 
 std::pair<bool, bool> DSRGraph::fullgraph_request_thread()
 {
-    bool sync = false;
-    bool repeated = false;
+    std::atomic<bool> sync{false};
+    std::atomic<bool> repeated{false};
     auto lambda_request_answer = [&](eprosima::fastdds::dds::DataReader *reader, DSR::DSRGraph *graph)
     {
         while (true)

api/include/dsr/api/dsr_api.h

@@ -585,7 +585,6 @@ namespace DSR
         const bool copy;
         std::unique_ptr<Utilities> utils;
         std::unordered_set<std::string_view> ignored_attributes;
-        ThreadPool tp, tp_delta_attr;
         bool same_host;
         id_generator generator;
         GraphSettings::LOGLEVEL log_level;
@@ -678,6 +677,11 @@ namespace DSR
         std::unordered_multimap<uint64_t, std::tuple<uint64_t, std::string, mvreg<DSR::CRDTEdge>, uint64_t>> unprocessed_delta_edge_to;
         std::unordered_multimap<std::tuple<uint64_t, uint64_t, std::string>, std::tuple<std::string, mvreg<DSR::CRDTAttribute>, uint64_t>, hash_tuple> unprocessed_delta_edge_att;
 
+        // ThreadPools are declared after all data they access so that their
+        // destructors (which join worker threads) run before the data members
+        // are destroyed, preventing use-after-free data races on shutdown.
+        ThreadPool tp, tp_delta_attr;
+
         //Custom function for each rtps topic
         class NewMessageFunctor {
         public:

benchmarks/CMakeLists.txt

@@ -79,6 +79,16 @@ set_target_properties(dsr_benchmarks PROPERTIES
 
 target_compile_options(dsr_benchmarks PUBLIC -g -std=c++23)
 
+# -DTSAN=ON  enables ThreadSanitizer.  Requires the dsr_api/dsr_core libraries
+# to also be built with TSAN=ON (via the root CMakeLists.txt), otherwise TSan
+# will report false positives from uninstrumented library code.
+option(TSAN "Enable ThreadSanitizer" OFF)
+if (TSAN)
+    message(STATUS "ThreadSanitizer enabled for benchmarks")
+    target_compile_options(dsr_benchmarks PRIVATE -fsanitize=thread -fno-omit-frame-pointer)
+    target_link_options(dsr_benchmarks PRIVATE -fsanitize=thread)
+endif()
+
 # Include directories
 target_include_directories(dsr_benchmarks PRIVATE
     ${CMAKE_CURRENT_SOURCE_DIR}
@@ -107,6 +117,28 @@ add_custom_command(TARGET dsr_benchmarks POST_BUILD
     COMMENT "Creating results directory"
 )
 
+# Flamegraph target — generates one SVG per benchmark test case via perf.
+# Requires: perf, and FlameGraph scripts (flamegraph.pl + stackcollapse-perf.pl).
+# Set FG_DIR to the FlameGraph checkout if the scripts aren't on PATH, e.g.:
+#   cmake --build . --target flamegraph -j1 -- FG_DIR=/opt/FlameGraph
+# Or pass a Catch2 filter to profile a subset:
+#   cmake --build . --target flamegraph -j1 -- BENCH_FILTER=[LATENCY]
+set(FLAMEGRAPH_SCRIPT ${CMAKE_CURRENT_SOURCE_DIR}/flamegraph.sh)
+set(FLAMEGRAPH_OUTDIR ${CMAKE_CURRENT_BINARY_DIR}/results/flamegraphs)
+add_custom_target(flamegraph
+    COMMAND ${CMAKE_COMMAND} -E make_directory ${FLAMEGRAPH_OUTDIR}
+    COMMAND env
+        FG_DIR=$ENV{FG_DIR}
+        ${FLAMEGRAPH_SCRIPT}
+            -b $<TARGET_FILE:dsr_benchmarks>
+            -o ${FLAMEGRAPH_OUTDIR}
+            $ENV{BENCH_FILTER}
+    DEPENDS dsr_benchmarks
+    WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}
+    COMMENT "Generating per-benchmark flamegraphs in ${FLAMEGRAPH_OUTDIR}"
+    USES_TERMINAL
+)
+
 # Register tests with CTest (optional)
 # Disabled auto-discovery as it requires running the binary at build time
 # which may fail if libraries are not in LD_LIBRARY_PATH

benchmarks/consistency/conflict_rate_bench.cpp

@@ -215,9 +215,9 @@ TEST_CASE("Conflict rate benchmarks", "[CONSISTENCY][conflict][.multi]") {
         for (int round = 0; round < NUM_ROUNDS; ++round) {
             // Both agents try to create the same edge simultaneously
             auto edge_a = GraphGenerator::create_test_edge(
-                node1_id, node2_id, agent_a->get_agent_id(), "conflict_edge");
+                node1_id, node2_id, agent_a->get_agent_id(), "test_edge");
             auto edge_b = GraphGenerator::create_test_edge(
-                node1_id, node2_id, agent_b->get_agent_id(), "conflict_edge");
+                node1_id, node2_id, agent_b->get_agent_id(), "test_edge");
 
             std::thread ta([&]() { agent_a->insert_or_assign_edge(edge_a); });
             std::thread tb([&]() { agent_b->insert_or_assign_edge(edge_b); });
@@ -228,15 +228,15 @@ TEST_CASE("Conflict rate benchmarks", "[CONSISTENCY][conflict][.multi]") {
             fixture.wait_for_sync(std::chrono::milliseconds(200));
 
             // Check both agents see the edge
-            auto edge_on_a = agent_a->get_edge(node1_id, node2_id, "conflict_edge");
-            auto edge_on_b = agent_b->get_edge(node1_id, node2_id, "conflict_edge");
+            auto edge_on_a = agent_a->get_edge(node1_id, node2_id, "test_edge");
+            auto edge_on_b = agent_b->get_edge(node1_id, node2_id, "test_edge");
 
             if (!edge_on_a.has_value() || !edge_on_b.has_value()) {
                 conflicts++;
             }
 
             // Delete edge for next round
-            agent_a->delete_edge(node1_id, node2_id, "conflict_edge");
+            agent_a->delete_edge(node1_id, node2_id, "test_edge");
             fixture.wait_for_sync(std::chrono::milliseconds(100));
         }
 

benchmarks/fixtures/graph_generator.h

@@ -191,7 +191,8 @@ class GraphGenerator {
 private:
     std::string temp_filename() {
         static std::atomic<uint64_t> next_id{0};
-        return "/tmp/dsr_bench_" + std::to_string(next_id.fetch_add(1, std::memory_order_relaxed)) + ".json";
+        return "/tmp/dsr_bench_" + std::to_string(getpid()) + "_"
+               + std::to_string(next_id.fetch_add(1, std::memory_order_relaxed)) + ".json";
     }
 
     std::vector<uint64_t> generate_node_ids(uint32_t count) {

benchmarks/fixtures/multi_agent_fixture.h

@@ -65,6 +65,10 @@ class MultiAgentFixture {
             qWarning("Can't create agents");
             return false;
         }
+        if (config_file.empty()) {
+            qWarning("create_agents: config_file is empty — graph generator likely failed to write to /tmp (check permissions)");
+            return false;
+        }
 
         // Keep agent IDs deterministic while remaining disjoint across fixture
         // instances in the same process.