EventsCBGExecutor treats reentrant callback groups like they're mutually exclusive

[armaho]

Generated by Generative AI

No

Operating System:

Linux pi 7.0.0-1011-raspi-realtime #11-Ubuntu SMP PREEMPT_RT Tue May 26 15:20:46 UTC 2026 aarch64 GNU/Linux

ROS version or commit hash:

rolling

RMW implementation (if applicable):

No response

RMW Configuration (if applicable):

No response

Client library (if applicable):

No response

'ros2 doctor --report' output

No response

Steps to reproduce issue

I used this piece of code. I also used yaets for tracing the callbacks.

#include "rclcpp/rclcpp.hpp"
#include <fstream>

// #include "yaets/tracing.hpp"

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/int32.hpp"

using namespace std::chrono_literals;
using std::placeholders::_1;


// yaets::TraceSession session("session1.log");

class ProducerNode : public rclcpp::Node
{
public:
  ProducerNode() : Node("producer_node")
  {
    pub_1_ = create_publisher<std_msgs::msg::Int32>("topic_1", 100);
    pub_2_ = create_publisher<std_msgs::msg::Int32>("topic_2", 100);
    timer_ = create_wall_timer(1ms, std::bind(&ProducerNode::timer_callback, this));
  }

  void timer_callback()
  {
    // TRACE_EVENT(session);
    message_.data += 1;
    pub_1_->publish(message_);
    message_.data += 1;
    pub_2_->publish(message_);
  }

private:
  rclcpp::Publisher<std_msgs::msg::Int32>::SharedPtr pub_1_, pub_2_;
  rclcpp::TimerBase::SharedPtr timer_;
  std_msgs::msg::Int32 message_;
};

class ConsumerNode : public rclcpp::Node
{
public:
  ConsumerNode() : Node("consumer_node")
  {
    auto cg = this->create_callback_group(rclcpp::CallbackGroupType::Reentrant);

    rclcpp::SubscriptionOptions opt;
    opt.callback_group = cg;

    sub_2_ = create_subscription<std_msgs::msg::Int32>(
      "topic_2", 100, std::bind(&ConsumerNode::cb_2, this, _1), opt);
    sub_1_ = create_subscription<std_msgs::msg::Int32>(
      "topic_1", 100, std::bind(&ConsumerNode::cb_1, this, _1), opt);
 
    timer_ = create_wall_timer(10ms, std::bind(&ConsumerNode::timer_callback, this));
  }

  void cb_1(const std_msgs::msg::Int32::SharedPtr msg)
  {
    // TRACE_EVENT(session);

    waste_time(500us);
  }

  void cb_2(const std_msgs::msg::Int32::SharedPtr msg)
  {
    // TRACE_EVENT(session);

    waste_time(500us);
  }

  void timer_callback()
  {
    // TRACE_EVENT(session);

    waste_time(5ms);
  }

  void waste_time(const rclcpp::Duration & duration)
  {
    auto start = now();
    while (now() - start < duration);
  }

private:
  rclcpp::Subscription<std_msgs::msg::Int32>::SharedPtr sub_1_;
  rclcpp::Subscription<std_msgs::msg::Int32>::SharedPtr sub_2_;
  rclcpp::TimerBase::SharedPtr timer_;
};

int main(int argc, char * argv[])
{
  rclcpp::init(argc, argv);

  auto node_pub = std::make_shared<ProducerNode>();
  auto node_sub = std::make_shared<ConsumerNode>();

  // rclcpp::executors::SingleThreadedExecutor executor;
  // rclcpp::executors::MultiThreadedExecutor executor(rclcpp::ExecutorOptions(), 8);
  rclcpp::executors::EventsCBGExecutor executor(rclcpp::ExecutorOptions(), 8);

  executor.add_node(node_pub);
  executor.add_node(node_sub);

  executor.spin();

  rclcpp::shutdown();
  return 0;
}

Expected behavior

As you can see, sub_1_ and sub_2_ are part of the same reentrant callback group. So, like what happens with MultiThreadedExecutor they should be able to execute simultaneously:

Actual behavior

Under EventsCBGExecutor, These two callbacks are executed sequentially:

Additional information

I might be completely wrong here. Sorry if I am :)

I think that the problem comes from the CallbackGroupHandle structure. It has a not_ready field:

// will be set if cbg is mutual exclusive and something is executing
bool not_ready = false;

Based on the code, it appears that this field will be set to true regardless of the actual type of the underlying callback group. It happens by calling mark_as_executing here:

std::optional<CBGScheduler::ExecutableEntity> FirstInFirstOutCallbackGroupHandle::
get_next_ready_entity()
{
  std::lock_guard l(ready_mutex);

  while(!ready_entities.empty()) {
    auto & first = ready_entities.front();

    std::function<void()> exec_fun = first.get_execute_function();
    ready_entities.pop_front();
    if(!exec_fun) {
      // was deleted, or in case of timer was canceled
      continue;
    }

    mark_as_executing();

    return CBGScheduler::ExecutableEntity{exec_fun, this};
  }

  mark_as_skipped();

  return std::nullopt;
}

[jmachowinski]

This was simply never implemented by me. Patches are welcome ;-)

[armaho]

@jmachowinski

Let me investigate a bit more I think I can come up with something :)

[jmachowinski]

Hm, this might be tricky.
You will need to readd a callback group in get_next_ready_entity_intern directly if another entitiy in it is ready.
And will also need to change the logic in the FirstInFirstOutCallbackGroupHandle for the idle stuff to get the thread wakeup right.

[armaho]

@jmachowinski

I was able to hack together a fix for now, by not removing reentrant callback groups with at least one ready entity from FIFO scheduler:

    if (ready_cbg->get_type() == CallbackGroupType::MutuallyExclusive || !ready_cbg->has_ready_entities()) {
      ready_callback_groups.pop_front();
    }

Here's the execution trace after this change:

I have to dig deeper to make sure that this does not affect the scheduler in a bad way. I will create a PR when I'm more certain.

[jmachowinski]

Today is also client workgroup meeting, if you got questions you want to talk about, feel free to join :
https://calendar.google.com/calendar/u/0/embed?src=agf3kajirket8khktupm9go748@group.calendar.google.com&ctz=CET

https://meet.google.com/oij-atzj-znw

[armaho]

@jmachowinski

Thanks! Just one thing. Whenever I make a change in rclcpp (even if it's in the implementation files) I have to wait for like 20 minutes for colcon to build it. Is there any workaround for this? Can I read about it somewhere? :)

[jmachowinski]

I suppose you use some pkg XYZ to test your findings.

you can do a
"colcon build --mixin release ccache mold --packages-above rclcpp --packages-up-to XYZ"
That should only build rclcpp and the packages needed for your test package.

You can also specify the mixin "ccache" and "mold" to speed up compilation times.
20 minutes sound way to long, rebuilding if you only modified non exposed internals in rclcpp should take 1-2min for a full stack (on linux).

[armaho]

@jmachowinski

Hello, I made some progress on this. Here's the summary of what I have changed:

I added a type field to CallbackGroupHandle that reflects the type of the underlying callback group.

To make sure that reentrant callback groups remain in the ready_cbg_queue of the scheduler, I added a condition to push them back into the queue:

CBGScheduler::ExecutableEntityWithInfo FirstInFirstOutScheduler::get_next_ready_entity_intern()
{
  while(!ready_callback_groups.empty()) {
    // ...
    if (ready_cbg->get_type() == CallbackGroupType::Reentrant && ready_cbg->has_ready_entities()) {
      ready_callback_groups.push_back(ready_cbg);
    }
    // ...
  }
}

I'm pushing the group again rather than preventing its popping, cause I'm afraid that a reentrant callback group that always has ready entities will cause starvation for other groups. I have made similar changes to get_next_ready_entity_intern(max_id).

I changed the callback_group_ready() function to check whether the group was already in the queue or not:

void callback_group_ready(CallbackGroupHandle *handle, bool callback_group_was_idle)
{
  {
    std::lock_guard l(ready_callback_groups_mutex);

    // Reentrant callback groups might not be removed from the queue when one of
    // their entities starts executing.
    if (handle->get_type() == CallbackGroupType::Reentrant) {
      bool already_in_queue = false;

      for (auto it = ready_callback_groups.begin(); it != ready_callback_groups.end(); it++) {
        if (*it == handle) {
          already_in_queue = true;
          break;
        }
      }

      if (!already_in_queue) {
        ready_callback_groups.push_back(handle);
      }
    } else {
      ready_callback_groups.push_back(handle);
    }
  }

  if(callback_group_was_idle) {
    unblock_one_worker_thread();
  }
}

Here's the current behavior of the executor:

Do these changes look alright to you?

[jmachowinski]

Can you open a PR for the changes, this makes it easier for me to review them.

[armaho]

@jmachowinski I did.