refactor: Extract file recompression code to a class

diff --git a/src/core/CMakeLists.txt b/src/core/CMakeLists.txt
index 0a448b9856b37b14873dbc5bac74925761817218..6eee59c8230b7977a536557f68d07071a470e6b1 100644 (file)
--- a/src/core/CMakeLists.txt
+++ b/src/core/CMakeLists.txt
@@ -1,6 +1,7 @@
 set(
   sources
   CacheEntry.cpp
+  FileRecompressor.cpp
   Manifest.cpp
   MsvcShowIncludesOutput.cpp
   Result.cpp

diff --git a/src/core/FileRecompressor.cpp b/src/core/FileRecompressor.cpp
new file mode 100644 (file)
index 0000000..1327261
--- /dev/null
+++ b/src/core/FileRecompressor.cpp
@@ -0,0 +1,89 @@
+// Copyright (C) 2022 Joel Rosdahl and other contributors
+//
+// See doc/AUTHORS.adoc for a complete list of contributors.
+//
+// This program is free software; you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation; either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program; if not, write to the Free Software Foundation, Inc., 51
+// Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+
+#include "FileRecompressor.hpp"
+
+#include <AtomicFile.hpp>
+#include <Util.hpp>
+#include <core/CacheEntry.hpp>
+#include <core/exceptions.hpp>
+#include <util/expected.hpp>
+#include <util/file.hpp>
+
+namespace core {
+
+int64_t
+FileRecompressor::recompress(const std::string& cache_file,
+                             const std::optional<int8_t> level)
+{
+  core::CacheEntry::Header header(cache_file);
+
+  const int8_t wanted_level =
+    level ? (*level == 0 ? core::CacheEntry::default_compression_level : *level)
+          : 0;
+  const auto old_stat = Stat::lstat(cache_file, Stat::OnError::log);
+  Stat new_stat(old_stat);
+
+  if (header.compression_level != wanted_level) {
+    const auto cache_file_data = util::value_or_throw<core::Error>(
+      util::read_file<util::Bytes>(cache_file),
+      FMT("Failed to read {}: ", cache_file));
+    core::CacheEntry cache_entry(cache_file_data);
+    cache_entry.verify_checksum();
+
+    header.entry_format_version = core::CacheEntry::k_format_version;
+    header.compression_type =
+      level ? core::CompressionType::zstd : core::CompressionType::none;
+    header.compression_level = wanted_level;
+
+    AtomicFile new_cache_file(cache_file, AtomicFile::Mode::binary);
+    new_cache_file.write(
+      core::CacheEntry::serialize(header, cache_entry.payload()));
+    new_cache_file.commit();
+    new_stat = Stat::lstat(cache_file, Stat::OnError::log);
+
+    // Restore mtime/atime to keep cache LRU cleanup working as expected:
+    util::set_timestamps(cache_file, old_stat.mtime(), old_stat.atime());
+  }
+
+  m_content_size += header.entry_size;
+  m_old_size += old_stat.size_on_disk();
+  m_new_size += new_stat.size_on_disk();
+
+  return Util::size_change_kibibyte(old_stat, new_stat);
+}
+
+uint64_t
+FileRecompressor::content_size() const
+{
+  return m_content_size;
+}
+
+uint64_t
+FileRecompressor::old_size() const
+{
+  return m_old_size;
+}
+
+uint64_t
+FileRecompressor::new_size() const
+{
+  return m_new_size;
+}
+
+} // namespace core

diff --git a/src/core/FileRecompressor.hpp b/src/core/FileRecompressor.hpp
new file mode 100644 (file)
index 0000000..289b0d7
--- /dev/null
+++ b/src/core/FileRecompressor.hpp
@@ -0,0 +1,48 @@
+// Copyright (C) 2022 Joel Rosdahl and other contributors
+//
+// See doc/AUTHORS.adoc for a complete list of contributors.
+//
+// This program is free software; you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation; either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program; if not, write to the Free Software Foundation, Inc., 51
+// Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+
+#pragma once
+
+#include <atomic>
+#include <cstdint>
+#include <mutex>
+#include <optional>
+#include <string>
+
+namespace core {
+
+class FileRecompressor
+{
+public:
+  FileRecompressor() = default;
+
+  // Returns on-disk size change in KiB.
+  int64_t recompress(const std::string& cache_file,
+                     const std::optional<int8_t> level);
+
+  uint64_t content_size() const;
+  uint64_t old_size() const;
+  uint64_t new_size() const;
+
+private:
+  std::atomic<uint64_t> m_content_size = 0;
+  std::atomic<uint64_t> m_old_size = 0;
+  std::atomic<uint64_t> m_new_size = 0;
+};
+
+} // namespace core

diff --git a/src/storage/local/LocalStorage_compress.cpp b/src/storage/local/LocalStorage_compress.cpp
index 19c7cd251be17561bdd9cfc147138a37955fa22d..4aa587ad91d9d6d2634c5aa105b559fffc17fd12 100644 (file)
--- a/src/storage/local/LocalStorage_compress.cpp
+++ b/src/storage/local/LocalStorage_compress.cpp
@@ -26,6 +26,7 @@
 #include <ThreadPool.hpp>
 #include <assertions.hpp>
 #include <core/CacheEntry.hpp>
+#include <core/FileRecompressor.hpp>
 #include <core/Manifest.hpp>
 #include <core/Result.hpp>
 #include <core/exceptions.hpp>
@@ -38,125 +39,16 @@
 
 #include <third_party/fmt/core.h>
 
+#include <atomic>
+#include <memory>
+#include <string>
+
 #ifdef HAVE_UNISTD_H
 #  include <unistd.h>
 #endif
 
-#include <memory>
-#include <string>
-
 namespace storage::local {
 
-namespace {
-
-class RecompressionStatistics
-{
-public:
-  void update(uint64_t content_size,
-              uint64_t old_size,
-              uint64_t new_size,
-              uint64_t incompressible_size);
-  uint64_t content_size() const;
-  uint64_t old_size() const;
-  uint64_t new_size() const;
-  uint64_t incompressible_size() const;
-
-private:
-  mutable std::mutex m_mutex;
-  uint64_t m_content_size = 0;
-  uint64_t m_old_size = 0;
-  uint64_t m_new_size = 0;
-  uint64_t m_incompressible_size = 0;
-};
-
-void
-RecompressionStatistics::update(const uint64_t content_size,
-                                const uint64_t old_size,
-                                const uint64_t new_size,
-                                const uint64_t incompressible_size)
-{
-  std::unique_lock<std::mutex> lock(m_mutex);
-  m_incompressible_size += incompressible_size;
-  m_content_size += content_size;
-  m_old_size += old_size;
-  m_new_size += new_size;
-}
-
-uint64_t
-RecompressionStatistics::content_size() const
-{
-  std::unique_lock<std::mutex> lock(m_mutex);
-  return m_content_size;
-}
-
-uint64_t
-RecompressionStatistics::old_size() const
-{
-  std::unique_lock<std::mutex> lock(m_mutex);
-  return m_old_size;
-}
-
-uint64_t
-RecompressionStatistics::new_size() const
-{
-  std::unique_lock<std::mutex> lock(m_mutex);
-  return m_new_size;
-}
-
-uint64_t
-RecompressionStatistics::incompressible_size() const
-{
-  std::unique_lock<std::mutex> lock(m_mutex);
-  return m_incompressible_size;
-}
-
-} // namespace
-
-static void
-recompress_file(RecompressionStatistics& statistics,
-                const std::string& stats_file,
-                const CacheFile& cache_file,
-                const std::optional<int8_t> level)
-{
-  core::CacheEntry::Header header(cache_file.path());
-
-  const int8_t wanted_level =
-    level ? (*level == 0 ? core::CacheEntry::default_compression_level : *level)
-          : 0;
-  const auto old_stat = Stat::stat(cache_file.path(), Stat::OnError::log);
-
-  if (header.compression_level == wanted_level) {
-    statistics.update(header.entry_size, old_stat.size(), old_stat.size(), 0);
-    return;
-  }
-
-  const auto cache_file_data = util::value_or_throw<core::Error>(
-    util::read_file<util::Bytes>(cache_file.path()),
-    FMT("Failed to read {}: ", cache_file.path()));
-  core::CacheEntry cache_entry(cache_file_data);
-  cache_entry.verify_checksum();
-
-  header.entry_format_version = core::CacheEntry::k_format_version;
-  header.compression_type =
-    level ? core::CompressionType::zstd : core::CompressionType::none;
-  header.compression_level = wanted_level;
-
-  AtomicFile new_cache_file(cache_file.path(), AtomicFile::Mode::binary);
-  new_cache_file.write(
-    core::CacheEntry::serialize(header, cache_entry.payload()));
-  new_cache_file.commit();
-
-  // Restore mtime/atime to keep cache LRU cleanup working as expected:
-  util::set_timestamps(cache_file.path(), old_stat.mtime(), old_stat.atime());
-
-  const auto new_stat = Stat::stat(cache_file.path(), Stat::OnError::log);
-  StatsFile(stats_file).update([=](auto& cs) {
-    cs.increment(core::Statistic::cache_size_kibibyte,
-                 Util::size_change_kibibyte(old_stat, new_stat));
-  });
-  statistics.update(header.entry_size, old_stat.size(), new_stat.size(), 0);
-}
-
 CompressionStatistics
 LocalStorage::get_compression_statistics(
   const ProgressReceiver& progress_receiver) const
@@ -197,7 +89,9 @@ LocalStorage::recompress(const std::optional<int8_t> level,
   const size_t read_ahead =
     std::max(static_cast<size_t>(10), 2 * static_cast<size_t>(threads));
   ThreadPool thread_pool(threads, read_ahead);
-  RecompressionStatistics statistics;
+  core::FileRecompressor recompressor;
+
+  std::atomic<uint64_t> incompressible_size = 0;
 
   for_each_level_1_subdir(
     m_config.cache_dir(),
@@ -213,15 +107,22 @@ LocalStorage::recompress(const std::optional<int8_t> level,
         const auto& file = files[i];
 
         if (file.type() != CacheFile::Type::unknown) {
-          thread_pool.enqueue([&statistics, stats_file, file, level] {
-            try {
-              recompress_file(statistics, stats_file, file, level);
-            } catch (core::Error&) {
-              // Ignore for now.
-            }
-          });
+          thread_pool.enqueue(
+            [&recompressor, &incompressible_size, level, stats_file, file] {
+              try {
+                int64_t size_change_kibibyte =
+                  recompressor.recompress(file.path(), level);
+                StatsFile(stats_file).update([=](auto& cs) {
+                  cs.increment(core::Statistic::cache_size_kibibyte,
+                               size_change_kibibyte);
+                });
+              } catch (core::Error&) {
+                // Ignore for now.
+                incompressible_size += file.lstat().size_on_disk();
+              }
+            });
         } else if (!TemporaryFile::is_tmp_file(file.path())) {
-          statistics.update(0, 0, 0, file.lstat().size());
+          incompressible_size += file.lstat().size_on_disk();
         }
 
         sub_progress_receiver(0.1 + 0.9 * i / files.size());
@@ -242,29 +143,30 @@ LocalStorage::recompress(const std::optional<int8_t> level,
     PRINT_RAW(stdout, "\n\n");
   }
 
-  const double old_ratio =
-    statistics.old_size() > 0
-      ? static_cast<double>(statistics.content_size()) / statistics.old_size()
-      : 0.0;
+  const double old_ratio = recompressor.old_size() > 0
+                             ? static_cast<double>(recompressor.content_size())
+                                 / recompressor.old_size()
+                             : 0.0;
   const double old_savings =
     old_ratio > 0.0 ? 100.0 - (100.0 / old_ratio) : 0.0;
-  const double new_ratio =
-    statistics.new_size() > 0
-      ? static_cast<double>(statistics.content_size()) / statistics.new_size()
-      : 0.0;
+  const double new_ratio = recompressor.new_size() > 0
+                             ? static_cast<double>(recompressor.content_size())
+                                 / recompressor.new_size()
+                             : 0.0;
   const double new_savings =
     new_ratio > 0.0 ? 100.0 - (100.0 / new_ratio) : 0.0;
-  const int64_t size_difference = static_cast<int64_t>(statistics.new_size())
-                                  - static_cast<int64_t>(statistics.old_size());
+  const int64_t size_difference =
+    static_cast<int64_t>(recompressor.new_size())
+    - static_cast<int64_t>(recompressor.old_size());
 
   const std::string old_compr_size_str =
-    Util::format_human_readable_size(statistics.old_size());
+    Util::format_human_readable_size(recompressor.old_size());
   const std::string new_compr_size_str =
-    Util::format_human_readable_size(statistics.new_size());
+    Util::format_human_readable_size(recompressor.new_size());
   const std::string content_size_str =
-    Util::format_human_readable_size(statistics.content_size());
+    Util::format_human_readable_size(recompressor.content_size());
   const std::string incompr_size_str =
-    Util::format_human_readable_size(statistics.incompressible_size());
+    Util::format_human_readable_size(incompressible_size);
   const std::string size_difference_str =
     FMT("{}{}",
         size_difference < 0 ? "-" : (size_difference > 0 ? "+" : " "),