diff --git a/src/borg/archiver/help_cmd.py b/src/borg/archiver/help_cmd.py index ce34b6c06a..28980f7db6 100644 --- a/src/borg/archiver/help_cmd.py +++ b/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``, diff --git a/src/borg/archiver/tar_cmds.py b/src/borg/archiver/tar_cmds.py index f3fd0a8dd5..f4a49ed627 100644 --- a/src/borg/archiver/tar_cmds.py +++ b/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} diff --git a/src/borg/compress.pyi b/src/borg/compress.pyi index d11b7bff0f..e3575840ea 100644 --- a/src/borg/compress.pyi +++ b/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: ... diff --git a/src/borg/compress.pyx b/src/borg/compress.pyx index c31014dcb4..766b7ab756 100644 --- a/src/borg/compress.pyx +++ b/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 diff --git a/src/borg/testsuite/compress_test.py b/src/borg/testsuite/compress_test.py index 77981d94f3..73318fe0a9 100644 --- a/src/borg/testsuite/compress_test.py +++ b/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)