implement archive chunking

Teach pkgfile how to break a single tar file into multiple cpio files.
Doing this puts an upper bound on how much work a single thread has to
do and makes queries on huge repos like [extra] signficantly faster.
On my machine, --search operations see about a 75% reduction in latency,
whereas --list operations see closer to a 50% reduction in latency.

Implements a suggestion from #71.

Author: falconindy

build-aux/ci-test

@@ -21,15 +21,15 @@ endgroup() {
   echo "::endgroup::"
 }
 
-debug "Running with buildmode=$buildmode"
-
 group 'setup'
 
 buildmode=$1
 setup_flags=()
 builddir=build-$buildmode
 test_install=0
 
+debug "Running with buildmode=$buildmode"
+
 case $buildmode in
   plain)
     test_install=1
@@ -60,7 +60,7 @@ endgroup
 
 # test
 group 'test'
-meson test -C "$builddir"
+meson test -C "$builddir" --print-errorlogs -v
 endgroup
 
 # install

build-aux/test-san

@@ -6,4 +6,4 @@ sanitizer=${1:-address}
 dir=sanbuild$$
 
 meson setup "$dir" -Db_sanitize="$sanitizer" --buildtype=debugoptimized
-meson test -C "$dir"
+ninja -C "$dir" test

meson.build

@@ -53,6 +53,7 @@ libcommon = static_library(
     src/repo.cc src/repo.hh
     src/result.cc src/result.hh
     src/update.cc src/update.hh
+    src/queue.hh
   '''.split(),
     ),
     dependencies: [libpcre, libarchive, libcurl, pthreads, stdcppfs],

src/archive_converter.cc

@@ -4,6 +4,8 @@
 #include <sys/time.h>
 
 #include <filesystem>
+#include <format>
+#include <iostream>
 #include <sstream>
 
 namespace fs = std::filesystem;
@@ -12,10 +14,8 @@ namespace pkgfile {
 
 // static
 std::unique_ptr<ArchiveConverter> ArchiveConverter::New(
-    const std::string& reponame, int fd_in, const std::string& filename_out,
-    int compress) {
-  std::string tmpfile_path = filename_out;
-  tmpfile_path.append("~");
+    const std::string& reponame, int fd_in, std::string base_filename_out,
+    int compress, int repo_chunk_bytes) {
   const char* error;
 
   auto reader = ReadArchive::New(fd_in, &error);
@@ -25,15 +25,44 @@ std::unique_ptr<ArchiveConverter> ArchiveConverter::New(
     return nullptr;
   }
 
-  auto writer = WriteArchive::New(tmpfile_path, compress, &error);
+  auto writer = WriteArchive::New(
+      MakeArchiveChunkFilename(base_filename_out, 0, true), compress, &error);
   if (writer == nullptr) {
     fprintf(stderr, "error: failed to open file for writing: %s: %s\n",
-            tmpfile_path.c_str(), error);
+            base_filename_out.c_str(), error);
     return nullptr;
   }
 
-  return std::make_unique<ArchiveConverter>(reponame, std::move(reader),
-                                            std::move(writer));
+  return std::make_unique<ArchiveConverter>(
+      reponame, std::move(base_filename_out), compress, repo_chunk_bytes,
+      std::move(reader), std::move(writer));
+}
+
+std::string ArchiveConverter::MakeArchiveChunkFilename(
+    const std::string& base_filename, int chunk_number, bool tempfile) {
+  return std::format("{}.{:03d}{}", base_filename, chunk_number,
+                     tempfile ? "~" : "");
+}
+
+bool ArchiveConverter::NextArchiveChunk() {
+  if (!out_->Close()) {
+    return false;
+  }
+
+  const char* error;
+
+  auto writer = WriteArchive::New(
+      MakeArchiveChunkFilename(base_filename_out_, ++chunk_number_, true),
+      compress_, &error);
+  if (writer == nullptr) {
+    fprintf(stderr, "error: failed to open file for writing: %s: %s\n",
+            base_filename_out_.c_str(), error);
+    return false;
+  }
+
+  out_ = std::move(writer);
+
+  return true;
 }
 
 int ArchiveConverter::WriteCpioEntry(archive_entry* ae,
@@ -50,7 +79,7 @@ int ArchiveConverter::WriteCpioEntry(archive_entry* ae,
     entry_data << '/' << line << '\n';
   }
 
-  const auto entry = entry_data.str();
+  const std::string entry = entry_data.str();
 
   // adjust the entry size for removing the first line and adding slashes
   archive_entry_set_size(ae, entry.size());
@@ -74,7 +103,7 @@ int ArchiveConverter::WriteCpioEntry(archive_entry* ae,
     return -errno;
   }
 
-  return 0;
+  return entry.size();
 }
 
 bool ArchiveConverter::Finalize() {
@@ -92,32 +121,57 @@ bool ArchiveConverter::Finalize() {
       {st.st_mtim.tv_sec, 0},
   };
 
-  if (utimes(out_->path().c_str(), times) < 0) {
-    fprintf(stderr, "warning: failed to set filetimes on %s: %s\n",
-            out_->path().c_str(), strerror(errno));
+  for (int i = 0; i <= chunk_number_; ++i) {
+    std::string path = MakeArchiveChunkFilename(base_filename_out_, i, true);
+
+    if (utimes(path.c_str(), times) < 0) {
+      fprintf(stderr, "warning: failed to set filetimes on %s: %s\n",
+              out_->path().c_str(), strerror(errno));
+    }
+
+    const fs::path dest = path.substr(0, path.size() - 1);
+
+    std::error_code ec;
+    if (fs::rename(path, dest, ec); ec.value() != 0) {
+      fprintf(stderr, "error: renaming tmpfile to %s failed: %s\n",
+              dest.c_str(), ec.message().c_str());
+    }
   }
 
-  auto dest = out_->path().substr(0, out_->path().size() - 1);
+  for (int i = chunk_number_ + 1;; ++i) {
+    std::string path = MakeArchiveChunkFilename(base_filename_out_, i, false);
 
-  std::error_code ec;
-  if (fs::rename(out_->path(), dest, ec); ec.value() != 0) {
-    fprintf(stderr, "error: renaming tmpfile to %s failed: %s\n", dest.c_str(),
-            ec.message().c_str());
-    return false;
+    if (unlink(path.c_str()) != 0) {
+      break;
+    }
   }
 
   return true;
 }
 
 bool ArchiveConverter::RewriteArchive() {
   archive_entry* ae;
+  int chunk_size = 0;
+
   while (archive_read_next_header(in_->read_archive(), &ae) == ARCHIVE_OK) {
+    if (chunk_size > repo_chunk_bytes_) {
+      if (!NextArchiveChunk()) {
+        return false;
+      }
+
+      chunk_size = 0;
+    }
+
     fs::path entryname = archive_entry_pathname(ae);
+
     // ignore everything but the /files metadata
     if (entryname.filename() == "files") {
-      if (WriteCpioEntry(ae, entryname) < 0) {
+      const int bytes_written = WriteCpioEntry(ae, entryname);
+      if (bytes_written < 0) {
         return false;
       }
+
+      chunk_size += bytes_written;
     }
   }
 

src/archive_converter.hh

@@ -17,9 +17,15 @@ namespace pkgfile {
 // packed in CPIO).
 class ArchiveConverter {
  public:
-  ArchiveConverter(std::string reponame, std::unique_ptr<ReadArchive> in,
+  ArchiveConverter(std::string reponame, std::string base_filename_out,
+                   int compress, int repo_chunk_bytes,
+                   std::unique_ptr<ReadArchive> in,
                    std::unique_ptr<WriteArchive> out)
       : reponame_(std::move(reponame)),
+        base_filename_out_(std::move(base_filename_out)),
+        compress_(compress),
+        repo_chunk_bytes_(repo_chunk_bytes <= 0 ? kDefaultRepoChunkMax
+                                                : repo_chunk_bytes),
         in_(std::move(in)),
         out_(std::move(out)) {}
 
@@ -31,19 +37,32 @@ class ArchiveConverter {
 
   static std::unique_ptr<ArchiveConverter> New(const std::string& reponame,
                                                int fd_in,
-                                               const std::string& filename_out,
-                                               int compress);
+                                               std::string base_filename_out,
+                                               int compress,
+                                               int repo_chunk_bytes);
 
   bool RewriteArchive();
 
  private:
+  static constexpr int kDefaultRepoChunkMax = 40 * (1 << 20);
+
   int WriteCpioEntry(archive_entry* ae, const std::filesystem::path& entryname);
   bool Finalize();
 
+  static std::string MakeArchiveChunkFilename(const std::string& base_filename,
+                                              int chunk_number, bool tempfile);
+
+  bool NextArchiveChunk();
+
   std::string reponame_;
-  std::string destfile_;
+  std::string base_filename_out_;
+  int compress_;
+  int repo_chunk_bytes_;
+
   std::unique_ptr<ReadArchive> in_;
   std::unique_ptr<WriteArchive> out_;
+
+  int chunk_number_ = 0;
 };
 
 }  // namespace pkgfile