Add support for replaying multiple bags

[christophebedard]

This PR adds support for replaying multiple bags with ros2 bag play and the underlying classes (rosbag2_transport::Player/rosbag2_py::Player).

To replay multiple bags, use the new -i,--input CLI option.

$ ros2 bag play -i bag1 -i bag2 -i bag3 [storage_id]

Like with the ros2 bag convert option, the -i,--input option takes an URI to the bag folder or dedicated bag file and an optional explicit storage ID. This option can be used multiple times to be able to specify multiple bags. We keep the bag_path positional argument for backward compatibility and simplicity of usage in cases where users need to play just one bag file. However, the standalone --storage CLI option has been deprecated. In the future, users will need to use -i bag [storage_id] if they need to specify the storage ID explicitly.

The rosbag2_transport::Player now uses a std::priority_queue, which required changing the logic and interaction between PlayerImpl::play_messages_from_queue() and PlayerImpl::play_next(). In short, message publication is now all handled by PlayerImpl::play_messages_from_queue(). PlayerImpl::play_next() simply tells it to play the next message and then waits for the result. While the queue is not lock-free anymore, this change does simplify the interaction between PlayerImpl::play_messages_from_queue() and PlayerImpl::play_next() a bit. Note: the priority queue depth is defined by the --read-ahead-queue-size CLI option, which is set to 1000 elements by default.

Messages are played in order, based on their recv_timestamp (recorder reception timestamp). However, with the priority queue, this could eventually be changed (e.g., through PlayOptions) to the send_timestamp (original publication timestamp).

Other notes for reviewers:

1. Like ros2 bag convert, I added the -i, --input option that takes a bag URI and an optional explicit storage ID. This option can be used multiple times for multiple bags.
2. I kept the bag_path positional argument (but made it optional) because it is quite nice to be able to just do ros2 bag play my-bag. However, I deprecated the storage ID option (--storage); users will have to use -i, --input to specify an explicit storage ID for an input bag.
3. This PR doesn't introduce API/ABI breaking changes and can be backported to the jazzy branch. Some APIs needed to be added (mainly constructors) to support N input bags.
   • We can consider deprecating some APIs in another PR.

[MichaelOrlov]

Hi @christophebedard, Thank you for your contribution.

After thoroughly reviewing and considering the newly added class MetaPriorityQueue.
I think it will not be much more efficient than std::priority_queue protected with mutex from the STL.
For a few reasons:

1. The MetaPriorityQueue push(message) and pop() became blocking calls and also had to be protected by a mutex.
2. We impose a constraint on the MetaPriorityQueue that items queued into the same queue shall be in order of priority. That might not be true if we need to switch playback to use send_timestamp.
3. If considering performance, the std::priority_queue has log(n) runtime complexity for insertion and extraction operations and constant for peaking maximum element. While we are trying to achieve a constant time performance when using lock-free queues for relatively small numbers, the log(n) overhead is going to be very insignificant. For instance, we currently use 1000 elements for the queue by default. If we try to read out from 5 files and increase the maximum queue size by a factor of 5, i.e., to be 5000 elements, the log(5000) = 8.5. In the worst-case scenario, we will require 8.5 pointers swaps or iterations in the array. In comparison to the current design with MetaPriorityQueue, we will have at least 5 iterations over 5 queues trying to find and update the current minimum index, and also, the overall underlying implementation of the lock-free queue, is not free and has some extra operations inside. The difference is going to be in the numbers of the 5-8 iterations over the array or swaps. Which is IMO negligible in this case.
   I would prefer to have a more cleaner and straightforward impelmentation based on the std::priority_queue.

[christophebedard]

Changed to std::priority_queue.

[MichaelOrlov]

@christophebedard Thanks for the updates and one more iteration.
Overall seems logically correct. See my comments and requests for changes inline.

[christophebedard]

I'll rebase and fix the DCO & merge conflicts once the new changes look good.

[MichaelOrlov]

@christophebedard Thank you for the next round of changes and improvements.
I started liking the way how it is implemented.
I've made one more round of review. Mostly in the rosbag2_transport player classes. See my comments inline.

I also noticed that if we will not change get_storage_options() API in the player.hpp file, this PR will likely become API/ABI compatible, and we will be able to backport it to Jazzy.

[MichaelOrlov]

@christophebedard I've verified previous changes and made one more round of full review.
I have some suggestions for improvements but they are not require changes in the logic.
One of the generic suggestions is to rename local variables and constructor parameters input_bags as well as private class member variable input_bags_ to the readers_with_options.
IMO it is more clear what it is from the name. It will be helpful for future code supportability. at least for the local member variables and ctor parameters.

Another comment is to create a separate add_standard_multi_reader_args(parser: ArgumentParser) function.
The rationale for that request is a difference in behavior. i.e., We deprecate --storage and provide --input parameters only for the ros2 bag play verb. The logic in the combined function became overcomplicated and not clear what is applied and when.
Also, see my other comments inline.

[MichaelOrlov]

@christophebedard Thank you for addressing all comments.
Now looks good to me.
Please rebase and fix DCO.

[christophebedard]

I just squashed to fix the DCO and rebased.

I also fixed a linter issue: https://github.com/ros2/rosbag2/compare/a82f2597188a67cb11752815481d130fec18a9bb..eeca3aad929d4b5d7c6882d47ddc128af4a09181

[MichaelOrlov]

@christophebedard It seems we've got a regression in rosbag2_transport tests on RPR job: https://build.ros2.org/job/Rpr__rosbag2__ubuntu_noble_amd64/448/testReport/

[christophebedard]

@ros-pull-request-builder retest this please

[MichaelOrlov]

It seems it was a false alarm. The RPR job passed green from the second start. I also was not able to reproduce any failures on my local setup. Running full CI then.

[MichaelOrlov]

Pulls: #1848
Gist: https://gist.githubusercontent.com/MichaelOrlov/a8e3f30612a2ef94072b2023ba655197/raw/89ae0fbb84ac37308f77b3376800249f2a8c1bf8/ros2.repos
BUILD args: --packages-above-and-dependencies rosbag2_cpp rosbag2_transport rosbag2_py ros2bag rosbag2
TEST args: --packages-above rosbag2_cpp rosbag2_transport rosbag2_py ros2bag rosbag2
ROS Distro: rolling
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/14877

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

[MichaelOrlov]

Two warnings on the Windows build are in the StaticSingleThreaded executor

'rclcpp::executors::StaticSingleThreadedExecutor::~StaticSingleThreadedExecutor': Use rclcpp::executors::SingleThreadedExecutor (compiling source file C:\ci\ws\src\ros2\rclcpp\rclcpp\test\rclcpp\executors\test_executors.cpp)

and unrelated to the changes from this PR. Merging then to the Rolling.

cc: @christophebedard I changed a bit PR description and title to be more descriptive for others.

[christophebedard]

It seems it was a false alarm. The RPR job passed green from the second start. I also was not able to reproduce any failures on my local setup. Running full CI then.

I ran the rosbag2_transport test_play_* tests under stress --cpu $num_cpu for ~5 hours yesterday and couldn't reproduce.

[christophebedard]

Thanks for the review, @MichaelOrlov

[MichaelOrlov]

https://github.com/Mergifyio backport jazzy

[mergify]

backport jazzy

✅ Backports have been created

• #1873 Add support for replaying multiple bags (backport #1848) has been created for branch jazzy but encountered conflicts

[christophebedard]

@MichaelOrlov should I proceed with marking some of the old APIs (i.e., single input bag) as deprecated? In Rolling only, of course.

[MichaelOrlov]

@christophebedard Yes please. I would appreciate a follow-up PR.
During my previous code review, I had in mind deprecating some constructors and put them on a separate branch for follow-up. rolling...morlov/deprecate-some-player-ctor.
Please feel free to cherry-pick or consider taking them for your PR.