refactor(zip): make the archive's read callback re-entrant

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.
 

src/odr/internal/zip/zip_util.cpp

@@ -6,6 +6,7 @@
 
 #include <algorithm>
 #include <array>
+#include <cstring>
 
 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<std::uint64_t>(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<std::istream> 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,25 +117,21 @@ 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<char, MZ_ZIP_MAX_ARCHIVE_FILENAME_SIZE> filename{};
   mz_zip_reader_get_filename(m_archive->zip(), m_index, filename.data(),
                              static_cast<mz_uint>(filename.size()));
   return RelPath(filename.data());
 }
 
 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<abstract::File> Archive::Entry::file() const {
   return std::make_shared<FileInZip>(m_archive->shared_from_this(), m_index);
 }
 
-Archive::Archive(std::shared_ptr<abstract::File> file)
+ReadSource::ReadSource(std::shared_ptr<abstract::File> 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<std::size_t>(size, m_memory->size() - offset);
+    std::memcpy(buffer, m_memory->data() + offset, amount);
+    return amount;
+  }
+
+  std::unique_ptr<std::istream> 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<std::streamoff>(offset));
+  stream->read(static_cast<char *>(buffer), static_cast<std::streamsize>(size));
+  const auto result = static_cast<std::size_t>(stream->gcount());
+
+  {
+    std::lock_guard lock(m_mutex);
+    m_streams.push_back(std::move(stream));
+  }
+
+  return result;
+}
+
+Archive::Archive(std::shared_ptr<abstract::File> file)
+    : m_file{std::move(file)} {
+  if (m_file == nullptr) {
+    throw NullPointerError("Archive: file is nullptr");
+  }
+  m_source = std::make_unique<ReadSource>(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<abstract::File> 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<std::istream *>(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<std::streamsize>(offset));
-    in->read(static_cast<char *>(buffer), static_cast<std::streamsize>(size));
-    return static_cast<std::size_t>(in->gcount());
+    return static_cast<const ReadSource *>(opaque)->read(offset, buffer, size);
   };
   const bool state = mz_zip_reader_init(
       &archive, file.size(), MZ_ZIP_FLAG_DO_NOT_SORT_CENTRAL_DIRECTORY);

src/odr/internal/zip/zip_util.hpp

@@ -6,7 +6,10 @@
 #include <istream>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <string>
+#include <string_view>
+#include <vector>
 
 #include <miniz/miniz.h>
 #include <miniz/miniz_zip.h>
@@ -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<abstract::File> file);
+
+  [[nodiscard]] std::size_t read(std::uint64_t offset, void *buffer,
+                                 std::size_t size) const;
+
+private:
+  std::shared_ptr<abstract::File> m_file;
+  std::optional<std::string_view> m_memory;
+
+  mutable std::mutex m_mutex;
+  mutable std::vector<std::unique_ptr<std::istream>> 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<Archive> {
 public:
   explicit Archive(std::shared_ptr<abstract::File> file);
   ~Archive();
 
-  [[nodiscard]] std::mutex &mutex() const;
   [[nodiscard]] mz_zip_archive *zip() const;
 
   [[nodiscard]] std::shared_ptr<abstract::File> file() const noexcept;
@@ -101,14 +122,13 @@ class Archive final : public std::enable_shared_from_this<Archive> {
 
 private:
   std::shared_ptr<abstract::File> m_file;
-  std::unique_ptr<std::istream> m_stream;
+  std::unique_ptr<ReadSource> 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.

test/src/internal/zip/zip_archive_test.cpp

@@ -14,6 +14,7 @@
 #include <fstream>
 #include <memory>
 #include <string>
+#include <thread>
 #include <vector>
 
 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<std::string> 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<char>('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<MemoryFile>(expected[i]));
+    }
+    std::ofstream out(path, std::ios::binary);
+    zip.save(out);
+  }
+
+  const std::vector<std::shared_ptr<abstract::File>> sources{
+      std::make_shared<DiskFile>(path),
+      std::make_shared<MemoryFile>(DiskFile(path))};
+
+  for (const auto &source : sources) {
+    const auto zip = std::make_shared<util::Archive>(source);
+
+    std::vector<std::string> actual(entry_count);
+    std::vector<std::thread> 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<char>(*stream),
+                                std::istreambuf_iterator<char>());
+      });
+    }
+    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,