feat(EventsCBGExecutor): Added worker queue

[jmachowinski]

Description

The worker queue should improve the usage of cold vs hot threads, by reusing threads that have been recently used.

There are also some improvements around not waking up threads if they can't get work anyway.

Is this user-facing behavior change?

No

Did you use Generative AI?

Yes for code reviews and checks

Additional Information

[jmachowinski]

Pulls: #3277
Gist: https://gist.githubusercontent.com/jmachowinski/fa6541200506e08495142fae71d2d6b2/raw/751aaa1a4d154e4dee96b231bbc5e488917bb539/ros2.repos
BUILD args:
TEST args:
ROS Distro: rolling
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/20486

• Linux
• Linux-aarch64
• Linux-rhel
• Windows

[github-actions]

ABI Compliance Check

❌ Verdict: incompatible

Library	Verdict	Summary
libcomponent_manager.so	✅ compatible	No ABI changes detected.
librclcpp.so	❌ incompatible	ABI-incompatible changes detected.
librclcpp_action.so	✅ compatible	No ABI changes detected.
librclcpp_lifecycle.so	✅ compatible	No ABI changes detected.

✅ libcomponent_manager.so — full abidiff report

Compared:

• Base: lib-base/libcomponent_manager.so
• Head: lib-pr/libcomponent_manager.so @ ec8c1cf

(empty report — no differences printed by abidiff)

❌ librclcpp.so — full abidiff report

Compared:

• Base: lib-base/librclcpp.so
• Head: lib-pr/librclcpp.so @ ec8c1cf

Functions changes summary: 2 Removed (3 filtered out), 2 Changed (134 filtered out), 10 Added (46 filtered out) functions
Variables changes summary: 0 Removed, 0 Changed, 0 Added variable
Function symbols changes summary: 2 Removed, 4 Added function symbols not referenced by debug info
Variable symbols changes summary: 0 Removed, 0 Added variable symbol not referenced by debug info

2 Removed functions:

  [D] 'method void rclcpp::executors::cbg_executor::CBGScheduler::block_worker_thread()'    {_ZN6rclcpp9executors12cbg_executor12CBGScheduler19block_worker_threadEv}
  [D] 'method void rclcpp::executors::cbg_executor::CBGScheduler::block_worker_thread_for(std::chrono::nanoseconds)'    {_ZN6rclcpp9executors12cbg_executor12CBGScheduler23block_worker_thread_forENSt6chrono8durationIlSt5ratioILl1ELl1000000000EEEE}

10 Added functions:

  [A] 'method void rclcpp::executors::cbg_executor::CBGScheduler::block_worker_thread(rclcpp::executors::cbg_executor::Worker*)'    {_ZN6rclcpp9executors12cbg_executor12CBGScheduler19block_worker_threadEPNS1_6WorkerE}
  [A] 'method void rclcpp::executors::cbg_executor::CBGScheduler::block_worker_thread_for(rclcpp::executors::cbg_executor::Worker*, std::chrono::nanoseconds)'    {_ZN6rclcpp9executors12cbg_executor12CBGScheduler23block_worker_thread_forEPNS1_6WorkerENSt6chrono8durationIlSt5ratioILl1ELl1000000000EEEE}
  [A] 'method bool rclcpp::executors::cbg_executor::CBGScheduler::prepare_and_enqueue_worker(rclcpp::executors::cbg_executor::Worker*)'    {_ZN6rclcpp9executors12cbg_executor12CBGScheduler26prepare_and_enqueue_workerEPNS1_6WorkerE}
  [A] 'method void rclcpp::executors::cbg_executor::CBGScheduler::suppress_thread_wakeup()'    {_ZN6rclcpp9executors12cbg_executor12CBGScheduler22suppress_thread_wakeupEv}
  [A] 'method void rclcpp::executors::cbg_executor::Worker::block()'    {_ZN6rclcpp9executors12cbg_executor6Worker5blockEv}
  [A] 'method void rclcpp::executors::cbg_executor::Worker::block_for(std::chrono::nanoseconds)'    {_ZN6rclcpp9executors12cbg_executor6Worker9block_forENSt6chrono8durationIlSt5ratioILl1ELl1000000000EEEE}
  [A] 'method void rclcpp::executors::cbg_executor::Worker::unblock()'    {_ZN6rclcpp9executors12cbg_executor6Worker7unblockEv}
  [A] 'method void rclcpp::executors::cbg_executor::Worker::unblock_thread_safe()'    {_ZN6rclcpp9executors12cbg_executor6Worker19unblock_thread_safeEv}
  [A] 'method rclcpp::executors::cbg_executor::Worker* rclcpp::executors::cbg_executor::WorkerQueue::pop_blocked_worker_thread()'    {_ZN6rclcpp9executors12cbg_executor11WorkerQueue25pop_blocked_worker_threadEv}
  [A] 'method void rclcpp::executors::cbg_executor::WorkerQueue::remove_worker_thread(rclcpp::executors::cbg_executor::Worker*)'    {_ZN6rclcpp9executors12cbg_executor11WorkerQueue20remove_worker_threadEPNS1_6WorkerE}

2 functions with some indirect sub-type change:

  [C] 'method rclcpp::executors::cbg_executor::CBGScheduler::CBGScheduler(std::function<void()>)' at scheduler.hpp:195:1 has some indirect sub-type changes:
    implicit parameter 0 of type 'rclcpp::executors::cbg_executor::CBGScheduler*' has sub-type changes:
      in pointed to type 'class rclcpp::executors::cbg_executor::CBGScheduler' at scheduler.hpp:163:1:
        type size changed from 1984 to 2624 (in bits)
        1 member function insertion:
          'method virtual rclcpp::executors::cbg_executor::CBGScheduler::~CBGScheduler()' at scheduler.hpp:330:1
        no member function changes (6 filtered);
        2 data member deletions:
          'bool release_workers', at offset 1344 (in bits) at scheduler.hpp:393:1
          'bool release_worker_once', at offset 1352 (in bits) at scheduler.hpp:394:1
        3 data member changes (1 filtered):
          name of 'rclcpp::executors::cbg_executor::CBGScheduler::work_ready_conditional' changed to 'rclcpp::executors::cbg_executor::CBGScheduler::worker_queue' at scheduler.hpp:569:1, size changed from 384 to 1024 (in bits) (by +640 bits)
          'bool worker_checking_for_work' offset changed from 1360 to 1344 (in bits) (by -16 bits)
          'std::__cxx11::list<std::unique_ptr<rclcpp::executors::cbg_executor::CBGScheduler::CallbackGroupHandle, std::default_delete<rclcpp::executors::cbg_executor::CBGScheduler::CallbackGroupHandle> >, std::allocator<std::unique_ptr<rclcpp::executors::cbg_executor::CBGScheduler::CallbackGroupHandle, std::default_delete<rclcpp::executors::cbg_executor::CBGScheduler::CallbackGroupHandle> > > > callback_groups' offset changed from 1792 to 2432 (in bits) (by +640 bits)

  [C] 'method virtual std::unique_ptr<rclcpp::executors::cbg_executor::CBGScheduler::CallbackGroupHandle, std::default_delete<rclcpp::executors::cbg_executor::CBGScheduler::CallbackGroupHandle> > rclcpp::executors::cbg_executor::FirstInFirstOutScheduler::get_handle_for_callback_group(const rclcpp::CallbackGroup::SharedPtr&)' at first_in_first_out_scheduler.cpp:149:1 has some indirect sub-type changes:
    implicit parameter 0 of type 'rclcpp::executors::cbg_executor::FirstInFirstOutScheduler*' has sub-type changes:
      in pointed to type 'class rclcpp::executors::cbg_executor::FirstInFirstOutScheduler' at first_in_first_out_scheduler.hpp:66:1:
        type size changed from 2176 to 2816 (in bits)
        1 base class change:
          'class rclcpp::executors::cbg_executor::CBGScheduler' at scheduler.hpp:32:1 changed:
            details were reported earlier
        no member function changes (3 filtered);
        1 data member change:
          'std::vector<std::unique_ptr<rclcpp::executors::cbg_executor::FirstInFirstOutCallbackGroupHandle, std::default_delete<rclcpp::executors::cbg_executor::FirstInFirstOutCallbackGroupHandle> >, std::allocator<std::unique_ptr<rclcpp::executors::cbg_executor::FirstInFirstOutCallbackGroupHandle, std::default_delete<rclcpp::executors::cbg_executor::FirstInFirstOutCallbackGroupHandle> > > > callback_group_handles' offset changed from 1984 to 2624 (in bits) (by +640 bits)

2 Removed function symbols not referenced by debug info:

  [D] _ZZN6rclcpp9executors12cbg_executor12CBGScheduler19block_worker_threadEvENKUlvE_clEv
  [D] _ZZN6rclcpp9executors12cbg_executor12CBGScheduler23block_worker_thread_forENSt6chrono8durationIlSt5ratioILl1ELl1000000000EEEEENKUlvE_clEv

4 Added function symbols not referenced by debug info:

  [A] _ZN6rclcpp9executors12cbg_executor11WorkerQueueC1Ev
  [A] _ZN6rclcpp9executors12cbg_executor11WorkerQueueC2Ev, aliases _ZN6rclcpp9executors12cbg_executor11WorkerQueueC1Ev
  [A] _ZN6rclcpp9executors12cbg_executor6WorkerC1Ev, aliases _ZN6rclcpp9executors12cbg_executor6WorkerC2Ev
  [A] _ZN6rclcpp9executors12cbg_executor6WorkerC2Ev

✅ librclcpp_action.so — full abidiff report

Compared:

• Base: lib-base/librclcpp_action.so
• Head: lib-pr/librclcpp_action.so @ ec8c1cf

(empty report — no differences printed by abidiff)

✅ librclcpp_lifecycle.so — full abidiff report

Compared:

• Base: lib-base/librclcpp_lifecycle.so
• Head: lib-pr/librclcpp_lifecycle.so @ ec8c1cf

(empty report — no differences printed by abidiff)

_{Updated for commit ec8c1cf · suppressions: /home/runner/work/_temp/ros2-abi-suppressions.txt}

[jmachowinski]

@fujitatomoya this is clearly a false positive in the abi checker. It fails to realize that everything that changed is part of a pimple pattern and not public ABI.

[jmachowinski]

Also why are we running it for rolling ?

[skyegalaxy]

@jmachowinski - I think it's so we can know if a change to rolling will successfully backport to older distros. but yeah this should probably also take into account PIMPL and whether an affected class is part of the public API or not

[skyegalaxy]

quick benchmarks:

[jmachowinski]

Hu what is going on with these benchmarks, they don't match my expectations and tests at all....

[skyegalaxy]

ah wait I think this was with irobot benchmark before all of our recent changes. Won't be able to rerun them for a few days

[fujitatomoya]

@fujitatomoya this is clearly a false positive in the abi checker. It fails to realize that everything that changed is part of a pimple pattern and not public ABI.

@jmachowinski thanks for flagging this! i need to considerr pimpl pattern probably the policy requirement by the project... honestly i missed this case... good eye! (CC: @skyegalaxy thanks for the explanation.) for now, you can ignore those abi compliance result, this has been still tested in rolling branch to detect the missing cases like exactly this one.

edit: tracking issue for ROS 2 abi checker, fujitatomoya/ros2-abi-action#15

[mjcarroll]

Yeah, I wouldn't treat the ABI results as absolute yet. I think they are the direction that we need to move in, but it may take a few iterations to get it exactly right.

Because of that, I agree with @skyegalaxy that running on rolling can at least give us a good idea of backport risk (and help inform new rules).

[fujitatomoya]

overall, lgtm with a couple of comments.

[jmachowinski]

@skyegalaxy

jm@Rechenknecht:~/ros2/rolling/src/executor_performance_test/scripts$ ./start_benchmark.sh --ros-args   -p "executor_configs:=[cbg:0,cbg:1]"   -p "subscription_counts:=[20]"   -p "subs_per_cbg_list:=[5]"   -p max_rate_hz:=1200.0 -p "rate_step_hz:=300.0"  -p run_duration_s:=5.0 -p output_csv:=worker_queue.csv
[INFO] [1789987676.438071163] [bench_control]: Waiting for benchmark publisher service...
exec          thr  subs   spc    rate     sent     recv  drop%   lat_ms   latmed     cpu%   bench%    ctrl%     rmw%     mon%   pubcpu status        
cbg           inf    20     5     100    10000    10000   0.00    0.419    0.467    45.16    11.49     0.09    31.68     1.90    45.17 ok            
cbg           inf    20     5     400    40000    40000   0.00    0.328    0.326   115.63    39.62     0.10    73.82     2.08   127.54 ok            
cbg           inf    20     5     700    70000    69968   0.05    0.398    0.350   181.49    68.01     0.12   111.10     2.27   231.81 ok            
cbg           inf    20     5    1000   100000    99976   0.02  325.223  337.000   217.85    81.98     0.10   133.38     2.39   303.87 ok            
[INFO] [1789987704.594682478] [bench_control]: Stopping rate escalation for cbg:0 subs=20 spc=5: status=ok, drop=0.0%, latency=325.223ms
cbg             1    20     5     100    10000     9996   0.04    0.420    0.466    40.97     9.52     0.11    30.09     1.25    43.26 ok            
cbg             1    20     5     400    40000    39991   0.02    0.321    0.322   100.35    26.72     0.11    72.19     1.33   117.35 ok            
cbg             1    20     5     700    70000    69998   0.00    0.397    0.350   158.38    47.35     0.11   109.33     1.59   220.98 ok            
cbg             1    20     5    1000   100000    99976   0.02   12.441    8.406   201.35    65.72     0.11   133.77     1.74   305.34 ok            
[INFO] [1789987733.086288037] [bench_control]: Stopping rate escalation for cbg:1 subs=20 spc=5: status=ok, drop=0.0%, latency=12.441ms
[INFO] [1789987733.087373619] [bench_control]: Benchmark finished, results in worker_queue.csv
jm@Rechenknecht:~/ros2/rolling/src/executor_performance_test/scripts$ ./start_benchmark.sh --ros-args   -p "executor_configs:=[cbg:0,cbg:1]"   -p "subscription_counts:=[20]"   -p "subs_per_cbg_list:=[5]"   -p max_rate_hz:=1200.0 -p "rate_step_hz:=300.0"  -p run_duration_s:=5.0 -p output_csv:=rolling.csv
[INFO] [1789987823.448608104] [bench_control]: Waiting for benchmark publisher service...
exec          thr  subs   spc    rate     sent     recv  drop%   lat_ms   latmed     cpu%   bench%    ctrl%     rmw%     mon%   pubcpu status        
cbg           inf    20     5     100    10000    10000   0.00    0.441    0.472    47.17    12.73     0.11    32.23     2.10    46.17 ok            
cbg           inf    20     5     400    40000    39993   0.02    0.369    0.367   126.93    47.56     0.09    77.05     2.23   136.39 ok            
cbg           inf    20     5     700    70000    69963   0.05    0.429    0.386   201.22    85.67     0.10   113.09     2.36   244.19 ok            
cbg           inf    20     5    1000   100000    99888   0.11  312.295  327.866   241.84   105.03     0.10   134.30     2.41   312.34 ok            
[INFO] [1789987851.740289174] [bench_control]: Stopping rate escalation for cbg:0 subs=20 spc=5: status=ok, drop=0.1%, latency=312.295ms
cbg             1    20     5     100    10000    10000   0.00    0.430    0.474    42.15     9.60     0.09    31.12     1.35    44.39 ok            
cbg             1    20     5     400    40000    39985   0.04    0.347    0.338   110.69    29.21     0.10    79.89     1.49   140.48 ok            
cbg             1    20     5     700    70000    69985   0.02    0.356    0.336   156.30    45.43     0.11   109.17     1.59   224.94 ok            
cbg             1    20     5    1000   100000    99962   0.04    6.341    1.855   197.85    63.24     0.11   132.84     1.67   301.65 ok            
[INFO] [1789987880.178603217] [bench_control]: Stopping rate escalation for cbg:1 subs=20 spc=5: status=ok, drop=0.0%, latency=6.341ms

I get through the bank better performance results in my benchmark with this branch

[skyegalaxy]

Pulls: #3277
Gist: https://gist.githubusercontent.com/skyegalaxy/8f79c92400849745389781cc8e1c30b9/raw/751aaa1a4d154e4dee96b231bbc5e488917bb539/ros2.repos
BUILD args: --packages-above-and-dependencies rclcpp
TEST args: --packages-above rclcpp
ROS Distro: rolling
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/20582

• Linux
• Linux-aarch64
• Linux-rhel
• Windows