Compress bag files in separate threads #506

pjreed · 2020-08-14T19:12:57Z

This offloads all compression into separate threads. Doing so helps
to prevent rosbag2 from dropping messages due to the main thread
being busy, and it also helps to improve performance by spreading
the work across multiple CPU cores. It uses a producer/consumer
model with a fixed number of pre-allocated threads that consume
messages or files off of an incoming queue.

This adds two new command line options:

--compression-queue-size
- The number of messages or files that can be sitting in the queue
  waiting for a thread to consume them. If the queue is full, older
  messages or files will be discarded; this would lead to either
  messages being discarded or files not being compressed.
  The default value is 1, which should be sufficient if the system
  is capable of keeping up with the work load.
--compression-threads
- The number of threads that can be compressing data at once. The
  default is 0, and values less than 1 will be interpreted to mean
  the number of concurrent threads supported by the current
  hardware.

Closes #274.

Distribution Statement A; OPSEC #4584

Signed-off-by: P. J. Reed [email protected]

This offloads all compression into separate threads. Doing so helps to prevent rosbag2 from dropping messages due to the main thread being busy, and it also helps to improve performance by spreading the work across multiple CPU cores. It uses a producer/consumer model with a fixed number of pre-allocated threads that consume messages or files off of an incoming queue. This adds two new command line options: - --compression-queue-size - The number of messages or files that can be sitting in the queue waiting for a thread to consume them. If the queue is full, older messages or files will be discarded; this would lead to either messages being discarded or files not being compressed. The default value is 1, which should be sufficient if the system is capable of keeping up with the work load. - --compression-threads - The number of threads that can be compressing data at once. The default is 0, and values less than 1 will be interpreted to mean the number of concurrent threads supported by the current hardware. Closes ros2#274. Distribution Statement A; OPSEC #2893 Signed-off-by: P. J. Reed <[email protected]>

Distribution Statement A; OPSEC #2893 Signed-off-by: P. J. Reed <[email protected]>

rosbag2_transport/src/rosbag2_transport/rosbag2_transport_python.cpp

rosbag2_compression/test/rosbag2_compression/test_sequential_compression_writer.cpp

rosbag2_compression/src/rosbag2_compression/zstd_compressor.cpp

rosbag2_compression/src/rosbag2_compression/sequential_compression_writer.cpp

Distribution Statement A; OPSEC #2893 Signed-off-by: P. J. Reed <[email protected]>

emersonknapp · 2020-09-04T23:59:26Z

High level - should the compression work queue be independently configurable from the writer queue? It would make sense to me that compression would work on the preexisting writer queue and would need to complete before submitting that queue to the storage impl.

pjreed · 2020-09-09T14:16:46Z

High level - should the compression work queue be independently configurable from the writer queue?

Hmm. I don't think I have a strong opinion one way or another on this; I can't think of any situations where you would need to configure them to be different values, but I also don't think it hurts anything to keep them separate. Although, is the writer queue size currently configurable? I just looked over the list of command line arguments and don't see anything that obviously maps to that.

pjreed · 2020-09-15T18:41:51Z

@emersonknapp Just poking this again, do you think anything else needs to be done?

…ronously

pjreed · 2020-10-05T15:15:49Z

Looks like another recent merge has introduce some conflicts with these changes, I'll see if I can get it sorted out...

emersonknapp · 2020-10-06T16:55:37Z

Sorry about that @pjreed i've been in and out of office - bad hand-off of responsibilities on my part. Change #526 went in to largely deduplicate the SequentialWriter vs SequentialCompressionWriter code - which will be better for all caching logic that goes in, since it was just written in two places before. That probably throws a wrench in this PR as it was written though. This could be good though to end up with something a little cleaner - the CompressionWriter can handle spinning up the thread and dispatching compression jobs to it, but not have to take care of just about anything else.

The new structure with get_writable_message is not asynchronous right now, though. Maybe that should be dispatched to a thread from the SequentialWriter side, then, in which case compression may not need to know it is running in a thread? Perhaps message converters would also benefit from being put off of the transport thread, though they are less overhead than compression.

…ronously Signed-off-by: P. J. Reed <[email protected]>

Signed-off-by: P. J. Reed <[email protected]>

pjreed · 2020-10-14T15:39:07Z

@emersonknapp Ok, I think this is ready to go now.

I spent a little while thinking about it, and I'm not sure that it makes a lot of sense to make the base SequentialWriter multi-threaded. It doesn't really provide it any benefit, and if anything, the added contention from having multiple threads trying to write at once could actually decrease performance. Even for subclasses that use threads, the way they handle offloading work into threads can vary from case to case; in the SequentialCompressionWriter's case, for example, it's actually got two different ways that it can use threads, since in per-message mode every message is compressed and written in a separate thread, while in per-file mode, the method to write messages to files is still single-threaded, but files are handed over to the compressor threads when it checks to see when it's time to split bag files.

ros2bag/ros2bag/verb/record.py

Also fix an issue that could cause compressor threads to hang when exiting Signed-off-by: P. J. Reed <[email protected]>

…ronously

Signed-off-by: P. J. Reed <[email protected]>

emersonknapp

This LGTM. Thanks for the performance benchmark investigation

rosbag2_compression/src/rosbag2_compression/sequential_compression_writer.cpp

rosbag2_compression/include/rosbag2_compression/sequential_compression_writer.hpp

Signed-off-by: P. J. Reed <[email protected]>

emersonknapp · 2020-10-26T21:56:30Z

Running this CI job
Gist: https://gist.githubusercontent.com/emersonknapp/86f84d73973ab681f5e23fe69e236da6/raw/e11f2391e32fb748418534862ada3b47c0871e82/ros2.repos
BUILD args: --packages-up-to ros2bag rosbag2_compression rosbag2_performance_writer_benchmarking rosbag2_transport rosbag2
TEST args: --packages-select ros2bag rosbag2_compression rosbag2_performance_writer_benchmarking rosbag2_transport
Job: ci_launcher

Linux
Linux-aarch64
macOS
Windows

…ronously Signed-off-by: P. J. Reed <[email protected]>

Signed-off-by: P. J. Reed <[email protected]>

pjreed · 2020-11-03T16:25:16Z

It looks like another merge caused a conflict, but it should be resolved now.

pjreed · 2020-11-09T18:22:57Z

Just pinging @emersonknapp about this again.

rosbag2_compression/src/rosbag2_compression/sequential_compression_writer.cpp

…ronously Signed-off-by: P. J. Reed <[email protected]>

emersonknapp · 2020-12-04T00:33:38Z

Left some comments on the discussion about write-order/determinism. Not blocking merging this, just adding context. I am going to try and resolve these merge conflicts now and see if I can get this merged this evening. Thanks for the patience and the important feature.

Signed-off-by: Emerson Knapp <[email protected]>

emersonknapp · 2020-12-04T03:41:25Z

Gist: https://gist.githubusercontent.com/emersonknapp/c901a3ec7225a38da9d08b156fca658e/raw/e9efac2d5f3ca1f383e469e9578f0b328a88a70e/ros2.repos
BUILD args: --packages-up-to ros2bag rosbag2_compression rosbag2_performance_writer_benchmarking rosbag2_transport rosbag2
TEST args: --packages-select ros2bag rosbag2_compression rosbag2_performance_writer_benchmarking rosbag2_transport rosbag2
Job: ci_launcher

Linux
Linux-aarch64
macOS
Windows

…ronously Signed-off-by: Emerson Knapp <[email protected]>

…ging Signed-off-by: Emerson Knapp <[email protected]>

emersonknapp · 2020-12-04T20:42:30Z

Rerun with OSX warning fix

Linux
Linux-aarch64
macOS
Windows

…tion variable Signed-off-by: Emerson Knapp <[email protected]>

emersonknapp · 2020-12-04T22:24:55Z

I think I've fixed the synchronization logic that the OSX test wast throwing on - we were not using the same mutex for calls to the condition variable, which is what you want to do if you are synchronizing access to a shared resource (in this case the message/file queue)

Linux
Linux-aarch64
macOS
Windows

adamdbrw · 2020-12-14T12:24:42Z

rosbag2_performance/rosbag2_performance_writer_benchmarking/src/writer_benchmark.cpp

+      compression_options);
+  } else {
+    writer_ = std::make_shared<rosbag2_cpp::writers::SequentialWriter>();
+  }
  writer_ = std::make_shared<rosbag2_cpp::writers::SequentialWriter>();


The writer is overwritten here, meaning compression is never applied

adamdbrw · 2020-12-15T12:29:31Z

...nce_writer_benchmarking/include/rosbag2_performance_writer_benchmarking/writer_benchmark.hpp

@@ -46,6 +47,11 @@ class WriterBenchmark : public rclcpp::Node
  std::vector<std::thread> producer_threads_;
  std::vector<std::unique_ptr<ByteProducer>> producers_;
  std::vector<std::shared_ptr<ByteMessageQueue>> queues_;
+
+  std::string compression_format_;
+  rosbag2_compression::CompressionMode compression_mode_;


this seems unused

* Compress bag files in separate threads This offloads all compression into separate threads. Doing so helps to prevent rosbag2 from dropping messages due to the main thread being busy, and it also helps to improve performance by spreading the work across multiple CPU cores. It uses a producer/consumer model with a fixed number of pre-allocated threads that consume messages or files off of an incoming queue. This adds two new command line options: - --compression-queue-size - The number of messages or files that can be sitting in the queue waiting for a thread to consume them. If the queue is full, older messages or files will be discarded; this would lead to either messages being discarded or files not being compressed. The default value is 1, which should be sufficient if the system is capable of keeping up with the work load. - --compression-threads - The number of threads that can be compressing data at once. The default is 0, and values less than 1 will be interpreted to mean the number of concurrent threads supported by the current hardware. Closes #274. Distribution Statement A; OPSEC #2893 Signed-off-by: P. J. Reed <[email protected]> Co-authored-by: Emerson Knapp <[email protected]>

pjreed added 2 commits August 14, 2020 14:11

Remove unused import

af93ce1

Distribution Statement A; OPSEC #2893 Signed-off-by: P. J. Reed <[email protected]>

piraka9011 reviewed Aug 14, 2020

View reviewed changes

rosbag2_transport/src/rosbag2_transport/rosbag2_transport_python.cpp Outdated Show resolved Hide resolved

piraka9011 reviewed Aug 14, 2020

View reviewed changes

rosbag2_compression/test/rosbag2_compression/test_sequential_compression_writer.cpp Outdated Show resolved Hide resolved

piraka9011 reviewed Aug 14, 2020

View reviewed changes

rosbag2_compression/src/rosbag2_compression/zstd_compressor.cpp Show resolved Hide resolved

piraka9011 reviewed Aug 14, 2020

View reviewed changes

rosbag2_compression/src/rosbag2_compression/sequential_compression_writer.cpp Show resolved Hide resolved

Make recommended changes

0e1a1ea

Distribution Statement A; OPSEC #2893 Signed-off-by: P. J. Reed <[email protected]>

piraka9011 requested review from Karsten1987 and emersonknapp August 18, 2020 14:11

piraka9011 assigned pjreed Aug 18, 2020

Merge remote-tracking branch 'origin/master' into 274/compress-asynch…

9a73937

…ronously

emersonknapp mentioned this pull request Oct 6, 2020

SequentialWriter to cache by message size instead of message count #530

Merged

pjreed added 2 commits October 13, 2020 14:42

Merge remote-tracking branch 'origin/master' into 274/compress-asynch…

fc4caa5

…ronously Signed-off-by: P. J. Reed <[email protected]>

Merge remote-tracking branch 'origin/master' into 274/compress-asynch…

efe7242

…ronously Signed-off-by: P. J. Reed <[email protected]>

pjreed requested a review from a team as a code owner October 13, 2020 19:44

Fix lint issues

1e527a4

Signed-off-by: P. J. Reed <[email protected]>

pjreed marked this pull request as draft October 13, 2020 20:08

pjreed added 5 commits October 13, 2020 15:31

Refactor compression code

1090792

Signed-off-by: P. J. Reed <[email protected]>

Fix unit tests

e3efb53

Signed-off-by: P. J. Reed <[email protected]>

uncrustify

44074d5

Signed-off-by: P. J. Reed <[email protected]>

Fix segfault

6d8fa60

Signed-off-by: P. J. Reed <[email protected]>

uncrustify

99cf6da

Signed-off-by: P. J. Reed <[email protected]>

pjreed marked this pull request as ready for review October 14, 2020 15:39

dabonnie reviewed Oct 14, 2020

View reviewed changes

ros2bag/ros2bag/verb/record.py Show resolved Hide resolved

pjreed added 3 commits October 14, 2020 17:13

Add compression support to rosbag2_performance_writer_benchmarking

23f3f7d

Also fix an issue that could cause compressor threads to hang when exiting Signed-off-by: P. J. Reed <[email protected]>

Merge remote-tracking branch 'origin/master' into 274/compress-asynch…

65a3c1b

…ronously

Move data instead of copying it

f97978d

Signed-off-by: P. J. Reed <[email protected]>

pjreed marked this pull request as ready for review October 16, 2020 15:33

pjreed mentioned this pull request Oct 20, 2020

Use ZSTD's streaming interface for compressing files #543

Merged

emersonknapp approved these changes Oct 23, 2020

View reviewed changes

rosbag2_compression/src/rosbag2_compression/sequential_compression_writer.cpp Outdated Show resolved Hide resolved

rosbag2_compression/include/rosbag2_compression/sequential_compression_writer.hpp Outdated Show resolved Hide resolved

Address comments in PR

15e03b3

Signed-off-by: P. J. Reed <[email protected]>

pjreed added 4 commits November 2, 2020 16:05

Merge remote-tracking branch 'origin/master' into 274/compress-asynch…

1b7a832

…ronously Signed-off-by: P. J. Reed <[email protected]>

Build fixes

3439083

Signed-off-by: P. J. Reed <[email protected]>

More fixes

5a3fc52

Signed-off-by: P. J. Reed <[email protected]>

Uncrustify

4586d74

Signed-off-by: P. J. Reed <[email protected]>

murbanga reviewed Nov 24, 2020

View reviewed changes

rosbag2_compression/src/rosbag2_compression/sequential_compression_writer.cpp Show resolved Hide resolved

Merge remote-tracking branch 'origin/master' into 274/compress-asynch…

74cfe26

…ronously Signed-off-by: P. J. Reed <[email protected]>

Merge branch 'master' into 274/compress-asynchronously

e16838b

Signed-off-by: Emerson Knapp <[email protected]>

Emerson Knapp added 2 commits December 4, 2020 10:19

Merge remote-tracking branch 'origin/master' into 274/compress-asynch…

d411579

…ronously Signed-off-by: Emerson Knapp <[email protected]>

Avoid OSX vs Linux int-type formatting difference by using stream log…

5a0bfc9

…ging Signed-off-by: Emerson Knapp <[email protected]>

Fix locking logic to use the same mutex across all waits to the condi…

ff26a0c

…tion variable Signed-off-by: Emerson Knapp <[email protected]>

emersonknapp merged commit 2d45abb into ros2:master Dec 4, 2020

adamdbrw reviewed Dec 14, 2020

View reviewed changes

adamdbrw mentioned this pull request Dec 14, 2020

Updating performance writer #570

Merged

adamdbrw reviewed Dec 15, 2020

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Compress bag files in separate threads #506

Compress bag files in separate threads #506

pjreed commented Aug 14, 2020 •

edited

Loading

emersonknapp commented Sep 4, 2020

pjreed commented Sep 9, 2020

pjreed commented Sep 15, 2020

pjreed commented Oct 5, 2020

emersonknapp commented Oct 6, 2020

pjreed commented Oct 14, 2020

emersonknapp left a comment

emersonknapp commented Oct 26, 2020 •

edited

Loading

pjreed commented Nov 3, 2020

pjreed commented Nov 9, 2020

emersonknapp commented Dec 4, 2020

emersonknapp commented Dec 4, 2020

emersonknapp commented Dec 4, 2020

emersonknapp commented Dec 4, 2020

adamdbrw Dec 14, 2020

adamdbrw Dec 15, 2020

Compress bag files in separate threads #506

Compress bag files in separate threads #506

Conversation

pjreed commented Aug 14, 2020 • edited Loading

emersonknapp commented Sep 4, 2020

pjreed commented Sep 9, 2020

pjreed commented Sep 15, 2020

pjreed commented Oct 5, 2020

emersonknapp commented Oct 6, 2020

pjreed commented Oct 14, 2020

emersonknapp left a comment

Choose a reason for hiding this comment

emersonknapp commented Oct 26, 2020 • edited Loading

pjreed commented Nov 3, 2020

pjreed commented Nov 9, 2020

emersonknapp commented Dec 4, 2020

emersonknapp commented Dec 4, 2020

emersonknapp commented Dec 4, 2020

emersonknapp commented Dec 4, 2020

adamdbrw Dec 14, 2020

Choose a reason for hiding this comment

adamdbrw Dec 15, 2020

Choose a reason for hiding this comment

pjreed commented Aug 14, 2020 •

edited

Loading

emersonknapp commented Oct 26, 2020 •

edited

Loading