From e4906f626611cfe2a68a70cfefbd673bf7064651 Mon Sep 17 00:00:00 2001 From: Andreas Stefl Date: Sat, 29 Aug 2026 19:59:55 +0200 Subject: [PATCH] refactor(zip): make the archive's read callback re-entrant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `util::Archive` carried a `mutable std::mutex` that every operation took — `is_file`, `path`, `method`, `size`, opening a stream, and each 4 KiB `underflow`. It existed for one reason: the read callback drove a single shared `std::istream` with `seekg`/`read`, so two readers would fight over its position. miniz needs none of that. It keeps decompression state per iterator and addresses the input by absolute offset, so the callback only has to be re-entrant. `ReadSource` makes it so: a memory-backed file is read straight out of its buffer with no lock, and anything else takes a stream from a free list, reads, and puts it back — the lock covers the list, never the read. The archive-wide mutex is gone with it, so the metadata calls no longer serialise either. `mz_zip_archive::m_last_error` stays racy, as miniz documents; we never read it. `concurrent_entry_reads` reads eight 64 KiB entries from eight threads, over a disk file and a memory file. Against a shared unguarded stream it fails hard with `FileNotFound`, so it has teeth. --- CHANGELOG.md | 3 + src/odr/internal/zip/zip_util.cpp | 81 ++++++++++++++-------- src/odr/internal/zip/zip_util.hpp | 28 ++++++-- test/src/internal/zip/zip_archive_test.cpp | 50 +++++++++++++ 4 files changed, 131 insertions(+), 31 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 602a3dd29..88f733141 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,9 @@ The release run heads these entries with the version and opens a fresh ## Unreleased +- Two entries of the same zip document can be read at once; nothing serialises + on a single lock any more. + - The zip backend is `miniz/3.1.1`, up from `3.0.2`. Rendered output is unchanged. diff --git a/src/odr/internal/zip/zip_util.cpp b/src/odr/internal/zip/zip_util.cpp index 2fa6d2c15..2eb524278 100644 --- a/src/odr/internal/zip/zip_util.cpp +++ b/src/odr/internal/zip/zip_util.cpp @@ -6,6 +6,7 @@ #include #include +#include namespace odr::internal::zip::util { @@ -33,8 +34,6 @@ class ReaderBuffer final : public std::streambuf { return traits_type::eof(); } - std::lock_guard lock(m_archive->mutex()); - const std::uint64_t amount = std::min(m_remaining, m_buffer.size()); const std::uint32_t result = @@ -83,7 +82,6 @@ class FileInZip final : public abstract::File { return m_archive->file()->location(); } [[nodiscard]] std::size_t size() const override { - std::lock_guard lock(m_archive->mutex()); mz_zip_archive_file_stat stat{}; mz_zip_reader_file_stat(m_archive->zip(), m_index, &stat); return stat.m_uncomp_size; @@ -97,7 +95,6 @@ class FileInZip final : public abstract::File { } [[nodiscard]] std::unique_ptr stream() const override { - std::lock_guard lock(m_archive->mutex()); if (mz_zip_reader_is_file_encrypted(m_archive->zip(), m_index)) { throw UnsupportedOperation("cannot read encrypted zip entry"); } @@ -120,17 +117,14 @@ class FileInZip final : public abstract::File { } // namespace bool Archive::Entry::is_file() const { - std::lock_guard lock(m_archive->mutex()); return !mz_zip_reader_is_file_a_directory(m_archive->zip(), m_index); } bool Archive::Entry::is_directory() const { - std::lock_guard lock(m_archive->mutex()); return mz_zip_reader_is_file_a_directory(m_archive->zip(), m_index); } RelPath Archive::Entry::path() const { - std::lock_guard lock(m_archive->mutex()); std::array filename{}; mz_zip_reader_get_filename(m_archive->zip(), m_index, filename.data(), static_cast(filename.size())); @@ -138,7 +132,6 @@ RelPath Archive::Entry::path() const { } Method Archive::Entry::method() const { - std::lock_guard lock(m_archive->mutex()); mz_zip_archive_file_stat stat{}; mz_zip_reader_file_stat(m_archive->zip(), m_index, &stat); switch (stat.m_method) { @@ -158,21 +151,63 @@ std::shared_ptr Archive::Entry::file() const { return std::make_shared(m_archive->shared_from_this(), m_index); } -Archive::Archive(std::shared_ptr file) +ReadSource::ReadSource(std::shared_ptr file) : m_file{std::move(file)} { if (m_file == nullptr) { - throw NullPointerError("Archive: file is nullptr"); + throw NullPointerError("ReadSource: file is nullptr"); } - m_stream = m_file->stream(); - open_from_file(m_zip, *m_file, *m_stream); + m_memory = m_file->memory_data(); } -Archive::~Archive() { - std::lock_guard lock(m_mutex); - mz_zip_end(&m_zip); +std::size_t ReadSource::read(const std::uint64_t offset, void *buffer, + const std::size_t size) const { + if (m_memory.has_value()) { + if (offset >= m_memory->size()) { + return 0; + } + const std::size_t amount = + std::min(size, m_memory->size() - offset); + std::memcpy(buffer, m_memory->data() + offset, amount); + return amount; + } + + std::unique_ptr stream; + { + std::lock_guard lock(m_mutex); + if (!m_streams.empty()) { + stream = std::move(m_streams.back()); + m_streams.pop_back(); + } + } + if (stream == nullptr) { + stream = m_file->stream(); + } + + // A short read has to surface as one. Clear first so an earlier read past the + // end does not poison every later seek. + stream->clear(); + stream->seekg(static_cast(offset)); + stream->read(static_cast(buffer), static_cast(size)); + const auto result = static_cast(stream->gcount()); + + { + std::lock_guard lock(m_mutex); + m_streams.push_back(std::move(stream)); + } + + return result; +} + +Archive::Archive(std::shared_ptr file) + : m_file{std::move(file)} { + if (m_file == nullptr) { + throw NullPointerError("Archive: file is nullptr"); + } + m_source = std::make_unique(m_file); + open_from_file(m_zip, *m_file, *m_source); } -std::mutex &Archive::mutex() const { return m_mutex; } +Archive::~Archive() { mz_zip_end(&m_zip); } mz_zip_archive *Archive::zip() const { return &m_zip; } @@ -183,7 +218,6 @@ std::shared_ptr Archive::file() const noexcept { Archive::Iterator Archive::begin() const { return {*this, 0}; } Archive::Iterator Archive::end() const { - std::lock_guard lock(m_mutex); return {*this, mz_zip_reader_get_num_files(&m_zip)}; } @@ -198,18 +232,11 @@ Archive::Iterator Archive::find(const RelPath &path) const { namespace odr::internal::zip { void util::open_from_file(mz_zip_archive &archive, const abstract::File &file, - std::istream &stream) { - archive.m_pIO_opaque = &stream; + ReadSource &source) { + archive.m_pIO_opaque = &source; archive.m_pRead = [](void *opaque, const std::uint64_t offset, void *buffer, const std::size_t size) { - const auto in = static_cast(opaque); - // Reporting `size` regardless would hand miniz whatever was already in the - // buffer; a short read has to surface as one. Clear first so an earlier - // read past the end does not poison every later seek. - in->clear(); - in->seekg(static_cast(offset)); - in->read(static_cast(buffer), static_cast(size)); - return static_cast(in->gcount()); + return static_cast(opaque)->read(offset, buffer, size); }; const bool state = mz_zip_reader_init( &archive, file.size(), MZ_ZIP_FLAG_DO_NOT_SORT_CENTRAL_DIRECTORY); diff --git a/src/odr/internal/zip/zip_util.hpp b/src/odr/internal/zip/zip_util.hpp index 329d13e54..ff17ad13f 100644 --- a/src/odr/internal/zip/zip_util.hpp +++ b/src/odr/internal/zip/zip_util.hpp @@ -6,7 +6,10 @@ #include #include #include +#include #include +#include +#include #include #include @@ -23,12 +26,30 @@ enum class Method { DEFLATED, }; +/// Re-entrant source for `mz_zip_archive::m_pRead`, which reads by absolute +/// offset. The lock covers the stream free list, never a read. +class ReadSource final { +public: + explicit ReadSource(std::shared_ptr file); + + [[nodiscard]] std::size_t read(std::uint64_t offset, void *buffer, + std::size_t size) const; + +private: + std::shared_ptr m_file; + std::optional m_memory; + + mutable std::mutex m_mutex; + mutable std::vector> m_streams; +}; + +/// Entries can be read concurrently. `mz_zip_archive::m_last_error` cannot, and +/// is never read. class Archive final : public std::enable_shared_from_this { public: explicit Archive(std::shared_ptr file); ~Archive(); - [[nodiscard]] std::mutex &mutex() const; [[nodiscard]] mz_zip_archive *zip() const; [[nodiscard]] std::shared_ptr file() const noexcept; @@ -101,14 +122,13 @@ class Archive final : public std::enable_shared_from_this { private: std::shared_ptr m_file; - std::unique_ptr m_stream; + std::unique_ptr m_source; - mutable std::mutex m_mutex; mutable mz_zip_archive m_zip{}; }; void open_from_file(mz_zip_archive &archive, const abstract::File &file, - std::istream &stream); + ReadSource &source); /// `archive`'s write callback has to honour the offset it is given — local /// headers are rewritten with the entry size. diff --git a/test/src/internal/zip/zip_archive_test.cpp b/test/src/internal/zip/zip_archive_test.cpp index 9b0187df8..0cc565fe2 100644 --- a/test/src/internal/zip/zip_archive_test.cpp +++ b/test/src/internal/zip/zip_archive_test.cpp @@ -14,6 +14,7 @@ #include #include #include +#include #include using namespace odr; @@ -129,6 +130,55 @@ TEST(ZipArchive, create_order) { } } +/// The read callback has to be re-entrant, for a memory and a stream source. +TEST(ZipArchive, concurrent_entry_reads) { + const std::string path = + (std::filesystem::current_path() / "concurrent.zip").string(); + constexpr std::size_t entry_count = 8; + + std::vector expected; + expected.reserve(entry_count); + for (std::size_t i = 0; i < entry_count; ++i) { + // large enough that a single entry spans many reads + expected.emplace_back(64 * 1024, static_cast('a' + i)); + } + + { + ZipArchive zip; + for (std::size_t i = 0; i < entry_count; ++i) { + zip.insert_file(std::end(zip), RelPath("entry" + std::to_string(i)), + std::make_shared(expected[i])); + } + std::ofstream out(path, std::ios::binary); + zip.save(out); + } + + const std::vector> sources{ + std::make_shared(path), + std::make_shared(DiskFile(path))}; + + for (const auto &source : sources) { + const auto zip = std::make_shared(source); + + std::vector actual(entry_count); + std::vector threads; + threads.reserve(entry_count); + for (std::size_t i = 0; i < entry_count; ++i) { + threads.emplace_back([&zip, &actual, i] { + const auto entry = zip->find(RelPath("entry" + std::to_string(i))); + const auto stream = entry->file()->stream(); + actual[i] = std::string(std::istreambuf_iterator(*stream), + std::istreambuf_iterator()); + }); + } + for (auto &thread : threads) { + thread.join(); + } + + EXPECT_EQ(expected, actual); + } +} + namespace { std::uint32_t read_le(const std::string &data, const std::size_t offset,