Slow timed callback monopolizes all processing in the multi threaded executor

I have the following situation: a node has multiple subscriptions (input data) and a timer callback (processing). While all subscriptions are rather fast (just copy the input data) the processing callback may be slow.

I have observed that when the node is run by the multithreaded executor, for as long as the processing callback is slower than the timer period, none of the subscription callback is ever executed. In other words, timer callbacks seems to have priority over subscription callbacks. This is not the case with the single threaded executor.

In the documentation, it is mentionned such prioritization existed in the past, but that it has been removed since then (https://docs.ros.org/en/humble/Concepts/Intermediate/About-Executors.html?highlight=executor#scheduling-semantics)

Bug report

Required Info:

• Operating System:
  • Ubuntu 22.04.3
• Installation type:
  • Binary
• Version or commit hash:
  • humble
• DDS implementation:
  • CycloneDDS
• Client library (if applicable):
  • rclcpp

Steps to reproduce issue

This behavior can be reproduced with the component example from https://github.com/ros2/demos

I have made the following modifications:

diff --git a/composition/include/composition/listener_component.hpp b/composition/include/composition/listener_component.hpp
index 3e56c8f..7e5aa1c 100644
--- a/composition/include/composition/listener_component.hpp
+++ b/composition/include/composition/listener_component.hpp
@@ -30,8 +30,13 @@ public:
 
 private:
   rclcpp::Subscription<std_msgs::msg::String>::SharedPtr sub_;
+  rclcpp::TimerBase::SharedPtr timer_;
+
+  void onTimer() const;
+
+  std::string curr_str_{};
 };
 
 }  // namespace composition
 
-#endif  // COMPOSITION__LISTENER_COMPONENT_HPP_
+#endif  // COMPOSITION__LISTENER_COMPONENT_HPP_
\ No newline at end of file
diff --git a/composition/launch/composition_demo.launch.py b/composition/launch/composition_demo.launch.py
index e202c65..f49b620 100644
--- a/composition/launch/composition_demo.launch.py
+++ b/composition/launch/composition_demo.launch.py
@@ -25,7 +25,7 @@ def generate_launch_description():
             name='my_container',
             namespace='',
             package='rclcpp_components',
-            executable='component_container',
+            executable='component_container_mt',
             composable_node_descriptions=[
                 ComposableNode(
                     package='composition',
diff --git a/composition/src/listener_component.cpp b/composition/src/listener_component.cpp
index 823b2b3..e8a3451 100644
--- a/composition/src/listener_component.cpp
+++ b/composition/src/listener_component.cpp
@@ -16,6 +16,7 @@
 
 #include <iostream>
 #include <memory>
+#include <thread>
 
 #include "rclcpp/rclcpp.hpp"
 #include "std_msgs/msg/string.hpp"
@@ -23,6 +24,8 @@
 namespace composition
 {
 
+using namespace std::chrono_literals;
+
 // Create a Listener "component" that subclasses the generic rclcpp::Node base class.
 // Components get built into shared libraries and as such do not write their own main functions.
 // The process using the component's shared library will instantiate the class as a ROS node.
@@ -32,8 +35,9 @@ Listener::Listener(const rclcpp::NodeOptions & options)
   // Create a callback function for when messages are received.
   // Variations of this function also exist using, for example, UniquePtr for zero-copy transport.
   auto callback =
-    [this](std_msgs::msg::String::ConstSharedPtr msg) -> void
+    [this](const typename std_msgs::msg::String::SharedPtr msg) -> void
     {
+      curr_str_ = msg->data;
       RCLCPP_INFO(this->get_logger(), "I heard: [%s]", msg->data.c_str());
       std::flush(std::cout);
     };
@@ -43,6 +47,15 @@ Listener::Listener(const rclcpp::NodeOptions & options)
   // Note that not all publishers on the same topic with the same type will be compatible:
   // they must have compatible Quality of Service policies.
   sub_ = create_subscription<std_msgs::msg::String>("chatter", 10, callback);
+
+  timer_ = rclcpp::create_timer(this, get_clock(), 1s, std::bind(&Listener::onTimer, this));
+}
+
+void Listener::onTimer() const {
+  RCLCPP_INFO(this->get_logger(), "Current string: [%s]", this->curr_str_.c_str());
+    std::flush(std::cout);
+    // simulate slower processing than timer callback
+    std::this_thread::sleep_for(1.500s);
 }
 
 }  // namespace composition
@@ -52,4 +65,4 @@ Listener::Listener(const rclcpp::NodeOptions & options)
 // Register the component with class_loader.
 // This acts as a sort of entry point, allowing the component to be discoverable when its library
 // is being loaded into a running process.
-RCLCPP_COMPONENTS_REGISTER_NODE(composition::Listener)
+RCLCPP_COMPONENTS_REGISTER_NODE(composition::Listener)
\ No newline at end of file

Basically, I simply added a timer to the listener node, whose callback is slower than its execution period.

When I use the single threaded executor (executable='component_container' in composition_demo.launch.py), I get this:

$ ros2 launch composition composition_demo.launch.py
[...]
[component_container-1] [INFO] [1704864772.765324745] [talker]: Publishing: 'Hello World: 1'
[component_container-1] [INFO] [1704864772.765500181] [listener]: I heard: [Hello World: 1]
[component_container-1] [INFO] [1704864772.768922694] [listener]: Current string: [Hello World: 1]
[component_container-1] [INFO] [1704864774.269076401] [talker]: Publishing: 'Hello World: 2'
[component_container-1] [INFO] [1704864774.269121776] [listener]: Current string: [Hello World: 1]
[component_container-1] [INFO] [1704864775.769233579] [talker]: Publishing: 'Hello World: 3'
[component_container-1] [INFO] [1704864775.769307784] [listener]: Current string: [Hello World: 1]
[component_container-1] [INFO] [1704864777.269491127] [listener]: I heard: [Hello World: 2]
[component_container-1] [INFO] [1704864777.269549417] [talker]: Publishing: 'Hello World: 4'
[component_container-1] [INFO] [1704864777.269565038] [listener]: Current string: [Hello World: 2]
[component_container-1] [INFO] [1704864778.769749209] [listener]: I heard: [Hello World: 3]
[component_container-1] [INFO] [1704864778.769807402] [talker]: Publishing: 'Hello World: 5'
[component_container-1] [INFO] [1704864778.769822605] [listener]: Current string: [Hello World: 3]
[component_container-1] [INFO] [1704864780.270001615] [listener]: I heard: [Hello World: 4]
[component_container-1] [INFO] [1704864780.270059423] [talker]: Publishing: 'Hello World: 6'
[component_container-1] [INFO] [1704864780.270075196] [listener]: Current string: [Hello World: 4]
[component_container-1] [INFO] [1704864781.770250968] [listener]: I heard: [Hello World: 5]
[component_container-1] [INFO] [1704864781.770306203] [talker]: Publishing: 'Hello World: 7'
[component_container-1] [INFO] [1704864781.770321941] [listener]: Current string: [Hello World: 5]
[component_container-1] [INFO] [1704864783.270500143] [listener]: I heard: [Hello World: 6]
[component_container-1] [INFO] [1704864783.270555765] [talker]: Publishing: 'Hello World: 8'
[component_container-1] [INFO] [1704864783.270573654] [listener]: Current string: [Hello World: 6]
[component_container-1] [INFO] [1704864784.770755508] [listener]: I heard: [Hello World: 7]
[component_container-1] [INFO] [1704864784.770813820] [talker]: Publishing: 'Hello World: 9'
[component_container-1] [INFO] [1704864784.770830168] [listener]: Current string: [Hello World: 7]
[component_container-1] [INFO] [1704864786.271009191] [listener]: I heard: [Hello World: 8]
[component_container-1] [INFO] [1704864786.271072450] [talker]: Publishing: 'Hello World: 10'
[component_container-1] [INFO] [1704864786.271087417] [listener]: Current string: [Hello World: 8]
[component_container-1] [INFO] [1704864787.771269854] [listener]: I heard: [Hello World: 9]
[component_container-1] [INFO] [1704864787.771326406] [talker]: Publishing: 'Hello World: 11'
[component_container-1] [INFO] [1704864787.771342631] [listener]: Current string: [Hello World: 9]

You can see that both the timer and subscription callbacks are executed.

However if I use the multithreaded executor (executable='component_container_mt'), then the subscription callback is never executed:

$ ros2 launch composition composition_demo.launch.py
[...]
[INFO] [launch_ros.actions.load_composable_nodes]: Loaded node '/listener' in container '/my_container'
[component_container_mt-1] [INFO] [1704866665.903208167] [talker]: Publishing: 'Hello World: 1'
[component_container_mt-1] [INFO] [1704866665.903320661] [listener]: I heard: [Hello World: 1]
[component_container_mt-1] [INFO] [1704866665.906383657] [listener]: Current string: [Hello World: 1]
[component_container_mt-1] [INFO] [1704866666.903197114] [talker]: Publishing: 'Hello World: 2'
[component_container_mt-1] [INFO] [1704866667.406648073] [listener]: Current string: [Hello World: 1]
[component_container_mt-1] [INFO] [1704866667.903164950] [talker]: Publishing: 'Hello World: 3'
[component_container_mt-1] [INFO] [1704866668.903140064] [talker]: Publishing: 'Hello World: 4'
[component_container_mt-1] [INFO] [1704866668.906855404] [listener]: Current string: [Hello World: 1]
[component_container_mt-1] [INFO] [1704866669.903160298] [talker]: Publishing: 'Hello World: 5'
[component_container_mt-1] [INFO] [1704866670.407132270] [listener]: Current string: [Hello World: 1]
[component_container_mt-1] [INFO] [1704866670.903111712] [talker]: Publishing: 'Hello World: 6'
[component_container_mt-1] [INFO] [1704866671.903094628] [talker]: Publishing: 'Hello World: 7'
[component_container_mt-1] [INFO] [1704866671.907396505] [listener]: Current string: [Hello World: 1]
[component_container_mt-1] [INFO] [1704866672.903083101] [talker]: Publishing: 'Hello World: 8'
[component_container_mt-1] [INFO] [1704866673.407675984] [listener]: Current string: [Hello World: 1]
[component_container_mt-1] [INFO] [1704866673.903063941] [talker]: Publishing: 'Hello World: 9'

Expected behavior

The single and multithreaded executor should have consistent behavior in this case.

Actual behavior

With the multithreaded executor, the timer callback seems to have priority over the subscription callback

Additional information

---

Feature request

Feature description

Implementation considerations