Fix deadlock in TimeSource::destroy_clock_sub (#2962)

rclcpp/src/rclcpp/time_source.cpp

@@ -244,6 +244,12 @@ class TimeSource::NodeState final
     rclcpp::node_interfaces::NodeClockInterface::SharedPtr node_clock_interface,
     rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters_interface)
   {
+    {
+      // A previous detachNode() on this NodeState set detaching_; clear
+      // it so a re-attach can create the clock subscription again.
+      std::lock_guard<std::mutex> guard(clock_sub_lock_);
+      detaching_ = false;
+    }
     std::lock_guard<std::mutex> guard(node_base_lock_);
     node_base_ = std::move(node_base_interface);
     node_topics_ = std::move(node_topics_interface);
@@ -291,6 +297,16 @@ class TimeSource::NodeState final
   // Detach the attached node
   void detachNode()
   {
+    {
+      // Block create_clock_sub() before tearing anything down: without
+      // this, a concurrent post_set_parameters(true) could install a
+      // fresh subscription and clock executor thread after
+      // destroy_clock_sub() returned empty but before the node
+      // interfaces below are cleared, leaving the new thread running
+      // against null interfaces.
+      std::lock_guard<std::mutex> guard(clock_sub_lock_);
+      detaching_ = true;
+    }
     // destroy_clock_sub() *must* be first here, to ensure that the executor
     // can't possibly call any of the callbacks as we are cleaning up.
     destroy_clock_sub();
@@ -350,6 +366,13 @@ class TimeSource::NodeState final
   using SubscriptionT = rclcpp::Subscription<rosgraph_msgs::msg::Clock>;
   std::shared_ptr<SubscriptionT> clock_subscription_{nullptr};
   std::mutex clock_sub_lock_;
+  // Set while detachNode() is tearing the clock subscription down.
+  // Guarded by clock_sub_lock_: create_clock_sub() refuses to start a
+  // new clock executor thread while this is set, so no thread can be
+  // created in the window between destroy_clock_sub() returning and
+  // detachNode() clearing the node interfaces (ros2/rclcpp#3412 review
+  // feedback on #2962).
+  bool detaching_{false};
   rclcpp::CallbackGroup::SharedPtr clock_callback_group_;
   rclcpp::executors::SingleThreadedExecutor::SharedPtr clock_executor_;
 
@@ -371,6 +394,11 @@ class TimeSource::NodeState final
   void create_clock_sub()
   {
     std::lock_guard<std::mutex> guard(clock_sub_lock_);
+    if (detaching_) {
+      // detachNode() is tearing down; do not start a new clock executor
+      // thread that would outlive the node interfaces.
+      return;
+    }
     if (clock_subscription_) {
       // Subscription already created.
       return;
@@ -396,10 +424,16 @@ class TimeSource::NodeState final
       clock_executor_ =
         std::make_shared<rclcpp::executors::SingleThreadedExecutor>(exec_options);
       if (!clock_executor_thread_.joinable()) {
+        // Capture the executor and callback group by value: the thread
+        // must not read the clock_executor_/clock_callback_group_ members,
+        // because destroy_clock_sub() moves them out to join the thread
+        // without holding clock_sub_lock_ (see ros2/rclcpp#2962). Reading
+        // the members here would race that move and dereference a null
+        // executor if the thread starts after the move.
         clock_executor_thread_ = std::thread(
-          [this]() {
-            clock_executor_->add_callback_group(clock_callback_group_, node_base_);
-            clock_executor_->spin();
+          [this, executor = clock_executor_, group = clock_callback_group_]() {
+            executor->add_callback_group(group, node_base_);
+            executor->spin();
           }
         );
       }
@@ -429,13 +463,59 @@ class TimeSource::NodeState final
   // Destroy the subscription for the clock topic
   void destroy_clock_sub()
   {
-    std::lock_guard<std::mutex> guard(clock_sub_lock_);
-    if (clock_executor_thread_.joinable()) {
-      clock_executor_->cancel();
-      clock_executor_thread_.join();
-      clock_executor_->remove_callback_group(clock_callback_group_);
+    // Tear down the clock subscription and its executor thread without
+    // holding clock_sub_lock_ across the thread join (see ros2/rclcpp#2962).
+    //
+    // The previous implementation held clock_sub_lock_ while joining
+    // clock_executor_thread_. The executor thread's shutdown path can
+    // contend on the same lock, deadlocking the join. Moving the thread,
+    // executor, callback group and subscription into locals under the lock
+    // and joining outside of it breaks the cycle.
+    //
+    // The subscription is moved (not reset) under the lock, so a concurrent
+    // create_clock_sub() sees no subscription and builds a fresh one instead
+    // of returning early on a stale subscription that this function would
+    // then reset, which would leave sim time enabled with no subscription.
+    // The old subscription is kept alive in a local until after the join,
+    // so the executor thread can finish any in-flight callback without
+    // touching a destroyed subscription.
+    //
+    // The teardown is repeated if a concurrent create_clock_sub() installed
+    // a new subscription/thread while the old thread was being joined (only
+    // possible when post_set_parameters(false) races
+    // post_set_parameters(true); the detachNode() race is closed by the
+    // detaching_ flag, which blocks create_clock_sub() for the whole
+    // teardown). destroy_clock_sub() must never return with a clock executor
+    // thread still spinning behind it: detachNode() resets the node
+    // interfaces right after, and the executor thread's lambda captures
+    // `this`.
+    for (;; ) {
+      std::thread thread_to_join;
+      rclcpp::executors::SingleThreadedExecutor::SharedPtr executor_to_stop;
+      rclcpp::CallbackGroup::SharedPtr callback_group_to_remove;
+      std::shared_ptr<SubscriptionT> subscription_to_keep_alive;
+      {
+        std::lock_guard<std::mutex> guard(clock_sub_lock_);
+        if (!clock_executor_thread_.joinable() && !clock_subscription_) {
+          return;
+        }
+        thread_to_join = std::move(clock_executor_thread_);
+        executor_to_stop = std::move(clock_executor_);
+        callback_group_to_remove = std::move(clock_callback_group_);
+        subscription_to_keep_alive = std::move(clock_subscription_);
+      }
+
+      if (executor_to_stop) {
+        executor_to_stop->cancel();
+      }
+      if (thread_to_join.joinable()) {
+        thread_to_join.join();
+      }
+      if (executor_to_stop && callback_group_to_remove) {
+        executor_to_stop->remove_callback_group(callback_group_to_remove);
+      }
+      // subscription_to_keep_alive is destroyed here, after the join.
     }
-    clock_subscription_.reset();
   }
 
   // On set Parameters callback handle

rclcpp/test/rclcpp/CMakeLists.txt

@@ -437,6 +437,12 @@ if(TARGET test_time_source)
   target_link_libraries(test_time_source ${PROJECT_NAME} rcl::rcl)
 endif()
 
+ament_add_ros_isolated_gtest(test_time_source_deadlock test_time_source_deadlock.cpp
+  APPEND_LIBRARY_DIRS "${append_library_dirs}")
+if(TARGET test_time_source_deadlock)
+  target_link_libraries(test_time_source_deadlock ${PROJECT_NAME})
+endif()
+
 ament_add_ros_isolated_gtest(test_utilities test_utilities.cpp
   APPEND_LIBRARY_DIRS "${append_library_dirs}")
 ament_add_test_label(test_utilities mimick)

rclcpp/test/rclcpp/test_time_source_deadlock.cpp

@@ -0,0 +1,225 @@
+// Copyright 2026 Open Source Robotics Foundation, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/**
+ * Stress tests for the TimeSource::destroy_clock_sub deadlock fix.
+ *
+ * Exercises the conditions from ros2/rclcpp#2962: rapid node
+ * construction/destruction with use_sim_time enabled. If the
+ * deadlock is present, these tests hang (caught by the test timeout).
+ */
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <memory>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "gtest/gtest.h"
+
+#include "rclcpp/rclcpp.hpp"
+#include "rclcpp/time_source.hpp"
+#include "rosgraph_msgs/msg/clock.hpp"
+
+using namespace std::chrono_literals;
+
+class TestTimeSourceDeadlock : public ::testing::Test
+{
+public:
+  static void SetUpTestCase()
+  {
+    rclcpp::init(0, nullptr);
+  }
+
+  static void TearDownTestCase()
+  {
+    rclcpp::shutdown();
+  }
+};
+
+/// Reproducer for ros2/rclcpp#2962: rapid construction and destruction
+/// of a node with use_sim_time=true. The clock executor thread must
+/// be joined without deadlock.
+TEST_F(TestTimeSourceDeadlock, rapid_construct_destroy_sim_time)
+{
+  for (int i = 0; i < 50; ++i) {
+    rclcpp::NodeOptions opts;
+    opts.parameter_overrides({rclcpp::Parameter("use_sim_time", true)});
+    auto node = std::make_shared<rclcpp::Node>(
+      "deadlock_test_" + std::to_string(i), "/test_ns", opts);
+
+    // Immediately destroy -- the clock executor thread must be
+    // joined without deadlock.
+    node.reset();
+  }
+}
+
+/// Create and destroy nodes with use_sim_time from multiple threads
+/// to stress the lock/join interaction.
+TEST_F(TestTimeSourceDeadlock, concurrent_construct_destroy)
+{
+  const int num_threads = 4;
+  const int iterations = 10;
+  std::vector<std::future<void>> futures;
+
+  for (int t = 0; t < num_threads; ++t) {
+    futures.push_back(
+      std::async(
+        std::launch::async, [t, iterations]() {
+          for (int i = 0; i < iterations; ++i) {
+            rclcpp::NodeOptions opts;
+            opts.parameter_overrides({rclcpp::Parameter("use_sim_time", true)});
+            auto node = std::make_shared<rclcpp::Node>(
+              "concurrent_test_" + std::to_string(t) + "_" + std::to_string(i),
+              "/test_ns", opts);
+            node.reset();
+          }
+        }));
+  }
+
+  for (auto & f : futures) {
+    ASSERT_NO_THROW(f.get());
+  }
+}
+
+/// Toggle use_sim_time parameter while spinning -- exercises the
+/// create_clock_sub / destroy_clock_sub interplay.
+TEST_F(TestTimeSourceDeadlock, toggle_use_sim_time_while_spinning)
+{
+  auto node = std::make_shared<rclcpp::Node>("toggle_test", "/test_ns");
+  rclcpp::executors::SingleThreadedExecutor executor;
+  executor.add_node(node);
+
+  // Spin in background
+  auto spin_future = std::async(
+    std::launch::async, [&executor]() {
+      executor.spin();
+    });
+
+  // Toggle use_sim_time several times
+  for (int i = 0; i < 10; ++i) {
+    node->set_parameter(rclcpp::Parameter("use_sim_time", true));
+    std::this_thread::sleep_for(10ms);
+    node->set_parameter(rclcpp::Parameter("use_sim_time", false));
+    std::this_thread::sleep_for(10ms);
+  }
+
+  executor.cancel();
+  spin_future.get();
+}
+
+/// Verify the clock thread is properly cleaned up after destruction.
+/// After destroying a sim-time node, no threads from its clock
+/// executor should be lingering.
+TEST_F(TestTimeSourceDeadlock, no_lingering_thread_after_destroy)
+{
+  {
+    rclcpp::NodeOptions opts;
+    opts.parameter_overrides({rclcpp::Parameter("use_sim_time", true)});
+    auto node = std::make_shared<rclcpp::Node>(
+      "linger_test", "/test_ns", opts);
+    // Node destroyed here -- clock thread must be fully joined
+  }
+
+  // If the thread were dangling, subsequent operations on the
+  // same context would be unreliable. Create another node to
+  // verify the context is clean.
+  auto node2 = std::make_shared<rclcpp::Node>("after_linger", "/test_ns");
+  ASSERT_NE(node2, nullptr);
+}
+
+/// Rapid construct/destroy while /clock traffic is flowing, so the
+/// clock executor thread is actively dispatching callbacks while
+/// destroy_clock_sub() runs -- the exact window from ros2/rclcpp#2962.
+TEST_F(TestTimeSourceDeadlock, rapid_construct_destroy_with_clock_traffic)
+{
+  auto clock_pub_node = std::make_shared<rclcpp::Node>("clock_pub", "/test_ns");
+  auto clock_pub = clock_pub_node->create_publisher<rosgraph_msgs::msg::Clock>("/clock", 10);
+
+  std::atomic<bool> stop{false};
+  auto pub_thread = std::thread(
+    [&]() {
+      rclcpp::executors::SingleThreadedExecutor pub_executor;
+      pub_executor.add_node(clock_pub_node);
+      auto timer = clock_pub_node->create_wall_timer(
+        5ms, [&]() {
+          rosgraph_msgs::msg::Clock msg;
+          msg.clock.sec = 1;
+          clock_pub->publish(msg);
+        });
+      while (!stop.load()) {
+        pub_executor.spin_some();
+        std::this_thread::sleep_for(1ms);
+      }
+    });
+
+  for (int i = 0; i < 50; ++i) {
+    rclcpp::NodeOptions opts;
+    opts.parameter_overrides({rclcpp::Parameter("use_sim_time", true)});
+    auto node = std::make_shared<rclcpp::Node>(
+      "traffic_test_" + std::to_string(i), "/test_ns", opts);
+    node.reset();
+  }
+
+  stop.store(true);
+  pub_thread.join();
+}
+
+/// Regression test for the detach/create race flagged in review of
+/// ros2/rclcpp#3412: a post_set_parameters(true) landing between
+/// destroy_clock_sub() returning and detachNode() clearing the node
+/// interfaces could start a fresh clock executor thread against null
+/// interfaces. Hammer detachNode() while another thread toggles
+/// use_sim_time: the detach must never crash or hang, and the
+/// TimeSource must still be usable (re-attachable) afterwards.
+/// Without the detaching_ flag this window is open; with it,
+/// create_clock_sub() no-ops for the whole teardown.
+TEST_F(TestTimeSourceDeadlock, detach_races_sim_time_enable)
+{
+  for (int iter = 0; iter < 20; ++iter) {
+    auto node = std::make_shared<rclcpp::Node>(
+      "detach_race_" + std::to_string(iter), "/test_ns");
+    rclcpp::TimeSource time_source(rclcpp::QoS(10), true);
+    time_source.attachNode(node);
+
+    std::atomic<bool> stop{false};
+    auto toggler = std::thread(
+      [&]() {
+        bool enable = true;
+        while (!stop.load()) {
+          try {
+            node->set_parameter(rclcpp::Parameter("use_sim_time", enable));
+          } catch (...) {
+            // detachNode() may unregister the parameter callbacks
+            // mid-call; that is expected here.
+          }
+          enable = !enable;
+        }
+      });
+
+    // Detach while the toggler hammers use_sim_time: no clock executor
+    // thread may be created past this point.
+    ASSERT_NO_THROW(time_source.detachNode());
+    stop.store(true);
+    toggler.join();
+
+    // Re-attaching after a detach must still work (the detaching flag
+    // is cleared by attachNode) and a clean detach must follow.
+    ASSERT_NO_THROW(time_source.attachNode(node));
+    ASSERT_NO_THROW(node->set_parameter(rclcpp::Parameter("use_sim_time", true)));
+    ASSERT_NO_THROW(time_source.detachNode());
+  }
+}