Calculate linter.config.jobs in cgroupsv2 environments

pylint/lint/run.py

@@ -65,6 +65,18 @@ def _query_cpu() -> int | None:
             cpu_shares = int(file.read().rstrip())
         # For AWS, gives correct value * 1024.
         avail_cpu = int(cpu_shares / 1024)
+    elif Path("/sys/fs/cgroup/cpu.max").is_file():
+        # Cgroupv2 systems
+        with open("/sys/fs/cgroup/cpu.max", encoding="utf-8") as file:
+            line = file.read().rstrip()
+            fields = line.split()
+            if len(fields) == 2:
+                cpu_quota = fields[0]
+                cpu_period = int(fields[1])
+                # Make sure this is not in an unconstrained cgroup
+                if cpu_quota != "max":
+                    cpu_quota = int(cpu_quota)
+                    avail_cpu = int(cpu_quota / cpu_period)
 
     # In K8s Pods also a fraction of a single core could be available
     # As multiprocessing is not able to run only a "fraction" of process

tests/lint/test_run.py

@@ -0,0 +1,58 @@
+# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
+# For details: https://github.com/pylint-dev/pylint/blob/main/LICENSE
+# Copyright (c) https://github.com/pylint-dev/pylint/blob/main/CONTRIBUTORS.txt
+
+from pathlib import Path
+
+import pytest
+from unittest.mock import MagicMock, mock_open, patch
+
+from typing import Any, Protocol
+from io import BufferedReader
+
+from pylint import lint
+from pylint.testutils.utils import _test_cwd
+
+
+@pytest.mark.parametrize(
+    "contents,expected",
+    [
+        ("50000 100000", 1),
+        ("100000 100000", 1),
+        ("200000 100000", 2),
+        ("299999 100000", 2),
+        ("300000 100000", 3),
+        # Unconstrained cgroup
+        ("max 100000", None),
+    ],
+)
+def test_query_cpu_cgroupv2(
+        tmp_path: Path,
+        contents: str,
+        expected: int,
+) -> None:
+    """Check that `pylint.lint.run._query_cpu` generates realistic values in cgroupsv2 systems.
+    """
+    builtin_open = open
+
+    def _mock_open(*args: Any, **kwargs: Any) -> BufferedReader:
+        if args[0] == "/sys/fs/cgroup/cpu.max":
+            return mock_open(read_data=contents)(*args, **kwargs)  # type: ignore[no-any-return]
+        return builtin_open(*args, **kwargs)  # type: ignore[no-any-return]
+
+    pathlib_path = Path
+
+    def _mock_path(*args: str, **kwargs: Any) -> Path:
+        if args[0] == "/sys/fs/cgroup/cpu/cpu.shares":
+            return MagicMock(is_file=lambda: False)
+        if args[0] == "/sys/fs/cgroup/cpu/cfs_quota_us":
+            return MagicMock(is_file=lambda: False)
+        if args[0] == "/sys/fs/cgroup/cpu.max":
+            return MagicMock(is_file=lambda: True)
+        return pathlib_path(*args, **kwargs)
+
+    with _test_cwd(tmp_path):
+        with patch("builtins.open", _mock_open):
+            with patch("pylint.lint.run.Path", _mock_path):
+                cpus = lint.run._query_cpu()
+                assert cpus == expected

tests/test_pylint_runners.py

@@ -100,3 +100,45 @@ def _mock_path(*args: str, **kwargs: Any) -> pathlib.Path:
                 with patch("pylint.lint.run.Path", _mock_path):
                     Run(testargs, reporter=Reporter())
         assert err.value.code == 0
+
+
+@pytest.mark.parametrize(
+    "contents",
+    [
+        "1 2",
+        "max 100000",
+    ],
+)
+def test_pylint_run_jobs_equal_zero_dont_crash_with_cgroupv2(
+        tmp_path: pathlib.Path,
+        contents: str,
+) -> None:
+    """Check that the pylint runner does not crash if `pylint.lint.run._query_cpu`
+    determines only a fraction of a CPU core to be available.
+    """
+    builtin_open = open
+
+    def _mock_open(*args: Any, **kwargs: Any) -> BufferedReader:
+        if args[0] == "/sys/fs/cgroup/cpu.max":
+            return mock_open(read_data=contents)(*args, **kwargs)  # type: ignore[no-any-return]
+        return builtin_open(*args, **kwargs)  # type: ignore[no-any-return]
+
+    pathlib_path = pathlib.Path
+
+    def _mock_path(*args: str, **kwargs: Any) -> pathlib.Path:
+        if args[0] == "/sys/fs/cgroup/cpu/cpu.shares":
+            return MagicMock(is_file=lambda: False)
+        if args[0] == "/sys/fs/cgroup/cpu/cfs_quota_us":
+            return MagicMock(is_file=lambda: False)
+        if args[0] == "/sys/fs/cgroup/cpu.max":
+            return MagicMock(is_file=lambda: True)
+        return pathlib_path(*args, **kwargs)
+
+    filepath = os.path.abspath(__file__)
+    testargs = [filepath, "--jobs=0"]
+    with _test_cwd(tmp_path):
+        with pytest.raises(SystemExit) as err:
+            with patch("builtins.open", _mock_open):
+                with patch("pylint.lint.run.Path", _mock_path):
+                    Run(testargs, reporter=Reporter())
+        assert err.value.code == 0