diff --git a/components/core/src/clp/clg/CMakeLists.txt b/components/core/src/clp/clg/CMakeLists.txt index a0ca5e9d0..c3c8e3aea 100644 --- a/components/core/src/clp/clg/CMakeLists.txt +++ b/components/core/src/clp/clg/CMakeLists.txt @@ -130,7 +130,7 @@ target_compile_features(clg PRIVATE cxx_std_20) target_include_directories(clg PRIVATE "${PROJECT_SOURCE_DIR}/submodules") target_link_libraries(clg PRIVATE - Boost::filesystem Boost::iostreams Boost::program_options + Boost::filesystem Boost::program_options fmt::fmt log_surgeon::log_surgeon MariaDBClient::MariaDBClient diff --git a/components/core/src/clp/clo/CMakeLists.txt b/components/core/src/clp/clo/CMakeLists.txt index 931bffeaf..39cc72b60 100644 --- a/components/core/src/clp/clo/CMakeLists.txt +++ b/components/core/src/clp/clo/CMakeLists.txt @@ -158,7 +158,7 @@ target_compile_features(clo PRIVATE cxx_std_20) target_include_directories(clo PRIVATE "${PROJECT_SOURCE_DIR}/submodules") target_link_libraries(clo PRIVATE - Boost::filesystem Boost::iostreams Boost::program_options + Boost::filesystem Boost::program_options fmt::fmt log_surgeon::log_surgeon ${MONGOCXX_TARGET} diff --git a/components/core/src/clp/clp/CMakeLists.txt b/components/core/src/clp/clp/CMakeLists.txt index 53342f3a9..eff32ce46 100644 --- a/components/core/src/clp/clp/CMakeLists.txt +++ b/components/core/src/clp/clp/CMakeLists.txt @@ -171,7 +171,7 @@ target_compile_features(clp PRIVATE cxx_std_20) target_include_directories(clp PRIVATE "${PROJECT_SOURCE_DIR}/submodules") target_link_libraries(clp PRIVATE - Boost::filesystem Boost::iostreams Boost::program_options + Boost::filesystem Boost::program_options fmt::fmt log_surgeon::log_surgeon spdlog::spdlog diff --git a/components/core/src/clp/make_dictionaries_readable/CMakeLists.txt b/components/core/src/clp/make_dictionaries_readable/CMakeLists.txt index fd62a39fb..9779d137f 100644 --- a/components/core/src/clp/make_dictionaries_readable/CMakeLists.txt +++ b/components/core/src/clp/make_dictionaries_readable/CMakeLists.txt @@ -45,7 +45,7 @@ target_compile_features(make-dictionaries-readable PRIVATE cxx_std_20) target_include_directories(make-dictionaries-readable PRIVATE "${PROJECT_SOURCE_DIR}/submodules") target_link_libraries(make-dictionaries-readable PRIVATE - Boost::filesystem Boost::iostreams Boost::program_options + Boost::filesystem Boost::program_options log_surgeon::log_surgeon spdlog::spdlog clp::string_utils diff --git a/components/core/src/clp/streaming_archive/reader/Segment.cpp b/components/core/src/clp/streaming_archive/reader/Segment.cpp index aa43e1d1f..7732dc5f8 100644 --- a/components/core/src/clp/streaming_archive/reader/Segment.cpp +++ b/components/core/src/clp/streaming_archive/reader/Segment.cpp @@ -3,12 +3,16 @@ #include #include +#include #include #include +#include +#include "../../ErrorCode.hpp" #include "../../FileReader.hpp" #include "../../spdlog_with_specializations.hpp" +#include "../../TraceableException.hpp" using std::make_unique; using std::string; @@ -33,47 +37,37 @@ ErrorCode Segment::try_open(string const& segment_dir_path, segment_id_t segment return ErrorCode_Success; } - // Get the size of the compressed segment file - boost::system::error_code boost_error_code; - size_t segment_file_size = boost::filesystem::file_size(segment_path, boost_error_code); - if (boost_error_code) { - SPDLOG_ERROR( - "streaming_archive::reader::Segment: Unable to obtain file size for segment: " - "{}", - segment_path.c_str() - ); - SPDLOG_ERROR("streaming_archive::reader::Segment: {}", boost_error_code.message().c_str()); - return ErrorCode_Failure; - } - - // Sanity check: previously used memory mapped file should be closed before opening a new - // one - if (m_memory_mapped_segment_file.is_open()) { + // Sanity check: previously used memory mapped file should be closed before opening a new one + if (m_memory_mapped_segment_file.has_value()) { SPDLOG_WARN( "streaming_archive::reader::Segment: Previous segment should be closed before " "opening new one: {}", segment_path.c_str() ); - m_memory_mapped_segment_file.close(); + m_memory_mapped_segment_file.reset(); } - // Create read only memory mapped file - boost::iostreams::mapped_file_params memory_map_params; - memory_map_params.path = segment_path; - memory_map_params.flags = boost::iostreams::mapped_file::readonly; - memory_map_params.length = segment_file_size; - // Try to map it to the same memory location as the previous memory mapped file - memory_map_params.hint = m_memory_mapped_segment_file.data(); - m_memory_mapped_segment_file.open(memory_map_params); - if (!m_memory_mapped_segment_file.is_open()) { + + // Create read-only memory mapped file + try { + m_memory_mapped_segment_file.emplace(segment_path); + } catch (TraceableException const& ex) { + auto const error_code{ex.get_error_code()}; + auto const formatted_error{ + ErrorCode_errno == error_code + ? fmt::format("errno={}", errno) + : fmt::format("error_code={}, message={}", error_code, ex.what()) + }; SPDLOG_ERROR( "streaming_archive::reader:Segment: Unable to memory map the compressed " - "segment with path: {}", - segment_path.c_str() + "segment with path: {}. Error: {}", + segment_path.c_str(), + formatted_error ); return ErrorCode_Failure; } - m_decompressor.open(m_memory_mapped_segment_file.data(), segment_file_size); + auto const view{m_memory_mapped_segment_file.value().get_view()}; + m_decompressor.open(view.data(), view.size()); m_segment_path = segment_path; return ErrorCode_Success; @@ -82,7 +76,7 @@ ErrorCode Segment::try_open(string const& segment_dir_path, segment_id_t segment void Segment::close() { if (!m_segment_path.empty()) { m_decompressor.close(); - m_memory_mapped_segment_file.close(); + m_memory_mapped_segment_file.reset(); m_segment_path.clear(); } } diff --git a/components/core/src/clp/streaming_archive/reader/Segment.hpp b/components/core/src/clp/streaming_archive/reader/Segment.hpp index 9ed40ea60..cdcd51b4d 100644 --- a/components/core/src/clp/streaming_archive/reader/Segment.hpp +++ b/components/core/src/clp/streaming_archive/reader/Segment.hpp @@ -2,12 +2,12 @@ #define CLP_STREAMING_ARCHIVE_READER_SEGMENT_HPP #include +#include #include -#include - #include "../../Defs.h" #include "../../ErrorCode.hpp" +#include "../../ReadOnlyMemoryMappedFile.hpp" #include "../../streaming_compression/passthrough/Decompressor.hpp" #include "../../streaming_compression/zstd/Decompressor.hpp" #include "../Constants.hpp" @@ -53,7 +53,7 @@ class Segment { private: std::string m_segment_path; - boost::iostreams::mapped_file_source m_memory_mapped_segment_file; + std::optional m_memory_mapped_segment_file; #if USE_PASSTHROUGH_COMPRESSION streaming_compression::passthrough::Decompressor m_decompressor; diff --git a/components/core/tests/test-StreamingCompression.cpp b/components/core/tests/test-StreamingCompression.cpp index b43316b3f..747a38a05 100644 --- a/components/core/tests/test-StreamingCompression.cpp +++ b/components/core/tests/test-StreamingCompression.cpp @@ -1,10 +1,11 @@ +#include #include #include -#include #include #include +#include "../src/clp/ReadOnlyMemoryMappedFile.hpp" #include "../src/clp/streaming_compression/passthrough/Compressor.hpp" #include "../src/clp/streaming_compression/passthrough/Decompressor.hpp" #include "../src/clp/streaming_compression/zstd/Compressor.hpp" @@ -149,22 +150,10 @@ TEST_CASE("StreamingCompression", "[StreamingCompression]") { // Decompress // Memory map compressed file - // Create memory mapping for compressed_file_path, use boost read only memory mapped file - boost::system::error_code boost_error_code; - size_t compressed_file_size - = boost::filesystem::file_size(compressed_file_path, boost_error_code); - REQUIRE(!boost_error_code); - - boost::iostreams::mapped_file_params memory_map_params; - memory_map_params.path = compressed_file_path; - memory_map_params.flags = boost::iostreams::mapped_file::readonly; - memory_map_params.length = compressed_file_size; - boost::iostreams::mapped_file_source memory_mapped_compressed_file; - memory_mapped_compressed_file.open(memory_map_params); - REQUIRE(memory_mapped_compressed_file.is_open()); - + clp::ReadOnlyMemoryMappedFile const memory_mapped_compressed_file{compressed_file_path}; clp::streaming_compression::passthrough::Decompressor decompressor; - decompressor.open(memory_mapped_compressed_file.data(), compressed_file_size); + auto const compressed_file_view{memory_mapped_compressed_file.get_view()}; + decompressor.open(compressed_file_view.data(), compressed_file_view.size()); size_t uncompressed_bytes = 0; REQUIRE(ErrorCode_Success