[Compression - 9] Enable writer to compress files/messages #250
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,15 +15,16 @@ | |
| #include "rosbag2_cpp/writers/sequential_writer.hpp" | ||
|
|
||
| #include <algorithm> | ||
| #include <cassert> | ||
| #include <chrono> | ||
| #include <memory> | ||
| #include <stdexcept> | ||
| #include <string> | ||
| #include <utility> | ||
|
|
||
| #include "rcpputils/filesystem_helper.hpp" | ||
| #include "rcutils/filesystem.h" | ||
| #include "rosbag2_compression/zstd_compressor.hpp" | ||
|
Collaborator
There was a problem hiding this comment. new line above to group per package. |
||
|
|
||
| #include "rosbag2_cpp/info.hpp" | ||
| #include "rosbag2_cpp/storage_options.hpp" | ||
|
|
@@ -47,6 +48,7 @@ std::string format_storage_uri(const std::string & base_folder, uint64_t storage | |
| } | ||
| } // namespace | ||
|
|
||
| // TODO(piraka9011) Initialize defaults in header file instead. | ||
|
piraka9011 marked this conversation as resolved.
|
||
| SequentialWriter::SequentialWriter( | ||
| std::unique_ptr<rosbag2_storage::StorageFactoryInterface> storage_factory, | ||
| std::shared_ptr<SerializationFormatConverterFactoryInterface> converter_factory, | ||
|
|
@@ -56,16 +58,34 @@ SequentialWriter::SequentialWriter( | |
| storage_(nullptr), | ||
| metadata_io_(std::move(metadata_io)), | ||
| converter_(nullptr), | ||
| compressor_{nullptr}, | ||
| max_bagfile_size_(rosbag2_storage::storage_interfaces::MAX_BAGFILE_SIZE_NO_SPLIT), | ||
| topics_names_to_info_(), | ||
| metadata_(), | ||
| compression_mode_{CompressionMode::NONE} {} | ||
|
|
||
|
|
||
| SequentialWriter::~SequentialWriter() | ||
| { | ||
| reset(); | ||
| } | ||
|
|
||
| void SequentialWriter::init_compression(const CompressionOptions & compression_options) | ||
|
piraka9011 marked this conversation as resolved.
|
||
| { | ||
| if (compression_options.compression_mode != rosbag2_cpp::CompressionMode::NONE) { | ||
| if (compression_options.compression_format == "zstd") { | ||
| compressor_ = std::make_unique<rosbag2_compression::ZstdCompressor>(); | ||
|
Collaborator
There was a problem hiding this comment. I am having a problem with this line as it's imposing a hard dependency on
Contributor
Author
There was a problem hiding this comment. We have this issue on our backlog next to make compression a plugin instead so that its not hardcoded. Do you have any other suggestions in the meantime on how to avoid the hard dependency? auto zstd_compressor = std::make_unique<rosbag2_compression::ZstdCompressor>();
compressor_ = std::make_unique<rosbag2_compression::BaseCompressorInterface>(std::move(zstd_compressor));If not, I suggest we work with the current implementation and then after introducing the plugin changes to
Collaborator
There was a problem hiding this comment. Not sure I get your proposition right. We can either use a factory for loading it as a plugin dynamically or as discussed in a previous PR, inherit from the The moment we include a compression header in the rosbag2_cpp package we introduce the dependency. That being said, the current PR also lacks the dependency of
Contributor
Author
There was a problem hiding this comment. If I understand correctly, you prefer to isolate compression from One option I see is compressing at the transport layer instead, though I'm not sure we want to do that.
Collaborator
There was a problem hiding this comment. Introducing the compression functionality is not the problem. So things like compression_options and general compression related structs might very well be living in this package. However, hard coding ztd is giving people no change to not compile zstd and its vendor package etc. This might not suit very well to every use case.
Contributor
Author
There was a problem hiding this comment.
I agree, which is why we want to work on making compression a plugin and work like However, since this is the first iteration, we want to make sure the whole compression pipeline works with a single compression format before extending it to other formats though a plugin. P.S. Accidentally edited your comment, my bad.
Collaborator
There was a problem hiding this comment.
That's one way to steer the direction of aa conversation ;-)
That's fine. I think having a compression writer/reader implementation within
Contributor
Author
There was a problem hiding this comment.
What about splitting then? Basically implementing a |
||
| } else { | ||
| std::stringstream err; | ||
| err << "Unsupported compression format " << compression_options.compression_format; | ||
| throw std::invalid_argument{err.str()}; | ||
| } | ||
| } | ||
| metadata_.compression_format = compression_options.compression_format; | ||
| metadata_.compression_mode = | ||
| rosbag2_cpp::compression_mode_to_string(compression_options.compression_mode); | ||
| } | ||
|
|
||
| void SequentialWriter::init_metadata() | ||
| { | ||
| metadata_ = rosbag2_storage::BagMetadata{}; | ||
|
|
@@ -77,10 +97,12 @@ void SequentialWriter::init_metadata() | |
|
|
||
| void SequentialWriter::open( | ||
| const StorageOptions & storage_options, | ||
| const ConverterOptions & converter_options, | ||
| const CompressionOptions & compression_options) | ||
| { | ||
| max_bagfile_size_ = storage_options.max_bagfile_size; | ||
| base_folder_ = storage_options.uri; | ||
| compression_mode_ = compression_options.compression_mode; | ||
|
|
||
| if (converter_options.output_serialization_format != | ||
| converter_options.input_serialization_format) | ||
|
|
@@ -89,25 +111,28 @@ void SequentialWriter::open( | |
| } | ||
|
|
||
| const auto storage_uri = format_storage_uri(base_folder_, 0); | ||
| storage_ = storage_factory_->open_read_write(storage_uri, storage_options.storage_id); | ||
| if (!storage_) { | ||
| throw std::runtime_error("No storage could be initialized. Abort"); | ||
| } | ||
|
|
||
| if (storage_options.max_bagfile_size != 0 && | ||
| storage_options.max_bagfile_size < storage_->get_minimum_split_file_size()) | ||
| { | ||
| throw std::invalid_argument{ | ||
| "Invalid bag splitting size given. Please provide a different value."}; | ||
| } | ||
|
|
||
| init_metadata(); | ||
| init_compression(compression_options); | ||
|
mm318 marked this conversation as resolved.
|
||
| } | ||
|
|
||
| void SequentialWriter::reset() | ||
| { | ||
| if (!base_folder_.empty()) { | ||
| if (compressor_ && compression_mode_ == CompressionMode::FILE) { | ||
|
piraka9011 marked this conversation as resolved.
Outdated
|
||
| compress_last_file(); | ||
| } | ||
| finalize_metadata(); | ||
| metadata_io_->write_metadata(base_folder_, metadata_); | ||
| } | ||
|
|
@@ -163,17 +188,26 @@ void SequentialWriter::remove_topic(const rosbag2_storage::TopicMetadata & topic | |
| } | ||
| } | ||
|
|
||
| void SequentialWriter::compress_last_file() | ||
| { | ||
| assert(compressor_ != nullptr); | ||
| metadata_.relative_file_paths.back() = | ||
| compressor_->compress_uri(metadata_.relative_file_paths.back()); | ||
| } | ||
|
|
||
| void SequentialWriter::split_bagfile() | ||
| { | ||
| if (compression_mode_ == CompressionMode::FILE) { | ||
| compress_last_file(); | ||
| } | ||
|
|
||
| const auto storage_uri = format_storage_uri( | ||
| base_folder_, | ||
| metadata_.relative_file_paths.size()); | ||
| storage_ = storage_factory_->open_read_write(storage_uri, metadata_.storage_identifier); | ||
| if (!storage_) { | ||
| std::stringstream errmsg; | ||
| errmsg << "Failed to rollover bagfile to new file: \"" << storage_uri << "\"!"; | ||
| throw std::runtime_error(errmsg.str()); | ||
| } | ||
|
|
||
|
|
@@ -185,6 +219,14 @@ void SequentialWriter::split_bagfile() | |
| } | ||
| } | ||
|
|
||
| void SequentialWriter::compress_message( | ||
| std::shared_ptr<rosbag2_storage::SerializedBagMessage> message) | ||
| { | ||
| assert(compressor_ != nullptr); | ||
| auto converted_message = converter_ ? converter_->convert(message) : message; | ||
| compressor_->compress_serialized_bag_message(converted_message.get()); | ||
| } | ||
|
|
||
| void SequentialWriter::write(std::shared_ptr<rosbag2_storage::SerializedBagMessage> message) | ||
| { | ||
| if (!storage_) { | ||
|
|
@@ -205,6 +247,10 @@ void SequentialWriter::write(std::shared_ptr<rosbag2_storage::SerializedBagMessa | |
| const auto duration = message_timestamp - metadata_.starting_time; | ||
| metadata_.duration = std::max(metadata_.duration, duration); | ||
|
|
||
| if (compression_mode_ == CompressionMode::MESSAGE) { | ||
| compress_message(message); | ||
| } | ||
|
|
||
| storage_->write(converter_ ? converter_->convert(message) : message); | ||
| } | ||
|
|
||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.