make multi threaded compression configurable and use single threaded

by default.

4 files changed, 31 insertions, 20 deletions

diff --git a/src/libstore/binary-cache-store.cc b/src/libstore/binary-cache-store.cc
index ab971dd8b6d9..d34adbd60c71 100644
--- a/src/libstore/binary-cache-store.cc
+++ b/src/libstore/binary-cache-store.cc
@@ -149,7 +149,7 @@ void BinaryCacheStore::addToStore(const ValidPathInfo & info, const ref<std::str
     /* Compress the NAR. */
     narInfo->compression = compression;
     auto now1 = std::chrono::steady_clock::now();
-    auto narCompressed = compress(compression, *nar);
+    auto narCompressed = compress(compression, *nar, settings.parallelCompression);
     auto now2 = std::chrono::steady_clock::now();
     narInfo->fileHash = hashString(htSHA256, *narCompressed);
     narInfo->fileSize = narCompressed->size();
diff --git a/src/libstore/globals.hh b/src/libstore/globals.hh
index 20ac8fe4e9ae..aafec2ea269b 100644
--- a/src/libstore/globals.hh
+++ b/src/libstore/globals.hh
@@ -174,6 +174,9 @@ public:
         "Whether to compress logs.",
         {"build-compress-log"}};
 
+    Setting<bool> parallelCompression{this, false, "parallel-compression",
+        "Whether to enable parallel compression, only supported with xz currently"};
+
     Setting<unsigned long> maxLogSize{this, 0, "max-build-log-size",
         "Maximum number of bytes a builder can write to stdout/stderr "
         "before being killed (0 means no limit).",
diff --git a/src/libutil/compression.cc b/src/libutil/compression.cc
index a36c4405f8bb..ed15761b32a2 100644
--- a/src/libutil/compression.cc
+++ b/src/libutil/compression.cc
@@ -151,10 +151,10 @@ static ref<std::string> decompressBrotli(const std::string & in)
 #endif // HAVE_BROTLI
 }
 
-ref<std::string> compress(const std::string & method, const std::string & in)
+ref<std::string> compress(const std::string & method, const std::string & in, const bool parallel)
 {
     StringSink ssink;
-    auto sink = makeCompressionSink(method, ssink);
+    auto sink = makeCompressionSink(method, ssink, parallel);
     (*sink)(in);
     sink->finish();
     return ssink.s;
@@ -189,20 +189,28 @@ struct XzSink : CompressionSink
     lzma_stream strm = LZMA_STREAM_INIT;
     bool finished = false;
 
-    XzSink(Sink & nextSink) : nextSink(nextSink)
+    XzSink(Sink & nextSink, const bool parallel) : nextSink(nextSink)
     {
-        lzma_mt mt_options = {};
-        mt_options.flags = 0;
-        mt_options.timeout = 300; // Using the same setting as the xz cmd line
-        mt_options.check = LZMA_CHECK_CRC64;
-        mt_options.threads = lzma_cputhreads();
-        mt_options.block_size = 0;
-        if (mt_options.threads == 0)
-            mt_options.threads = 1;
-        // FIXME: maybe use lzma_stream_encoder_mt_memusage() to control the
-        // number of threads.
-        lzma_ret ret = lzma_stream_encoder_mt(
-            &strm, &mt_options);
+        lzma_ret ret;
+        if (parallel) {
+            lzma_mt mt_options = {};
+            mt_options.flags = 0;
+            mt_options.timeout = 300; // Using the same setting as the xz cmd line
+            mt_options.preset = LZMA_PRESET_DEFAULT;
+            mt_options.filters = NULL;
+            mt_options.check = LZMA_CHECK_CRC64;
+            mt_options.threads = lzma_cputhreads();
+            mt_options.block_size = 0;
+            if (mt_options.threads == 0)
+                mt_options.threads = 1;
+            // FIXME: maybe use lzma_stream_encoder_mt_memusage() to control the
+            // number of threads.
+            ret = lzma_stream_encoder_mt(
+                &strm, &mt_options);
+        } else
+            ret = lzma_easy_encoder(
+                &strm, 6, LZMA_CHECK_CRC64);
+
         if (ret != LZMA_OK)
             throw CompressionError("unable to initialise lzma encoder");
         // FIXME: apply the x86 BCJ filter?
@@ -459,12 +467,12 @@ struct BrotliSink : CompressionSink
 };
 #endif // HAVE_BROTLI
 
-ref<CompressionSink> makeCompressionSink(const std::string & method, Sink & nextSink)
+ref<CompressionSink> makeCompressionSink(const std::string & method, Sink & nextSink, const bool parallel)
 {
     if (method == "none")
         return make_ref<NoneSink>(nextSink);
     else if (method == "xz")
-        return make_ref<XzSink>(nextSink);
+        return make_ref<XzSink>(nextSink, parallel);
     else if (method == "bzip2")
         return make_ref<BzipSink>(nextSink);
     else if (method == "br")
diff --git a/src/libutil/compression.hh b/src/libutil/compression.hh
index e3e6f5a99303..a0d7530d74fc 100644
--- a/src/libutil/compression.hh
+++ b/src/libutil/compression.hh
@@ -8,7 +8,7 @@
 
 namespace nix {
 
-ref<std::string> compress(const std::string & method, const std::string & in);
+ref<std::string> compress(const std::string & method, const std::string & in, const bool parallel = false);
 
 ref<std::string> decompress(const std::string & method, const std::string & in);
 
@@ -17,7 +17,7 @@ struct CompressionSink : BufferedSink
     virtual void finish() = 0;
 };
 
-ref<CompressionSink> makeCompressionSink(const std::string & method, Sink & nextSink);
+ref<CompressionSink> makeCompressionSink(const std::string & method, Sink & nextSink, const bool parallel = false);
 
 MakeError(UnknownCompressionMethod, Error);