From 803300a78b696f3f48074be6e7251d5db88ed987 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 8 Oct 2026 05:31:34 +0200 Subject: [PATCH 1/2] feat(cli): add -j and --jobs 0 for one process per core instead of an --async mode (#205) --- docs/usage.md | 10 +++- src/coretrace_python/cli.py | 21 +++++-- tests/test_jobs_option.py | 109 ++++++++++++++++++++++++++++++++++++ 3 files changed, 133 insertions(+), 7 deletions(-) create mode 100644 tests/test_jobs_option.py diff --git a/docs/usage.md b/docs/usage.md index 61e4de9..749e415 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,12 @@ 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, and the default is one. +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..552bef0 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,16 @@ 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 getattr(os, "process_cpu_count", os.cpu_count)() or 1 + return jobs + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/tests/test_jobs_option.py b/tests/test_jobs_option.py new file mode 100644 index 0000000..ed43b00 --- /dev/null +++ b/tests/test_jobs_option.py @@ -0,0 +1,109 @@ +"""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 available core, as ``runtime.jobs`` of ``ctrace``; +the default stays one process. The command always waits for the complete report, which +is the same whatever the number of processes. +""" + +from __future__ import annotations + +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 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: 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) + cores = getattr(os, "process_cpu_count", os.cpu_count)() or 1 + + assert report(capsys, str(project), "-j", "0") == sequential + assert used == [cores] + + +def test_the_default_stays_one_process( + project: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + 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) + 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 From ac766f8353fd78608db5e265062aeaf63cfac40e Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 8 Oct 2026 05:59:34 +0200 Subject: [PATCH 2/2] fix(cli): count the cores the process may run on and cap the pool at the platform's limit (#205) --- docs/usage.md | 5 +- src/coretrace_python/cli.py | 17 ++++++- src/coretrace_python/engine.py | 17 ++++++- tests/test_jobs_option.py | 92 ++++++++++++++++++++++++++++------ 4 files changed, 113 insertions(+), 18 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index 749e415..e6cb6cb 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -731,7 +731,10 @@ module is served from the cache, so editing one file re-analyses that file and i importers only. Entries are plain data; an unreadable entry is recomputed. `--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, and the default is one. +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` diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index 552bef0..df39187 100644 --- a/src/coretrace_python/cli.py +++ b/src/coretrace_python/cli.py @@ -340,9 +340,24 @@ def _processes(jobs: int | None) -> int: if jobs is None: return 1 if jobs == 0: - return getattr(os, "process_cpu_count", os.cpu_count)() or 1 + 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 index ed43b00..610a3eb 100644 --- a/tests/test_jobs_option.py +++ b/tests/test_jobs_option.py @@ -2,13 +2,15 @@ ``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 available core, as ``runtime.jobs`` of ``ctrace``; -the default stays one process. The command always waits for the complete report, which +``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 @@ -50,10 +52,7 @@ def test_j_is_jobs_and_the_report_does_not_depend_on_it( assert report(capsys, str(project), "--jobs", "2") == sequential -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)) +def recorded_jobs(monkeypatch: pytest.MonkeyPatch) -> list[int]: used: list[int] = [] analyze = engine.analyze_project @@ -62,23 +61,86 @@ def recording(*args: object, jobs: int = 1, **kwargs: object) -> engine.ProjectA return analyze(*args, jobs=jobs, **kwargs) # type: ignore[arg-type] monkeypatch.setattr(engine, "analyze_project", recording) - cores = getattr(os, "process_cpu_count", os.cpu_count)() or 1 + 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 == [cores] + assert used == [3] -def test_the_default_stays_one_process( +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: - used: list[int] = [] - analyze = engine.analyze_project + """Without ``os.process_cpu_count``, the affinity of the process, not every core of + the machine: 2 of 64 here.""" - def recording(*args: object, jobs: int = 1, **kwargs: object) -> engine.ProjectAnalysis: - used.append(jobs) - return analyze(*args, jobs=jobs, **kwargs) # type: ignore[arg-type] + 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) - monkeypatch.setattr(engine, "analyze_project", recording) + 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]