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,