Skip to content

refactor(clp::streaming_compression): Use clp::Array in zstd decompressor to replace C-style arrays. #699

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LinZhihao-723 merged 1 commit into from
Jan 28, 2025
Merged

refactor(clp::streaming_compression): Use clp::Array in zstd decompressor to replace C-style arrays. #699

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 16 additions & 15 deletions components/core/src/clp/streaming_compression/zstd/Decompressor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,21 +3,20 @@
#include <algorithm>

#include "../../Defs.h"
#include "../../ErrorCode.hpp"
#include "../../ReadOnlyMemoryMappedFile.hpp"
#include "../../spdlog_with_specializations.hpp"
#include "../../TraceableException.hpp"

namespace clp::streaming_compression::zstd {
Decompressor::Decompressor() : ::clp::streaming_compression::Decompressor{CompressorType::ZSTD} {
m_decompression_stream = ZSTD_createDStream();
Decompressor::Decompressor()
: ::clp::streaming_compression::Decompressor{CompressorType::ZSTD},
m_decompression_stream{ZSTD_createDStream()},
m_unused_decompressed_stream_block_buffer{ZSTD_DStreamOutSize()} {
if (nullptr == m_decompression_stream) {
SPDLOG_ERROR("streaming_compression::zstd::Decompressor: ZSTD_createDStream() error");
throw OperationFailed(ErrorCode_Failure, __FILENAME__, __LINE__);
}

// Create block to hold unused decompressed data
m_unused_decompressed_stream_block_size = ZSTD_DStreamOutSize();
m_unused_decompressed_stream_block_buffer
= std::make_unique<char[]>(m_unused_decompressed_stream_block_size);
}

Decompressor::~Decompressor() {
Expand Down Expand Up @@ -50,9 +49,13 @@ ErrorCode Decompressor::try_read(char* buf, size_t num_bytes_to_read, size_t& nu
}
break;
case InputType::ReaderInterface: {
if (false == m_read_buffer.has_value()) {
throw OperationFailed(ErrorCode_Corrupt, __FILENAME__, __LINE__);
}
auto& read_buffer{m_read_buffer.value()};
Comment on lines +52 to +55

@gibber9809 gibber9809 Jan 28, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this check necessary? Seems like the open/close flow will force m_read_buffer.has_value() to always be true here.

@LinZhihao-723 LinZhihao-723 Jan 28, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is to:

  • Prevent future developers from breaking the assumption about the current open/close flow (especially we don't have unit tests for this code path atm)
  • Silent clang-tidy warnings

If any later benchmark shows this becomes the bottleneck, we can remove it by manually silent clang-tidy for line 55.

auto error_code = m_reader->try_read(
reinterpret_cast<char*>(m_read_buffer.get()),
m_read_buffer_capacity,
read_buffer.data(),
read_buffer.size(),
m_read_buffer_length
);
if (ErrorCode_Success != error_code) {
Expand Down Expand Up @@ -116,11 +119,11 @@ ErrorCode Decompressor::try_seek_from_begin(size_t pos) {
ErrorCode error;
while (m_decompressed_stream_pos < pos) {
size_t num_bytes_to_decompress = std::min(
m_unused_decompressed_stream_block_size,
m_unused_decompressed_stream_block_buffer.size(),
pos - m_decompressed_stream_pos
);
error = try_read_exact_length(
m_unused_decompressed_stream_block_buffer.get(),
m_unused_decompressed_stream_block_buffer.data(),
num_bytes_to_decompress
);
if (ErrorCode_Success != error) {
Expand Down Expand Up @@ -164,11 +167,10 @@ void Decompressor::open(ReaderInterface& reader, size_t read_buffer_capacity) {
throw OperationFailed(rc, __FILENAME__, __LINE__);
}

m_read_buffer_capacity = read_buffer_capacity;
m_read_buffer = std::make_unique<char[]>(m_read_buffer_capacity);
m_read_buffer.emplace(read_buffer_capacity);
m_read_buffer_length = 0;

m_compressed_stream_block = {m_read_buffer.get(), m_read_buffer_length, 0};
m_compressed_stream_block = {m_read_buffer->data(), m_read_buffer_length, 0};

reset_stream();
}
Expand All @@ -180,7 +182,6 @@ void Decompressor::close() {
break;
case InputType::ReaderInterface:
m_read_buffer.reset();
m_read_buffer_capacity = 0;
m_read_buffer_length = 0;
m_reader = nullptr;
break;
Expand Down
10 changes: 6 additions & 4 deletions components/core/src/clp/streaming_compression/zstd/Decompressor.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,12 @@
#define CLP_STREAMING_COMPRESSION_ZSTD_DECOMPRESSOR_HPP

#include <memory>
#include <optional>
#include <string>

#include <zstd.h>

#include "../../Array.hpp"
#include "../../ReaderInterface.hpp"
#include "../../ReadOnlyMemoryMappedFile.hpp"
#include "../../TraceableException.hpp"
Expand Down Expand Up @@ -128,15 +130,15 @@ class Decompressor : public ::clp::streaming_compression::Decompressor {
std::unique_ptr<ReadOnlyMemoryMappedFile> m_memory_mapped_file;
ReaderInterface* m_reader{nullptr};
size_t m_reader_initial_pos{0ULL};
std::unique_ptr<char[]> m_read_buffer;

std::optional<Array<char>> m_read_buffer;
size_t m_read_buffer_length{0ULL};
size_t m_read_buffer_capacity{0ULL};

ZSTD_inBuffer m_compressed_stream_block{};

size_t m_decompressed_stream_pos{0ULL};
size_t m_unused_decompressed_stream_block_size{0ULL};
std::unique_ptr<char[]> m_unused_decompressed_stream_block_buffer;

Array<char> m_unused_decompressed_stream_block_buffer;
};
} // namespace clp::streaming_compression::zstd
#endif // CLP_STREAMING_COMPRESSION_ZSTD_DECOMPRESSOR_HPP