fix(dwarfsextract): progress didn't work with patterns (fixes gh #316)

Author: mhx

include/dwarfs/utility/filesystem_extractor.h

@@ -29,8 +29,8 @@
 #pragma once
 
 #include <filesystem>
-#include <functional>
 #include <memory>
+#include <optional>
 #include <ostream>
 #include <string>
 #include <string_view>
@@ -57,11 +57,16 @@ struct filesystem_extractor_archive_format;
 struct filesystem_extractor_options {
   size_t max_queued_bytes{static_cast<size_t>(512) << 20};
   bool continue_on_error{false};
-  std::function<void(std::string_view, uint64_t, uint64_t)> progress;
+  bool enable_progress{true};
 };
 
 class filesystem_extractor {
  public:
+  struct progress_info {
+    uint64_t extracted_bytes{0};
+    std::optional<uint64_t> total_bytes{};
+  };
+
   filesystem_extractor(logger& lgr, os_access const& os,
                        std::shared_ptr<file_access const> fa = nullptr);
 
@@ -96,6 +101,8 @@ class filesystem_extractor {
     return impl_->extract(fs, matcher, opts);
   }
 
+  progress_info get_progress() const { return impl_->get_progress(); }
+
   class impl {
    public:
     virtual ~impl() = default;
@@ -111,6 +118,7 @@ class filesystem_extractor {
     virtual bool
     extract(reader::filesystem_v2_lite const& fs, glob_matcher const* matcher,
             filesystem_extractor_options const& opts) = 0;
+    virtual progress_info get_progress() const = 0;
   };
 
  private:

src/utility/filesystem_extractor.cpp

@@ -254,6 +254,21 @@ class filesystem_extractor_ final : public filesystem_extractor::impl {
   extract(reader::filesystem_v2_lite const& fs, glob_matcher const* matcher,
           filesystem_extractor_options const& opts) override;
 
+  filesystem_extractor::progress_info get_progress() const override {
+    filesystem_extractor::progress_info pi;
+
+    {
+      std::lock_guard lock(bytes_total_mx_);
+      if (bytes_total_) {
+        pi.total_bytes = *bytes_total_;
+      }
+    }
+
+    pi.extracted_bytes = bytes_written_.load();
+
+    return pi;
+  }
+
  private:
   static la_ssize_t on_stream_write(struct archive* /*a*/, void* client_data,
                                     void const* buffer, size_t length) {
@@ -372,6 +387,9 @@ class filesystem_extractor_ final : public filesystem_extractor::impl {
   std::array<int, 2> pipefd_{-1, -1};
   std::unique_ptr<std::thread> iot_;
   sparse_file_mode sparse_mode_{sparse_file_mode::auto_detect};
+  std::atomic<uint64_t> bytes_written_{0};
+  std::mutex mutable bytes_total_mx_;
+  std::optional<uint64_t> bytes_total_{};
 };
 
 template <typename LoggerPolicy>
@@ -420,18 +438,12 @@ bool filesystem_extractor_<LoggerPolicy>::extract(
         });
   }
 
-  vfs_stat vfs;
-  fs.statvfs(&vfs);
-
   std::atomic<size_t> hard_error{0};
   std::atomic<size_t> soft_error{0};
-  std::atomic<uint64_t> bytes_written{0};
-  uint64_t const bytes_total{vfs.blocks};
 
   auto do_archive = [&](worker_ptr aptr, std::shared_ptr<::archive_entry> ae,
                         reader::inode_view const& entry) {
-    bool added{false};
-
+    // hard links will have size 0
     if (auto const size = ::archive_entry_size(ae.get());
         entry.is_regular_file() && size > 0) {
       reader::detail::file_reader fr(fs, entry);
@@ -449,8 +461,7 @@ bool filesystem_extractor_<LoggerPolicy>::extract(
       aptr->add_job<struct archive*>(
           [this, &hard_error, &soft_error, &opts, extents = std::move(extents),
            ranges = fr.read_sequential(data_ranges, sem, opts.max_queued_bytes),
-           ae = std::move(ae), size, &sparse_mode, &bytes_written,
-           bytes_total](struct archive* a) mutable {
+           ae = std::move(ae), size, &sparse_mode](struct archive* a) mutable {
             try {
               assert(ae);
 
@@ -521,9 +532,8 @@ bool filesystem_extractor_<LoggerPolicy>::extract(
                   extents.erase(extents.begin());
                 }
 
-                if (opts.progress) {
-                  bytes_written += r.size();
-                  opts.progress(path, bytes_written, bytes_total);
+                if (opts.enable_progress) {
+                  bytes_written_ += r.size();
                 }
               }
 
@@ -541,11 +551,7 @@ bool filesystem_extractor_<LoggerPolicy>::extract(
               }
             }
           });
-
-      added = true;
-    }
-
-    if (!added) {
+    } else {
       aptr->add_job<struct archive*>(
           [this, ae = std::move(ae), &hard_error](struct archive* a) {
             try {
@@ -567,11 +573,22 @@ bool filesystem_extractor_<LoggerPolicy>::extract(
   std::unordered_set<std::string> matched_dirs;
 
   if (matcher) {
+    std::unordered_set<uint32_t> seen_hardlinks;
+    uint64_t data_size{0};
+
     // Collect all directories that contain matching files to make sure
     // we descend into them during the extraction walk below.
     fs.walk([&](auto entry) {
-      if (!entry.inode().is_directory()) {
+      auto inode = entry.inode();
+      if (!inode.is_directory()) {
         if (matcher->match(entry.unix_path())) {
+          if (opts.enable_progress) {
+            auto stat = fs.getattr(inode);
+            if (stat.nlink() == 1 ||
+                seen_hardlinks.insert(inode.inode_num()).second) {
+              data_size += stat.allocated_size();
+            }
+          }
           while (auto parent = entry.parent()) {
             if (!matched_dirs.insert(parent->unix_path()).second) {
               break;
@@ -581,6 +598,16 @@ bool filesystem_extractor_<LoggerPolicy>::extract(
         }
       }
     });
+
+    if (opts.enable_progress) {
+      std::lock_guard lock(bytes_total_mx_);
+      bytes_total_.emplace(data_size);
+    }
+  } else if (opts.enable_progress) {
+    vfs_stat vfs;
+    fs.statvfs(&vfs);
+    std::lock_guard lock(bytes_total_mx_);
+    bytes_total_.emplace(vfs.blocks);
   }
 
   auto do_archive_entry = [&](auto const& entry) {
@@ -642,8 +669,8 @@ bool filesystem_extractor_<LoggerPolicy>::extract(
       if (ec) {
         LOG_ERROR << "readlink() failed: " << ec.message();
       }
-      if (opts.progress) {
-        bytes_written += link.size();
+      if (opts.enable_progress) {
+        bytes_written_ += link.size();
       }
 #ifdef _WIN32
       std::filesystem::path linkpath(string_to_u8string(link));
@@ -720,7 +747,7 @@ bool filesystem_extractor_<LoggerPolicy>::extract(
     return false;
   }
 
-  LOG_INFO << "extraction finished without errors";
+  LOG_VERBOSE << "extraction finished without errors";
 
   return true;
 }

test/tool_main_perfmon_test.cpp

@@ -43,7 +43,6 @@ TEST(dwarfsextract_test, perfmon) {
   EXPECT_THAT(errs, ::testing::HasSubstr("[filesystem_v2.readv_future]"));
   EXPECT_THAT(errs, ::testing::HasSubstr("[filesystem_v2.getattr]"));
   EXPECT_THAT(errs, ::testing::HasSubstr("[filesystem_v2.readlink_ec]"));
-  EXPECT_THAT(errs, ::testing::HasSubstr("[filesystem_v2.statvfs]"));
   EXPECT_THAT(errs, ::testing::HasSubstr("[inode_reader_v2.readv_future]"));
   static std::regex const perfmon_re{R"(\[filesystem_v2\.getattr\])"
                                      R"(\s+samples:\s+\d+)"

test/tools_test.cpp

@@ -1952,7 +1952,7 @@ TEST(tools_test, dwarfsextract_progress) {
 #endif
 
   ASSERT_TRUE(out);
-  EXPECT_GT(out->size(), 100) << *out;
+  EXPECT_GT(out->size(), 1) << *out;
 #ifdef _WIN32
   EXPECT_THAT(*out, ::testing::EndsWith("100%\r\n"));
 #else

tools/src/dwarfsextract_main.cpp

@@ -26,6 +26,7 @@
  * SPDX-License-Identifier: MIT
  */
 
+#include <chrono>
 #include <exception>
 #include <iostream>
 #include <string>
@@ -52,6 +53,7 @@
 #include <dwarfs_tool_manpage.h>
 
 namespace po = boost::program_options;
+using namespace std::chrono_literals;
 
 namespace dwarfs::tool {
 
@@ -65,6 +67,49 @@ constexpr std::string_view kDash{"-"};
 #endif
 #endif
 
+class progress_thread {
+ public:
+  using fn_type = std::function<void(bool last)>;
+
+  progress_thread(std::chrono::nanoseconds interval, fn_type fn)
+      : thread_([this, interval, fn] { this->run(interval, fn); }) {}
+
+  ~progress_thread() { stop(); }
+
+  void stop() {
+    try {
+      {
+        std::lock_guard lock(running_mx_);
+        if (!running_) {
+          return;
+        }
+        running_ = false;
+      }
+      cond_.notify_all();
+      thread_.join();
+    } catch (...) {
+      DWARFS_PANIC(
+          fmt::format("exception thrown in writer_progress destructor: {}",
+                      exception_str(std::current_exception())));
+    }
+  }
+
+ private:
+  void run(std::chrono::nanoseconds interval, fn_type fn) {
+    std::unique_lock lock(running_mx_);
+    do {
+      fn(false);
+      cond_.wait_for(lock, interval);
+    } while (running_);
+    fn(true);
+  }
+
+  mutable std::mutex running_mx_;
+  bool running_{true};
+  std::condition_variable cond_;
+  std::thread thread_;
+};
+
 } // namespace
 
 int dwarfsextract_main(int argc, sys_char** argv, iolayer const& iol) {
@@ -247,26 +292,33 @@ int dwarfsextract_main(int argc, sys_char** argv, iolayer const& iol) {
 
     fsx_opts.max_queued_bytes = fsopts.block_cache.max_bytes;
     fsx_opts.continue_on_error = continue_on_error;
-    int prog{-1};
+    fsx_opts.enable_progress = stdout_progress;
+
+    std::optional<progress_thread> prog;
+
     if (stdout_progress) {
-      fsx_opts.progress = [&prog, &iol](std::string_view, uint64_t extracted,
-                                        uint64_t total) {
-        int p = 100 * extracted / total;
-        if (p > prog) {
-          prog = p;
-          iol.out << "\r" << prog << "%";
-          iol.out.flush();
-        }
-        if (extracted == total) {
-          iol.out << "\n";
+      prog.emplace(40ms, [&fsx, &iol](bool last) {
+        if (!last) {
+          auto const info = fsx.get_progress();
+          int pct = 0;
+          if (info.total_bytes) {
+            pct = static_cast<int>(100 * info.extracted_bytes /
+                                   info.total_bytes.value());
+          }
+          iol.out << "\r" << pct << "%";
+        } else {
+          iol.out << "\r100%\n";
         }
-      };
+        iol.out.flush();
+      });
     }
 
     rv = fsx.extract(fs, matcher.get(), fsx_opts) ? 0 : 2;
 
     fsx.close();
 
+    prog.reset();
+
     if (perfmon) {
       perfmon->summarize(iol.err);
     }