feat(cli): add -j and --jobs 0 for one process per core instead of an --async mode (#205)

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

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())

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():

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