diff --git a/docs/usage.md b/docs/usage.md index 61e4de9..e6cb6cb 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -48,7 +48,7 @@ coretrace-python-analyzer [--check | --emit-ir [--ssa]] [options] [path] | `--plugins DIR` | Load the plugins found under `DIR` on top of the bundled ones. Repeatable. | | `--no-bundled-plugins` | Do not load the plugins shipped with the package. | | `--cache DIR` | Keep per-module results under `DIR` and reuse them for unchanged modules. | -| `--jobs N` | Analyse independent modules in `N` processes. | +| `-j N`, `--jobs N` | Analyse independent modules in `N` processes, `0` for one per available core. Default: `1`. | | `--sbom PATH` | Write a CycloneDX bill of materials of the dependencies to `PATH`. | | `--vex PATH` | Write an OpenVEX document to `PATH`: whether each advisory affecting a requirement affects the project. | | `--advisories FILE` | Read a local advisory file in addition to `advisories.json` at the root. Repeatable. | @@ -730,8 +730,15 @@ graph and the modules it imports. On the next run an unchanged module is served from the cache, so editing one file re-analyses that file and its importers only. Entries are plain data; an unreadable entry is recomputed. -`--jobs N` analyses independent modules in `N` processes. Modules are scheduled imports -first, so the result is the same whatever `N`. +`--jobs N` (`-j N`) analyses independent modules in `N` processes; `0` takes one per core +the process may run on, as `runtime.jobs` of `ctrace` does (its CPU affinity, read from +Python 3.13 on, and on Linux before; the cores of the machine otherwise), and the default +is one. On Windows, a pool holds at most 61 processes, the limit of Python's process pool +there, whatever `N`. +Modules are scheduled imports first, so the result is the same whatever `N`: the report, +the order of its findings, `--fail-on`, the cache, the baseline and the SBOM and VEX +documents. The command always waits for the complete report. There is no `--async` +option: `ctrace` deprecated its own for `-j`/`--jobs`, which this follows. ```bash coretrace-python-analyzer --check src/ --cache .coretrace --jobs 4 diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index bb8633b..df39187 100644 --- a/src/coretrace_python/cli.py +++ b/src/coretrace_python/cli.py @@ -109,11 +109,13 @@ def build_parser() -> argparse.ArgumentParser: "them for modules unchanged since the previous run", ) parser.add_argument( + "-j", "--jobs", type=int, default=None, metavar="N", - help="with --check on a directory, analyse independent modules in N processes", + help="with --check on a directory, analyse independent modules in N processes, " + "0 for one per available core; default: 1", ) parser.add_argument( "--sbom", @@ -214,8 +216,8 @@ def main(argv: list[str] | None = None) -> int: if args.jobs is not None and not (args.check and args.path.is_dir()): print("error: --jobs only applies to --check on a directory", file=sys.stderr) return EXIT_ERROR - if args.jobs is not None and args.jobs < 1: - print("error: --jobs must be at least 1", file=sys.stderr) + if args.jobs is not None and args.jobs < 0: + print("error: --jobs must be 0 (one per available core) or more", file=sys.stderr) return EXIT_ERROR if args.sbom is not None and not (args.check and args.path.is_dir()): print("error: --sbom only applies to --check on a directory", file=sys.stderr) @@ -270,7 +272,7 @@ def main(argv: list[str] | None = None) -> int: args.path, plugin_roots, cache=cache, - jobs=args.jobs or 1, + jobs=_processes(args.jobs), advisory_files=args.advisories, policy_file=args.policy, ) @@ -331,5 +333,31 @@ def main(argv: list[str] | None = None) -> int: return EXIT_CLEAN +def _processes(jobs: int | None) -> int: + """The number of processes ``--jobs`` asks for: 0 is one per core the process may run + on, as ``runtime.jobs`` of ``ctrace``; none is one.""" + + if jobs is None: + return 1 + if jobs == 0: + return _available_cores() + return jobs + + +def _available_cores() -> int: + """The cores the process may run on, where Python can tell its affinity + (``os.process_cpu_count`` from 3.13, ``os.sched_getaffinity`` on Linux before), else + the cores of the machine.""" + + process_cpu_count = getattr(os, "process_cpu_count", None) + if process_cpu_count is not None: + cores: int | None = process_cpu_count() + elif hasattr(os, "sched_getaffinity"): + cores = len(os.sched_getaffinity(0)) + else: + cores = os.cpu_count() + return cores or 1 + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index c951d6e..9a3a56c 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -9,6 +9,7 @@ import concurrent.futures import hashlib import multiprocessing +import sys from collections import Counter from collections.abc import Callable, Iterable, Mapping, Sequence from dataclasses import dataclass, field, replace @@ -360,6 +361,20 @@ def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph: return graph +# ``ProcessPoolExecutor`` refuses more workers on Windows, where it waits on at most 63 +# handles. +_WINDOWS_POOL_LIMIT = 61 + + +def process_pool_size(jobs: int, platform: str | None = None) -> int: + """The processes a pool asked for ``jobs`` holds: no more than ``platform`` (this one + by default) allows.""" + + if (platform or sys.platform) == "win32": + return min(jobs, _WINDOWS_POOL_LIMIT) + return jobs + + def analyze_project( root: Path, plugin_roots: Sequence[Path] = (), @@ -495,7 +510,7 @@ def analyze_project( pool = None if jobs > 1: pool = concurrent.futures.ProcessPoolExecutor( - max_workers=jobs, mp_context=multiprocessing.get_context("spawn") + max_workers=process_pool_size(jobs), mp_context=multiprocessing.get_context("spawn") ) try: for wave in graph.schedule(): diff --git a/tests/test_jobs_option.py b/tests/test_jobs_option.py new file mode 100644 index 0000000..610a3eb --- /dev/null +++ b/tests/test_jobs_option.py @@ -0,0 +1,171 @@ +"""Acceptance tests for issue #205: the concurrency contract of the command line. + +``ctrace`` deprecated ``--async`` for ``-j``/``--jobs`` (CoreTrace/coretrace#158): the +analyzer follows it rather than adding an ``--async`` mode. ``-j N`` is ``--jobs N``; +``0`` analyses in one process per core the process may run on, as ``runtime.jobs`` of +``ctrace`` (its affinity where Python can tell it); the default stays one process. The +pool never holds more processes than the platform allows (61 on Windows). The command always waits for the complete report, which +is the same whatever the number of processes. +""" + +from __future__ import annotations + +import concurrent.futures +import os +import re +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import EXIT_ERROR, main + +PROJECT = { + "app/__init__.py": "", + "app/a.py": "import os\n\n\ndef run(cmd):\n os.system(cmd)\n", + "app/b.py": "def handle(code):\n eval(code)\n", + "app/c.py": "import hashlib\n\n\ndef digest(data):\n return hashlib.md5(data)\n", +} + + +@pytest.fixture +def project(tmp_path: Path) -> Path: + root = tmp_path / "proj" + for name, text in PROJECT.items(): + (root / name).parent.mkdir(parents=True, exist_ok=True) + (root / name).write_text(text, encoding="utf-8") + return root + + +def report(capsys: pytest.CaptureFixture[str], *options: str) -> tuple[int, str]: + status = main(["--check", *options, "--format", "json"]) + out = capsys.readouterr().out + return status, out + + +def test_j_is_jobs_and_the_report_does_not_depend_on_it( + project: Path, capsys: pytest.CaptureFixture[str] +) -> None: + sequential = report(capsys, str(project)) + + assert report(capsys, str(project), "-j", "2") == sequential + assert report(capsys, str(project), "--jobs", "2") == sequential + + +def recorded_jobs(monkeypatch: pytest.MonkeyPatch) -> list[int]: + used: list[int] = [] + analyze = engine.analyze_project + + def recording(*args: object, jobs: int = 1, **kwargs: object) -> engine.ProjectAnalysis: + used.append(jobs) + return analyze(*args, jobs=jobs, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(engine, "analyze_project", recording) + return used + + +def test_zero_jobs_is_one_process_per_available_core( + project: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + sequential = report(capsys, str(project)) + used = recorded_jobs(monkeypatch) + monkeypatch.setattr(os, "process_cpu_count", lambda: 3, raising=False) + monkeypatch.setattr(os, "cpu_count", lambda: 64) + + assert report(capsys, str(project), "-j", "0") == sequential + assert used == [3] + + +def test_zero_jobs_counts_the_cores_the_process_may_run_on_before_python_3_13( + project: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + """Without ``os.process_cpu_count``, the affinity of the process, not every core of + the machine: 2 of 64 here.""" + + used = recorded_jobs(monkeypatch) + monkeypatch.delattr(os, "process_cpu_count", raising=False) + monkeypatch.setattr(os, "sched_getaffinity", lambda pid: {0, 5}, raising=False) + monkeypatch.setattr(os, "cpu_count", lambda: 64) + + report(capsys, str(project), "-j", "0") + assert used == [2] + + +@pytest.mark.parametrize(("cores", "expected"), [(5, 5), (None, 1)]) +def test_zero_jobs_falls_back_to_the_cores_of_the_machine( + project: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, + cores: int | None, + expected: int, +) -> None: + used = recorded_jobs(monkeypatch) + monkeypatch.delattr(os, "process_cpu_count", raising=False) + monkeypatch.delattr(os, "sched_getaffinity", raising=False) + monkeypatch.setattr(os, "cpu_count", lambda: cores) + + report(capsys, str(project), "-j", "0") + assert used == [expected] + + +@pytest.mark.parametrize( + ("jobs", "platform", "size"), + [(64, "win32", 61), (61, "win32", 61), (64, "linux", 64), (2, "win32", 2)], +) +def test_the_pool_holds_no_more_processes_than_the_platform_allows( + jobs: int, platform: str, size: int +) -> None: + """``ProcessPoolExecutor`` refuses more than 61 workers on Windows.""" + + assert engine.process_pool_size(jobs, platform) == size + + +def test_the_analysis_starts_the_pool_at_its_capped_size( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + sizes: list[int] = [] + pool = concurrent.futures.ProcessPoolExecutor + + def recording(max_workers: int, **kwargs: object) -> concurrent.futures.ProcessPoolExecutor: + sizes.append(max_workers) + return pool(max_workers=2, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(engine, "process_pool_size", lambda jobs, platform=None: 7) + monkeypatch.setattr(concurrent.futures, "ProcessPoolExecutor", recording) + + engine.analyze_project(project, [engine.BUNDLED_PLUGINS], jobs=64) + assert sizes == [7] + + +def test_the_default_stays_one_process( + project: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + used = recorded_jobs(monkeypatch) + report(capsys, str(project)) + + assert used == [1] + + +@pytest.mark.parametrize("value", ["-1", "-8"]) +def test_a_negative_number_of_jobs_is_an_error( + project: Path, capsys: pytest.CaptureFixture[str], value: str +) -> None: + assert main(["--check", str(project), "-j", value]) == EXIT_ERROR + assert "--jobs" in capsys.readouterr().err + + +def test_there_is_no_async_option(project: Path, capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit) as exit: + main(["--check", str(project), "--async"]) + + assert exit.value.code == EXIT_ERROR + assert "--async" in capsys.readouterr().err + + +def test_help_names_j_and_zero(capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit): + main(["--help"]) + text = " ".join(capsys.readouterr().out.split()) + + # Python 3.13 prints "-j, --jobs N", earlier versions "-j N, --jobs N". + assert re.search(r"-j( N)?, --jobs N", text) and "0 for one per available core" in text