Use thread pool to run benchmark publishers

[carlossvg]

Related to: #688

Depends on #1153

This PR changes benchmark_publishers to use a thread pool instead of running each publisher in separate threads. The motivation for this change is to overcome a limitation on the number of threads imposed by Cyclone DDS (limited to 128 threads).

[MichaelOrlov]

cc: @james-rms

[james-rms]

This PR adds a source of confusion - previously the number of threads was controlled by the producers configuration, now it's controlled by hardware_concurrency. From the description it's not clear why this is a neccessary feature.

Why run so many publishers, rather than publishing more messages from each individual publisher?

[carlossvg]

Thanks for the review.

This PR adds a source of confusion - previously the number of threads was controlled by the producers configuration, now it's controlled by hardware_concurrency. From the description it's not clear why this is a neccessary feature.

We could expose this parameter to make it configurable. For the default value I see tow options:

1. hardware_concurrency: this is the same option used by MultiThreadedExecutor
2. Number of thread equal to the number of publishers: This would be similar to what we have now

Why run so many publishers, rather than publishing more messages from each individual publisher?

The motivation for us is to be able to benchmark exactly a specific real setup. Publishing more messages from each individual publisher would be a different scenario in terms of resources, queues, concurrency, etc.

[MichaelOrlov]

@carlossvg Could you please reiterate on this PR and rebase it on top of latest rolling?

[MichaelOrlov]

@carlossvg Thanks for the PR and rebasing. I like the idea with tread pool.
However I have a number of suggestions and findings for improvements.

Side note: It was a conflicts in CMake file, I've tried to resolve it but git somehow managed to add one commit from baseline Rolling branch above your changes. Sorry for mess.
Feel free to rebase.

[carlossvg]

@MichaelOrlov I think there are a couple of things left to do:

1. Document the new added option number_of_threads
2. Updated writer_benchmark to use the thread pool so the behavior is consistent with benchmark_publishers
3. Compare benchmark results before and after the changes to double-check there are no big differences

[MichaelOrlov]

@carlossvg As regards to the

Updated writer_benchmark to use the thread pool so the behavior is consistent with benchmark_publishers

I have been thinking about it yesterday. Although come to the conclusion that it is not necessary and would be useless by design.
Because:

1. We can configure number of writers in config file and there are no differences in behavior if it would be less threads or less writers in config file. It make sense to have current approach when always one thread per one writer to it's own queue.
2. Allowing to have less threads than number of writers will not give us any advantages in testing or simulation. It will introduce some uncertainty where we want to avoid it. We usually using writer_benchmark scenario when we need to measure if rosbag2 can keep up in speed with some data stream but without transport layer. Reducing number of threads will impose uncertainty in this case for the volume of the data written to the rosbag2 backend per seconds. Which we are trying to strictly define via number of writers and message size with frame rate in config file.

Update:
However we can update writer_benchmark just for consistency since we will have a new config parameter number_of_threads. Or add a warning when number_of_threads parameter more than 0 and not equal to the number of the writers. That number_of_threads parameter will be ignored since not supported with writer_benchmark

[MichaelOrlov]

@carlossvg Semantically PR looks good for me.
Although I tried to test performance benchmarking with your branch and found a few issues.
Could you please fix them?

1. Need remove std::this_thread::sleep_for(1ms); in void WriterBenchmark::start_benchmark() it basically usles and could affect reproduction rate for figh frequency topics.
2. Need change topic.type from topic.type = "std_msgs::msgs::ByteMultiArray"; to the "rosbag2_performance_benchmarking_msgs/msg/ByteArray"; The writer_benchmark crashes without this fix if start benchmarking with no_transport = True.
3. It would be nice to add number_of_threads: 16 or 50 to some of the producers config files to show users in what exact place it's expected to be.

[carlossvg]

1. Need remove std::this_thread::sleep_for(1ms); in void WriterBenchmark::start_benchmark() it basically usles and could affect reproduction rate for figh frequency topics.

If we do that the loop becomes a busy wait loop while waiting on new messages, right?

As pointed out by this comment the proper fix would be to use a conditional variable to implement the wait.

// TODO(adamdbrw) Performance can be improved. Use conditional variables

I would implement that change in a separate MR.

2. Need change topic.type from topic.type = "std_msgs::msgs::ByteMultiArray"; to the >"rosbag2_performance_benchmarking_msgs/msg/ByteArray"; The writer_benchmark crashes without this fix if start >benchmarking with no_transport = True.

Good catch, fixed

3. It would be nice to add number_of_threads: 16 or 50 to some of the producers config files to show users in what exact place it's expected to be.

I order to make more explicit this parameter exists I added # number_of_threads: 16 commented in all producer config files. If we prefer not to include them as a comment we can create a new config file with this option enabled.

[MichaelOrlov]

@carlossvg We can fix issue with delay in separate PR - I agree.
However delimiters in topic type is expected to be with slashes instead of the ::

[MichaelOrlov]

@carlossvg Could you please rebase your branch? There are merge conflicts.

[MichaelOrlov]

@carlossvg It seems something went wrong with your rebase.
I see zero commits on your branch and no delta was merged.
Also please note that before merging to the parent ros2::rolling we need to run CI to make sure that we will not break anything.
I found reference to your latest commit in remote git log https://github.com/carlossvg/rosbag2/tree/eb1fd354fc3d4c5a484fb6891e1a7be50448a2b7
Although I am not able to check it out or clone it.
May be you will be able to do it on your local git repo by doing git checkout eb1fd354fc3d4c5a484fb6891e1a7be50448a2b7 or git reset --hard eb1fd354fc3d4c5a484fb6891e1a7be50448a2b7

[carlossvg]

@MichaelOrlov It looks like the PR was merged automatically when syncing the fork. I created a new PR here #1250