Replace the rosbag2_py recorder executor with the EventsCBGExecutor

[skyegalaxy]

Description

This PR swaps the SingleThreadedExecutor used in the rosbag2_py recorder class with the EventsCBGExecutor, resulting in on average 15 - 23% less CPU with the existing benchmark (each publisher publishing at the same time in a loop), and up to 43% less CPU with staggered publishing.

This PR also adds a few more benchmark configs to rosbag2_performance_benchmarking which are more focused on benchmarking the message rate rather than bandwidth / IO.

See more extensive benchmarks below in Additional Information

Follow-up to the discussion in #743

Is this user-facing behavior change?

Not really. Functionality is preserved, but with less CPU usage

Did you use Generative AI?

Claude Opus 5 was used to generate the benchmarking harnesses and new topologies, with a lot of paring down by me.

Additional Information

• All benchmarks were done on my x86 lenovo developer laptop.

• FastDDS was used for all these runs. Note, that at 1000hz, I was able to reproduce the already documented deadlock issues that this PR addresses. All runs were done with this PR cherry-picked into my workspace.

• All CPU values are expressed as % of a core.

Pre-existing rosbag benchmarks (3 runs each):

┌─────────────────────────────┬─────────┬──────┬────────┬─────────┬────────┬─────────┬─────────┐
│       Producer config       │  msg/s  │ MB/s │ Stock  │ Patched │ Change │ t value │ p value │
├─────────────────────────────┼─────────┼──────┼────────┼─────────┼────────┼─────────┼─────────┤
│ mixed_110Mbs                │ 100,010 │ 60   │ 159.8% │ 121.8%  │ −23.8% │ 41.5    │ 0.0006  │
├─────────────────────────────┼─────────┼──────┼────────┼─────────┼────────┼─────────┼─────────┤
│ mixed_110Mbs_low_pubs_count │ 20,010  │ 110  │ 33.3%  │ 30.9%   │ −7.2%  │ 6.7     │ 0.0218  │
├─────────────────────────────┼─────────┼──────┼────────┼─────────┼────────┼─────────┼─────────┤
│ automotive                  │ 1,124   │ 573  │ 58.3%  │ 59.2%   │ +1.6%  │ −0.3    │ 0.8245  │
├─────────────────────────────┼─────────┼──────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 100MBs_raw                  │ 500     │ 100  │ 15.9%  │ 15.4%   │ −2.9%  │ 1.8     │ 0.2149  │
├─────────────────────────────┼─────────┼──────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 200MBs_raw                  │ 500     │ 200  │ 27.3%  │ 26.8%   │ −2.0%  │ 3.7     │ 0.0669  │
├─────────────────────────────┼─────────┼──────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 300MBs_raw                  │ 500     │ 300  │ 35.3%  │ 35.8%   │ +1.2%  │ −0.3    │ 0.7685  │
└─────────────────────────────┴─────────┴──────┴────────┴─────────┴────────┴─────────┴─────────┘

many_small saturation ladder (20 runs each):

┌───────────────────┬─────────┬────────┬─────────┬────────┬─────────┬─────────┐
│ Publish frequency │  msg/s  │ Stock  │ Patched │ Change │ t value │ p value │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 200 Hz            │ 40,000  │ 50.4%  │ 42.6%   │ −15.5% │ 21.6    │ 8e−15   │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 400 Hz            │ 80,000  │ 88.3%  │ 74.6%   │ −15.6% │ 38.2    │ 2e−19   │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 600 Hz            │ 120,000 │ 129.4% │ 102.5%  │ −20.8% │ 65.1    │ 9e−24   │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 800 Hz            │ 160,000 │ 163.8% │ 147.5%  │ −9.9%  │ 15.8    │ 2e−12   │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 1000 Hz           │ 200,000 │ 175.0% │ 162.2%  │ −7.3%  │ 5.7     │ 2e−05   │
└───────────────────┴─────────┴────────┴─────────┴────────┴─────────┴─────────┘

CPU Usage by Thread was also inferred (according to the thread name and usage) across these runs, by sampling every thread under /proc/<pid>/task every 0.5 seconds.

The effects on latency were benchmarked by computing the recv_timestamp - send_timestamp in a utility script. In exchange for lower CPU, it's worth noting there is around a ~30 microsecond per message increase in latency, on average.

┌───────────────────┬───────────┬─────────────┬──────────────────┬─────────┐
│ Publish frequency │ Stock p50 │ Patched p50 │    Difference    │ t value │
├───────────────────┼───────────┼─────────────┼──────────────────┼─────────┤
│ 200 Hz            │ 633 µs    │ 660 µs      │ +27 µs (+4.3%)   │ 3.0     │
├───────────────────┼───────────┼─────────────┼──────────────────┼─────────┤
│ 400 Hz            │ 577 µs    │ 610 µs      │ +32 µs (+5.6%)   │ 3.6     │
├───────────────────┼───────────┼─────────────┼──────────────────┼─────────┤
│ 600 Hz            │ 632 µs    │ 615 µs      │ −17 µs (−2.7%)   │ −2.7    │
├───────────────────┼───────────┼─────────────┼──────────────────┼─────────┤
│ 800 Hz †          │ 427 µs    │ 473 µs      │ +46 µs (+10.8%)  │ 2.4     │
├───────────────────┼───────────┼─────────────┼──────────────────┼─────────┤
│ 1000 Hz †         │ 296 µs    │ 464 µs      │ +168 µs (+56.9%) │ 2.7     │
└───────────────────┴───────────┴─────────────┴──────────────────┴─────────┘

As a footnote, the use of multiple recording threads was experimented with, but this only increased the CPU usage without any real benefits to latency at any message publication frequency. Therefore, this PR simply replaces a single-threaded wait set executor with a single threaded events queue based one.

[skyegalaxy]

Update: per @jmachowinski 's suggestion, I made a slight modification to benchmark_publishers.cpp to stagger the publishing instead of publishing every message in a loop at the same time, which seems like a more realistic simulation of traffic on a heavily populated ROS graph. The results are even more dramatic in favor of the new executor.

┌───────────────────┬─────────┬────────┬─────────┬────────┬─────────┬─────────┐
│ Publish frequency │  msg/s  │ Stock  │ Patched │ Change │ t value │ p value │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 200 Hz            │ 40,000  │ 106.0% │ 60.8%   │ −42.6% │ 122.5   │ 8e-16   │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 400 Hz            │ 80,000  │ 120.5% │ 86.4%   │ −28.3% │ 83.5    │ 3e-14   │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 600 Hz            │ 120,000 │ 137.7% │ 101.6%  │ −26.2% │ 150.9   │ 1e-16   │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 800 Hz            │ 160,000 │ 154.2% │ 132.9%  │ −13.8% │ 13.1    │ 4e-07   │
├───────────────────┼─────────┼────────┼─────────┼────────┼─────────┼─────────┤
│ 1000 Hz †         │ 200,000 │ 177.6% │ 170.5%  │ −4.0%  │ 2.3     │ 0.0494  │
└───────────────────┴─────────┴────────┴─────────┴────────┴─────────┴─────────┘

When each message arrives staggered vs at the same time, the SingleThreadedExecutor is paying the same O(n) ready entity discovery cost for each arrival, whereas the events subsystem incurs no such cost. Which is why the baseline in staggered mode for the SingleThreadedExecutor is around 60% more than when message arrival isn't staggered.

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request

[ahcorde]

Pulls: #2472
Gist: https://gist.githubusercontent.com/ahcorde/fc693128a72f3e34a73dcd18ddec48e9/raw/9f1d5694e8a8350689ce29fa1c4a20a1e9de45c2/ros2.repos
BUILD args: --packages-up-to rosbag2_performance
TEST args: --packages-select rosbag2_performance
ROS Distro: rolling
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/20132

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

[jmachowinski]

I wonder if we should guard this with a parameter.
On a system that can't hold up with processing the messages this will lead to a memory leak over time.

[skyegalaxy]

Pulls: #2472
Gist: https://gist.githubusercontent.com/skyegalaxy/f3b97c3c21ff9c1cd77dff162656dcfb/raw/9f1d5694e8a8350689ce29fa1c4a20a1e9de45c2/ros2.repos
BUILD args: --packages-above-and-dependencies rosbag2_py
TEST args: --packages-above rosbag2_py
ROS Distro: rolling
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/20138

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

[fujitatomoya]

lgtm, i mean why not?

one minor comment is applying the same executor change to the Player? i understand the Recorder is the critical path, but any downside to apply EGB Executor to the Player as well?

[ahcorde]

lgtm, i mean why not?

one minor comment is applying the same executor change to the Player? i understand the Recorder is the critical path, but any downside to apply EGB Executor to the Player as well?

@skyegalaxy ?

[skyegalaxy]

lgtm, i mean why not?

one minor comment is applying the same executor change to the Player? i understand the Recorder is the critical path, but any downside to apply EGB Executor to the Player as well?

I don't think there'd be any downsides, we just wouldn't have as dramatic of performance improvement we're seeing here with the recorder since that node's only responsible for a handful of services. Happy to add it to this PR or follow up with another one, since we want to move in the direction of events based executors being the default anyway