binpkg: compress/decompress xz in parallel; compress zstd in parallel

- As before, needs >= xz-5.3.3_alpha for parallel decompression.
- zstd does not support parallel decompression.

Closes: https://github.com/gentoo/portage/pull/918
Signed-off-by: Sam James <sam@gentoo.org>

7 files changed, 60 insertions, 9 deletions

diff --git a/NEWS b/NEWS
index 28e8f97ca..a3591e94d 100644
--- a/NEWS
+++ b/NEWS
@@ -1,3 +1,10 @@
+portage-3.0.39 (UNRELEASED)
+--------------
+
+Features:
+* gpkg: Call xz and zstd with -T N where N is the number of jobs from ${MAKEOPTS}
+  for parallel compression (xz & zstd) and decompression (xz with newer versions).
+
 portage-3.0.38.1 (2022-10-04)
 --------------
 
diff --git a/bin/quickpkg b/bin/quickpkg
index 67eee697a..9ff8d7b5a 100755
--- a/bin/quickpkg
+++ b/bin/quickpkg
@@ -36,6 +36,7 @@ from portage.exception import (
     PermissionDenied,
 )
 from portage.util import ensure_dirs, shlex_split, varexpand, _xattr
+from portage.util.cpuinfo import makeopts_to_job_count
 
 xattr = _xattr.xattr
 from portage._sets import load_default_config, SETPREFIX
@@ -168,7 +169,12 @@ def quickpkg_atom(options, infos, arg, eout):
                         % (binpkg_compression, missing_package)
                     )
                     return 1
-                cmd = shlex_split(varexpand(compression["compress"], mydict=settings))
+
+                cmd = compression["compress"].replace(
+                    "{JOBS}",
+                    str(makeopts_to_job_count(settings.get("MAKEOPTS", "1"))),
+                )
+                cmd = shlex_split(varexpand(cmd, mydict=settings))
                 # Filter empty elements that make Popen fail
                 cmd = [x for x in cmd if x != ""]
                 with open(binpkg_tmpfile, "wb") as fobj:
diff --git a/lib/_emerge/BinpkgExtractorAsync.py b/lib/_emerge/BinpkgExtractorAsync.py
index 919837fc1..65b383998 100644
--- a/lib/_emerge/BinpkgExtractorAsync.py
+++ b/lib/_emerge/BinpkgExtractorAsync.py
@@ -10,6 +10,7 @@ from portage.util.compression_probe import (
     compression_probe,
     _compressors,
 )
+from portage.util.cpuinfo import makeopts_to_job_count
 from portage.process import find_binary
 from portage.util import (
     shlex_split,
@@ -53,6 +54,9 @@ class BinpkgExtractorAsync(SpawnProcess):
         decomp = _compressors.get(compression_probe(self.pkg_path))
         if decomp is not None:
             decomp_cmd = decomp.get("decompress")
+            decomp_cmd = decomp_cmd.replace(
+                "{JOBS}", str(makeopts_to_job_count(self.env.get("MAKEOPTS", "1")))
+            )
         elif tarfile.is_tarfile(
             portage._unicode_encode(
                 self.pkg_path, encoding=portage._encodings["fs"], errors="strict"
diff --git a/lib/portage/gpkg.py b/lib/portage/gpkg.py
index 7e1b98370..1f8fa1fd3 100644
--- a/lib/portage/gpkg.py
+++ b/lib/portage/gpkg.py
@@ -38,6 +38,7 @@ from portage.util._urlopen import urlopen
 from portage.util import writemsg
 from portage.util import shlex_split, varexpand
 from portage.util.compression_probe import _compressors
+from portage.util.cpuinfo import makeopts_to_job_count
 from portage.process import find_binary
 from portage.const import MANIFEST2_HASH_DEFAULTS, HASHING_BLOCKSIZE
 
@@ -1787,7 +1788,12 @@ class gpkg:
         if mode not in compressor:
             raise InvalidCompressionMethod("{}: {}".format(compression, mode))
 
-        cmd = shlex_split(varexpand(compressor[mode], mydict=self.settings))
+        cmd = compressor[mode]
+        cmd = cmd.replace(
+            "{JOBS}", str(makeopts_to_job_count(self.settings.get("MAKEOPTS", "1")))
+        )
+        cmd = shlex_split(varexpand(cmd, mydict=self.settings))
+
         # Filter empty elements that make Popen fail
         cmd = [x for x in cmd if x != ""]
 
diff --git a/lib/portage/package/ebuild/doebuild.py b/lib/portage/package/ebuild/doebuild.py
index d0d134b39..4f7049c21 100644
--- a/lib/portage/package/ebuild/doebuild.py
+++ b/lib/portage/package/ebuild/doebuild.py
@@ -111,7 +111,7 @@ from portage.util import (
     writemsg_stdout,
     write_atomic,
 )
-from portage.util.cpuinfo import get_cpu_count
+from portage.util.cpuinfo import get_cpu_count, makeopts_to_job_count
 from portage.util.lafilefixer import rewrite_lafile
 from portage.util.compression_probe import _compressors
 from portage.util.futures import asyncio
@@ -667,8 +667,12 @@ def doebuild_environment(
                 mysettings["PORTAGE_COMPRESSION_COMMAND"] = "cat"
         else:
             try:
+                compression_binary = compression["compress"].replace(
+                    "{JOBS}",
+                    str(makeopts_to_job_count(mysettings.get("MAKEOPTS", "1"))),
+                )
                 compression_binary = shlex_split(
-                    varexpand(compression["compress"], mydict=settings)
+                    varexpand(compression_binary, mydict=settings)
                 )[0]
             except IndexError as e:
                 writemsg(
@@ -683,9 +687,13 @@ def doebuild_environment(
                         % (binpkg_compression, missing_package)
                     )
                 else:
+                    compression_binary = compression["compress"].replace(
+                        "{JOBS}",
+                        str(makeopts_to_job_count(mysettings.get("MAKEOPTS", "1"))),
+                    )
                     cmd = [
                         varexpand(x, mydict=settings)
-                        for x in shlex_split(compression["compress"])
+                        for x in shlex_split(compression_binary)
                     ]
                     # Filter empty elements
                     cmd = [x for x in cmd if x != ""]
diff --git a/lib/portage/util/compression_probe.py b/lib/portage/util/compression_probe.py
index 423e786bf..bbff5a3c6 100644
--- a/lib/portage/util/compression_probe.py
+++ b/lib/portage/util/compression_probe.py
@@ -37,12 +37,12 @@ _compressors = {
         "package": "app-arch/lzop",
     },
     "xz": {
-        "compress": "xz ${BINPKG_COMPRESS_FLAGS}",
-        "decompress": "xz -d",
+        "compress": "xz -T{JOBS} ${BINPKG_COMPRESS_FLAGS}",
+        "decompress": "xz -T{JOBS} -d",
         "package": "app-arch/xz-utils",
     },
     "zstd": {
-        "compress": "zstd ${BINPKG_COMPRESS_FLAGS}",
+        "compress": "zstd -T{JOBS} ${BINPKG_COMPRESS_FLAGS}",
         # If the compression windowLog was larger than the default of 27,
         # then --long=windowLog needs to be passed to the decompressor.
         # Therefore, pass a larger --long=31 value to the decompressor
diff --git a/lib/portage/util/cpuinfo.py b/lib/portage/util/cpuinfo.py
index 3cbc5b650..77529bcda 100644
--- a/lib/portage/util/cpuinfo.py
+++ b/lib/portage/util/cpuinfo.py
@@ -1,7 +1,9 @@
 # Copyright 2015-2019 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
-__all__ = ["get_cpu_count"]
+import re
+
+__all__ = ["get_cpu_count", "makeopts_to_job_count"]
 
 # Before you set out to change this function, figure out what you're really
 # asking:
@@ -48,3 +50,21 @@ def get_cpu_count():
         return multiprocessing.cpu_count()
     except (ImportError, NotImplementedError):
         return None
+
+
+def makeopts_to_job_count(makeopts):
+    """
+    Parse the job count (-jN) from MAKEOPTS. Python version of
+    bin/isolated-functions.sh's ___makeopts_jobs().
+
+    @return: Number of jobs to run or number of CPUs if none set.
+    """
+    if not makeopts:
+        return get_cpu_count()
+
+    jobs = re.match(r".*(j|--jobs=\s)\s*([0-9]+)", makeopts)
+
+    if not jobs:
+        return get_cpu_count()
+
+    return jobs.groups()[1]