compression: cap the default zstd MT workers at 4 for chunks

src/borg/archiver/help_cmd.py

@@ -764,15 +764,24 @@ class HelpMixIn:
                 0 means "always multi-threaded", a very large value effectively disables multi-threading.
             BORG_ZSTD_MT_WORKERS
                 When set to a numeric value, use that many threads to zstd-compress a single chunk
-                (default: the cpu count). 0 or 1 means single-threaded compression.
+                (default: the cpu count, but at most 4). 0 or 1 means single-threaded compression.
                 Only relevant when compressing with ``zstd``.
                 Chunks below 768KiB are always compressed single-threaded: libzstd will not use a
                 compression job smaller than 512KiB, so a small chunk gets split very unevenly and
                 multi-threading it would be slower than not doing it at all.
+                The default is capped at 4 because a chunk of the size the default chunker aims at
+                (2MiB) splits into just 4 such jobs: threads beyond that get (nearly) no work, but
+                the whole thread pool is created again for every chunk. Measured on a 12-core
+                machine, 4 threads beat 12 on every test corpus at the default ``zstd,-4``
+                (+13% .. +37%). Raising the value only pays off if you configured the chunker
+                for much bigger chunks. ``borg export-tar`` compresses one long stream instead of
+                separate chunks and always defaults to the cpu count.
                 Multi-threading trades a little compression ratio for speed (measured at ``zstd,3``:
                 +0.05% archive size for 1MiB chunks, +0.64% for 8MiB ones, more at higher levels), and
                 it uses more cpu time in total to reduce the wallclock time. Set it to 1 if you would
                 rather have the smaller archive, or if borg has to share the cpu with other work.
+                Single-threaded can even be faster on data zstd races through anyway, e.g.
+                already-compressed/incompressible data or long-repeat data like VM images.
             BORG_FASTCDC_KERNEL / BORG_BUZHASH64_KERNEL
                 Select the scan kernel the ``fastcdc`` / ``buzhash64`` chunker uses (default:
                 ``scalar``, the plain sequential loop). Accepted values are ``avx512``, ``avx2``,

src/borg/archiver/tar_cmds.py

@@ -188,7 +188,7 @@ def create_zstd_filter(stream, stream_close, decompress):
     if decompress:
         zstream = zstd.ZstdFile(stream, "rb")
     else:
-        workers = get_zstd_mt_workers()
+        workers = get_zstd_mt_workers(stream=True)
         if workers > 1:
             params = zstd.CompressionParameter
             options = {params.compression_level: ZSTD_TAR_LEVEL, params.nb_workers: workers}

src/borg/compress.pyi

@@ -4,7 +4,7 @@ ZSTD_JOB_SIZE_MIN: int
 ZSTD_MT_MIN_SIZE: int
 
 def get_compressor(name: str, **kwargs) -> Any: ...
-def get_zstd_mt_workers() -> int: ...
+def get_zstd_mt_workers(stream: bool = ...) -> int: ...
 
 class Compressor:
     def __init__(self, name: Any = ..., **kwargs) -> None: ...

src/borg/compress.pyx

@@ -53,17 +53,30 @@ ZSTD_JOB_SIZE_MIN = 512 * 1024
 # 768KiB (= 1.5 jobs) keeps some margin over break-even, as the overhead is machine dependent.
 ZSTD_MT_MIN_SIZE = 768 * 1024
 
-_zstd_mt_workers = None
+_zstd_mt_workers = None  # cached (chunk workers, stream workers)
 
 
-def get_zstd_mt_workers():
-    """How many threads libzstd may use to compress a single chunk.
+def get_zstd_mt_workers(stream=False):
+    """How many threads libzstd may use to compress a single chunk (or stream).
 
-    Defaults to the cpu count, BORG_ZSTD_MT_WORKERS overrides it. 0 or 1 means
-    single-threaded compression, which also avoids the small loss of compression ratio
-    that splitting a chunk into jobs causes (measured at zstd,3: +0.05% for a 1MiB chunk,
-    +0.64% for an 8MiB one; higher levels lose a bit more as they rely on longer match
-    history).
+    BORG_ZSTD_MT_WORKERS overrides the defaults below (for chunks and streams alike).
+    0 or 1 means single-threaded compression, which also avoids the small loss of
+    compression ratio that splitting a chunk into jobs causes (measured at zstd,3:
+    +0.05% for a 1MiB chunk, +0.64% for an 8MiB one; higher levels lose a bit more as
+    they rely on longer match history).
+
+    For chunks (stream=False), the default is the cpu count, but at most 4: a chunk
+    only yields ceil(size / ZSTD_JOB_SIZE_MIN) jobs - 4 for the 2 MiB chunks the
+    default chunker aims at - so threads beyond that get (nearly) no work, while the
+    whole thread pool is still created and torn down again for every single chunk.
+    Measured on a 12-core machine, 4 workers beat 12 on every corpus tested at the
+    default zstd,-4, by +13% (source code) .. +37% (VM image) big-chunk throughput;
+    higher levels showed smaller differences, but no clear win for 12 anywhere.
+    Raise the value if you configured the chunker for much bigger chunks.
+
+    For long streams (stream=True, used by export-tar), the default is the cpu count:
+    one pool with libzstd's default (large) job size compresses the whole stream, so
+    there are enough jobs and the pool overhead is paid only once.
 
     This is called for every chunk, so the result is cached. The env var is evaluated on
     first use rather than at import time, so that an invalid value is reported via borg's
@@ -72,17 +85,19 @@ def get_zstd_mt_workers():
     global _zstd_mt_workers
     if _zstd_mt_workers is None:
         value = os.environ.get("BORG_ZSTD_MT_WORKERS")
+        cpus = os.cpu_count() or 1
         if value is None:
-            workers = os.cpu_count() or 1
+            workers = (min(cpus, 4), cpus)
         else:
             try:
-                workers = int(value)
+                configured = int(value)
             except ValueError:
                 raise Error(f"BORG_ZSTD_MT_WORKERS must be an integer, but is: {value!r}") from None
-            if workers < 0:
-                raise Error(f"BORG_ZSTD_MT_WORKERS must not be negative, but is: {workers}")
+            if configured < 0:
+                raise Error(f"BORG_ZSTD_MT_WORKERS must not be negative, but is: {configured}")
+            workers = (configured, configured)
         _zstd_mt_workers = workers
-    return _zstd_mt_workers
+    return _zstd_mt_workers[1 if stream else 0]
 
 cdef extern from "lz4.h":
     int LZ4_compress_default(const char* source, char* dest, int inputSize, int maxOutputSize) nogil

src/borg/testsuite/compress_test.py

@@ -339,17 +339,20 @@ def test_robj_specific_obfuscation(data_length, expected_padding, robj_type):
 
 
 def test_zstd_mt_workers_from_env(monkeypatch):
-    def workers_for(env_value):
+    def workers_for(env_value, stream=False):
         monkeypatch.setattr("borg.compress._zstd_mt_workers", None)  # drop the cache
         if env_value is None:
             monkeypatch.delenv("BORG_ZSTD_MT_WORKERS", raising=False)
         else:
             monkeypatch.setenv("BORG_ZSTD_MT_WORKERS", env_value)
-        return get_zstd_mt_workers()
-
-    assert workers_for(None) == (os.cpu_count() or 1)
-    for value, expected in [("0", 0), ("1", 1), ("4", 4)]:
-        assert workers_for(value) == expected
+        return get_zstd_mt_workers(stream=stream)
+
+    cpus = os.cpu_count() or 1
+    assert workers_for(None) == min(cpus, 4)  # per-chunk default is capped
+    assert workers_for(None, stream=True) == cpus  # stream default is not
+    for value, expected in [("0", 0), ("1", 1), ("4", 4), ("12", 12)]:
+        assert workers_for(value) == expected  # the env var is not capped
+        assert workers_for(value, stream=True) == expected
     for invalid in ["", "yes", "4x", "1.5", "-1"]:
         with pytest.raises(Error):
             workers_for(invalid)