Multi-Threaded Executor Starvation

rclcpp/CHANGELOG.rst

@@ -2,6 +2,25 @@
 Changelog for package rclcpp
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+29.2.0 (2024-11-25)
+-------------------
+* accept custom allocator for LoanedMessage. (`#2672 <https://github.com/ros2/rclcpp/issues/2672>`_)
+* Contributors: Tomoya Fujita
+
+29.1.0 (2024-11-20)
+-------------------
+* a couple of typo fixes in doc section for LoanedMessage. (`#2676 <https://github.com/ros2/rclcpp/issues/2676>`_)
+* Make sure callback_end tracepoint is triggered in AnyServiceCallback (`#2670 <https://github.com/ros2/rclcpp/issues/2670>`_)
+* Correct the incorrect comments in generic_client.hpp (`#2662 <https://github.com/ros2/rclcpp/issues/2662>`_)
+* Fix NodeOptions assignment operator (`#2656 <https://github.com/ros2/rclcpp/issues/2656>`_)
+* set QoS History KEEP_ALL explicitly for statistics publisher. (`#2650 <https://github.com/ros2/rclcpp/issues/2650>`_)
+* Fix test_intra_process_manager.cpp with rmw_zenoh_cpp (`#2653 <https://github.com/ros2/rclcpp/issues/2653>`_)
+* Fixed test_events_executors in zenoh (`#2643 <https://github.com/ros2/rclcpp/issues/2643>`_)
+* rmw_fastrtps supports service event gid uniqueness test. (`#2638 <https://github.com/ros2/rclcpp/issues/2638>`_)
+* print warning if event callback is not supported instead of passing exception. (`#2648 <https://github.com/ros2/rclcpp/issues/2648>`_)
+* Implement callback support of async_send_request for service generic client (`#2614 <https://github.com/ros2/rclcpp/issues/2614>`_)
+* Contributors: Alejandro Hernández Cordero, Barry Xu, Chris Lalancette, Christophe Bedard, Romain DESILLE, Tomoya Fujita
+
 29.0.0 (2024-10-03)
 -------------------
 * Fixed test qos rmw zenoh (`#2639 <https://github.com/ros2/rclcpp/issues/2639>`_)

rclcpp/include/rclcpp/loaned_message.hpp

@@ -31,34 +31,33 @@ namespace rclcpp
 template<typename MessageT, typename AllocatorT = std::allocator<void>>
 class LoanedMessage
 {
+public:
   using MessageAllocatorTraits = rclcpp::allocator::AllocRebind<MessageT, AllocatorT>;
   using MessageAllocator = typename MessageAllocatorTraits::allocator_type;
 
-public:
   /// Constructor of the LoanedMessage class.
   /**
    * The constructor of this class allocates memory for a given message type
    * and associates this with a given publisher.
    *
-   * Given the publisher instance, a case differentiation is being performaned
-   * which decides whether the underlying middleware is able to allocate the appropriate
-   * memory for this message type or not.
-   * In the case that the middleware can not loan messages, the passed in allocator instance
-   * is being used to allocate the message within the scope of this class.
-   * Otherwise, the allocator is being ignored and the allocation is solely performaned
+   * The underlying middleware is queried to determine whether it is able to allocate the
+   * appropriate memory for this message type or not.
+   * In the case that the middleware cannot loan messages, the passed in allocator instance
+   * is used to allocate the message within the scope of this class.
+   * Otherwise, the allocator is ignored and the allocation is solely performed
    * in the underlying middleware with its appropriate allocation strategy.
    * The need for this arises as the user code can be written explicitly targeting a middleware
    * capable of loaning messages.
    * However, this user code is ought to be usable even when dynamically linked against
    * a middleware which doesn't support message loaning in which case the allocator will be used.
    *
    * \param[in] pub rclcpp::Publisher instance to which the memory belongs
-   * \param[in] allocator Allocator instance in case middleware can not allocate messages
+   * \param[in] allocator Allocator instance in case middleware cannot allocate messages
    * \throws anything rclcpp::exceptions::throw_from_rcl_error can throw.
    */
   LoanedMessage(
     const rclcpp::PublisherBase & pub,
-    std::allocator<MessageT> allocator)
+    MessageAllocator allocator)
   : pub_(pub),
     message_(nullptr),
     message_allocator_(std::move(allocator))

rclcpp/include/rclcpp/node.hpp

@@ -1430,7 +1430,7 @@ class Node : public std::enable_shared_from_this<Node>
   rclcpp::Clock::ConstSharedPtr
   get_clock() const;
 
-  /// Returns current time from the time source specified by clock_type.
+  /// Returns current time from the node clock.
   /**
    * \sa rclcpp::Clock::now
    */

rclcpp/package.xml

@@ -2,7 +2,7 @@
 <?xml-model href="http://download.ros.org/schema/package_format2.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
 <package format="2">
   <name>rclcpp</name>
-  <version>29.0.0</version>
+  <version>29.2.0</version>
   <description>The ROS client library in C++.</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp/src/rclcpp/executor.cpp

@@ -729,13 +729,33 @@ Executor::wait_for_work(std::chrono::nanoseconds timeout)
   // Clear any previous wait result
   this->wait_result_.reset();
 
+  // we need to make sure that callback groups don't get out of scope
+  // during the wait. As in jazzy, they are not covered by the DynamicStorage,
+  // we explicitly hold them here as a bugfix
+  std::vector<rclcpp::CallbackGroup::SharedPtr> cbgs;
+
   {
     std::lock_guard<std::mutex> guard(mutex_);
+
     if (this->entities_need_rebuild_.exchange(false) || current_collection_.empty()) {
       this->collect_entities();
     }
+
+    auto callback_groups = this->collector_.get_all_callback_groups();
+    cbgs.resize(callback_groups.size());
+    for(const auto & w_ptr : callback_groups) {
+      auto shr_ptr = w_ptr.lock();
+      if(shr_ptr) {
+        cbgs.push_back(std::move(shr_ptr));
+      }
+    }
   }
+
   this->wait_result_.emplace(wait_set_.wait(timeout));
+
+  // drop references to the callback groups, before trying to execute anything
+  cbgs.clear();
+
   if (!this->wait_result_ || this->wait_result_->kind() == WaitResultKind::Empty) {
     RCUTILS_LOG_WARN_NAMED(
       "rclcpp",

rclcpp/src/rclcpp/executors/static_single_threaded_executor.cpp

@@ -110,10 +110,29 @@ StaticSingleThreadedExecutor::spin_once_impl(std::chrono::nanoseconds timeout)
 std::optional<rclcpp::WaitResult<rclcpp::WaitSet>>
 StaticSingleThreadedExecutor::collect_and_wait(std::chrono::nanoseconds timeout)
 {
+  // we need to make sure that callback groups don't get out of scope
+  // during the wait. As in jazzy, they are not covered by the DynamicStorage,
+  // we explicitly hold them here as a bugfix
+  std::vector<rclcpp::CallbackGroup::SharedPtr> cbgs;
+
   if (this->entities_need_rebuild_.exchange(false) || current_collection_.empty()) {
     this->collect_entities();
   }
+
+  auto callback_groups = this->collector_.get_all_callback_groups();
+  cbgs.resize(callback_groups.size());
+  for(const auto & w_ptr : callback_groups) {
+    auto shr_ptr = w_ptr.lock();
+    if(shr_ptr) {
+      cbgs.push_back(std::move(shr_ptr));
+    }
+  }
+
   auto wait_result = wait_set_.wait(std::chrono::nanoseconds(timeout));
+
+  // drop references to the callback groups, before trying to execute anything
+  cbgs.clear();
+
   if (wait_result.kind() == WaitResultKind::Empty) {
     RCUTILS_LOG_WARN_NAMED(
       "rclcpp",

rclcpp/test/rclcpp/executors/test_executors.cpp

@@ -388,7 +388,7 @@ to_nanoseconds_helper(DurationT duration)
 //   - works nominally (it can execute entities)
 //   - it can execute multiple items at once
 //   - it does not wait for work to be available before returning
-TYPED_TEST(TestExecutors, spin_some)
+TYPED_TEST(TestExecutors, spinSome)
 {
   using ExecutorType = TypeParam;
 
@@ -484,7 +484,7 @@ TYPED_TEST(TestExecutors, spin_some)
 //   do not properly implement max_duration (it seems), so disable this test
 //   for them in the meantime.
 //   see: https://github.com/ros2/rclcpp/issues/2462
-TYPED_TEST(TestExecutorsStable, spin_some_max_duration)
+TYPED_TEST(TestExecutorsStable, spinSomeMaxDuration)
 {
   using ExecutorType = TypeParam;
 
@@ -714,6 +714,13 @@ TYPED_TEST(TestExecutors, notifyTwiceWhileSpinning)
       sub1_msg_count++;
     });
 
+  // Wait for the subscription to be matched
+  size_t tries = 10000;
+  while (this->publisher->get_subscription_count() < 2 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(this->publisher->get_subscription_count(), 2);
+
   // Publish a message and verify it's received
   this->publisher->publish(test_msgs::msg::Empty());
   auto start = std::chrono::steady_clock::now();
@@ -731,6 +738,13 @@ TYPED_TEST(TestExecutors, notifyTwiceWhileSpinning)
       sub2_msg_count++;
     });
 
+  // Wait for the subscription to be matched
+  tries = 10000;
+  while (this->publisher->get_subscription_count() < 3 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(this->publisher->get_subscription_count(), 3);
+
   // Publish a message and verify it's received by both subscriptions
   this->publisher->publish(test_msgs::msg::Empty());
   start = std::chrono::steady_clock::now();
@@ -820,7 +834,7 @@ TEST(TestExecutors, testSpinWithNonDefaultContext)
   rclcpp::shutdown(non_default_context);
 }
 
-TYPED_TEST(TestExecutors, release_ownership_entity_after_spinning_cancel)
+TYPED_TEST(TestExecutors, releaseOwnershipEntityAfterSpinningCancel)
 {
   using ExecutorType = TypeParam;
   ExecutorType executor;
@@ -843,3 +857,82 @@ TYPED_TEST(TestExecutors, release_ownership_entity_after_spinning_cancel)
 
   EXPECT_EQ(server.use_count(), 1);
 }
+
+TYPED_TEST(TestExecutors, testRaceDropCallbackGroupFromSecondThread)
+{
+  using ExecutorType = TypeParam;
+<<<<<<< HEAD
+<<<<<<< HEAD
+
+  // Create an executor
+  ExecutorType executor;
+  executor.add_node(this->node);
+
+  // Start spinning
+  auto executor_thread = std::thread(
+    [&executor]() {
+      executor.spin();
+=======
+  // rmw_connextdds doesn't support events-executor
+  if (
+    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
+    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
+  {
+    GTEST_SKIP();
+  }
+=======
+>>>>>>> d7245365 (Re-enable executor test on rmw_connextdds. (#2693))
+
+  // Create an executor
+  ExecutorType executor;
+  executor.add_node(this->node);
+
+  // Start spinning
+  auto executor_thread = std::thread(
+<<<<<<< HEAD
+    [executor]() {
+      executor->spin();
+>>>>>>> e9b10042 (fix(Executor): Fix segfault if callback group is deleted during rmw_wait (#2683))
+=======
+    [&executor]() {
+      executor.spin();
+>>>>>>> 3310f9ea (Fix warnings on Windows. (#2692))
+    });
+
+  // As the problem is a race, we do this multiple times,
+  // to raise our chances of hitting the problem
+<<<<<<< HEAD
+<<<<<<< HEAD
+  for (size_t i = 0; i < 10; i++) {
+    auto cg = this->node->create_callback_group(
+      rclcpp::CallbackGroupType::MutuallyExclusive);
+=======
+  for(size_t i = 0; i < 10; i++) {
+    auto cg = this->node->create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive);
+>>>>>>> e9b10042 (fix(Executor): Fix segfault if callback group is deleted during rmw_wait (#2683))
+=======
+  for (size_t i = 0; i < 10; i++) {
+    auto cg = this->node->create_callback_group(
+      rclcpp::CallbackGroupType::MutuallyExclusive);
+>>>>>>> 3310f9ea (Fix warnings on Windows. (#2692))
+    auto timer = this->node->create_timer(1s, [] {}, cg);
+    // sleep a bit, so that the spin thread can pick up the callback group
+    // and add it to the executor
+    std::this_thread::sleep_for(5ms);
+
+    // At this point the callbackgroup should be used within the waitset of the executor
+    // as we leave the scope, the reference to cg will be dropped.
+    // If the executor has a race, we will experience a segfault at this point.
+  }
+
+<<<<<<< HEAD
+<<<<<<< HEAD
+  executor.cancel();
+=======
+  executor->cancel();
+>>>>>>> e9b10042 (fix(Executor): Fix segfault if callback group is deleted during rmw_wait (#2683))
+=======
+  executor.cancel();
+>>>>>>> 3310f9ea (Fix warnings on Windows. (#2692))
+  executor_thread.join();
+}