Multiprocessing for CPU-Bound CLI Tasks

Your command parses 5,000 log files, validates a large dataset, renders hundreds of templates or computes statistics over millions of rows — in pure Python — and it pins one CPU core at 100% for four minutes while the rest of the machine idles. You try a thread pool and it gets slightly slower. That is the global interpreter lock at work: in standard CPython only one thread executes Python bytecode at a time, so threads help with waiting but not with computing. To use every core, a CLI needs multiple processes. This guide shows when that is worth it, how to structure a ProcessPoolExecutor so it works identically on Linux, macOS and Windows, how to keep pickling costs from eating the gains, and how to stop cleanly on Ctrl+C. It is part of the concurrency and async topic.

Prerequisites

• Python 3.10+ and a Typer or Click CLI installed as a package (with a console-script entry point).
• Work that is genuinely CPU-bound in Python. Confirm it first: if --jobs 4 with threads is no faster than one thread, and the CPU is at 100% on one core, it is.
• Work that splits into independent pieces — per file, per record batch, per template.

Why threads do not help here

A thread pool shines when each task mostly waits: during a network call or disk read, the thread releases the GIL and others run. Pure-Python computation never waits, so threads take turns holding the lock and the total time stays the same — plus the overhead of switching.

Two caveats before reaching for processes. First, many "CPU-heavy" libraries do their work in C and release the GIL — hashlib on large buffers, zlib, NumPy, Pillow's resizing, lxml parsing. For those, a thread pool may already scale; measure before switching. Second, free-threaded CPython builds (3.13t and later) remove the GIL entirely. They are promising but still opt-in and not what your users have installed, so for distributed tools processes remain the portable answer.

The recipe

The example command computes per-file statistics over a directory of log files: request counts, error counts and the slowest endpoints. The parsing is regular-expression and string work — pure Python, CPU-bound.

# src/mytool/stats.py
from __future__ import annotations

import os
import re
from collections import Counter
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path

LINE = re.compile(r'"(?P<method>[A-Z]+) (?P<path>\S+) [^"]*" (?P<status>\d{3}) .* (?P<ms>\d+)ms$')


@dataclass
class Stats:
    requests: int = 0
    errors: int = 0
    slow: Counter[str] = field(default_factory=Counter)

    def merge(self, other: Stats) -> Stats:
        self.requests += other.requests
        self.errors += other.errors
        self.slow.update(other.slow)
        return self


def analyse_file(path: str) -> Stats:
    """Runs in a worker process: must be importable and take/return picklable values."""
    stats = Stats()
    with open(path, encoding="utf-8", errors="replace") as fh:
        for line in fh:
            m = LINE.search(line)
            if not m:
                continue
            stats.requests += 1
            if m["status"].startswith("5"):
                stats.errors += 1
            if int(m["ms"]) > 1000:
                stats.slow[m["path"]] += 1
    return stats


def default_jobs() -> int:
    try:
        return len(os.sched_getaffinity(0))   # respects container CPU limits on Linux
    except AttributeError:
        return os.cpu_count() or 1


def analyse(paths: list[Path], jobs: int) -> Stats:
    total = Stats()
    if jobs == 1:
        for p in paths:
            total.merge(analyse_file(str(p)))
        return total
    chunksize = max(1, len(paths) // (jobs * 4))
    with ProcessPoolExecutor(max_workers=jobs) as pool:
        try:
            for stats in pool.map(analyse_file, map(str, paths), chunksize=chunksize):
                total.merge(stats)
        except KeyboardInterrupt:
            pool.shutdown(wait=False, cancel_futures=True)
            raise
    return total

# src/mytool/cli.py
from pathlib import Path

import typer

from mytool.stats import analyse, default_jobs

app = typer.Typer()


@app.callback()
def main() -> None:
    """Log analysis."""


@app.command()
def stats(
    directory: Path = typer.Argument(..., exists=True, file_okay=False),
    jobs: int = typer.Option(0, "--jobs", "-j", min=0, help="Worker processes (0 = one per core)."),
) -> None:
    """Summarise every *.log file under DIRECTORY."""
    paths = sorted(directory.rglob("*.log"))
    try:
        total = analyse(paths, jobs or default_jobs())
    except KeyboardInterrupt:
        typer.echo("interrupted", err=True)
        raise typer.Exit(130)
    typer.echo(f"{len(paths)} files, {total.requests} requests, {total.errors} server errors")
    for path, count in total.slow.most_common(5):
        typer.echo(f"  {count:>6} slow  {path}")


if __name__ == "__main__":
    app()

The rules that make it portable

Worker functions live at module level. The pool sends work to other processes by pickling a reference to the function — its module and name — plus the arguments. Lambdas, nested functions and methods of unpicklable objects cannot be sent. analyse_file is a top-level function in an importable module, so any worker can find it.

Arguments and results are small and simple. Everything crosses the boundary by pickling, so send a path, not the file's contents, and return a compact summary, not every parsed row. Here each worker reads its own file and returns a few integers and a counter. If you find yourself pickling megabytes per task, the copying can cost more than the computation saves.

Chunk the work. With thousands of small items, one round trip per item is dominated by overhead. pool.map(..., chunksize=n) sends items in batches; roughly len(items) / (jobs * 4) keeps workers busy while still balancing load. With submit/as_completed, batch items yourself.

Guard the entry point. On macOS and Windows, and on Linux from Python 3.14, workers are started with spawn or forkserver: a fresh interpreter that imports your main module. If that module runs the CLI at import time, every worker would start the CLI again. The if __name__ == "__main__": guard prevents it, and console-script entry points generated by pip or uv already call your function from a guarded wrapper, so an installed CLI is safe by construction.

Keep --jobs 1 sequential. Running in-process for one job makes debugging normal — breakpoints work, tracebacks are local — and avoids process start-up for tiny inputs.

UX considerations

• Default to the cores the process may actually use. os.sched_getaffinity(0) respects CPU pinning and container limits on Linux, where os.cpu_count() would report the host's 64 cores to a job limited to 2.
• Expect a start-up cost. Spawning workers takes tens to hundreds of milliseconds, since each re-imports your package. For a handful of small files, sequential is faster; consider switching automatically below a size threshold.
• Show progress by completed chunks. With as_completed over batches, a Rich progress bar advanced per batch gives honest progress; pool.map yields in order, so a slow first file stalls the display.
• Memory multiplies. Each worker is a full interpreter with your imports loaded. Eight workers each holding a 500 MB dataset is 4 GB; let --jobs bring it down on smaller machines, and say so in the help text.
• Errors in workers arrive re-raised in the parent, with the original traceback attached as a cause. Catch the expected ones inside the worker and return them as data so one malformed file does not abort the run.

Testing the behaviour

The key tests prove that parallel and sequential runs produce identical results, and that the worker function behaves on its own. Running with jobs=2 in tests exercises the real pickling path; keep the inputs small so the suite stays fast:

# tests/test_stats.py
from pathlib import Path

from mytool.stats import analyse, analyse_file

SAMPLE = [
    '1.2.3.4 - - [18/Sep/2026:10:00:00] "GET /api/users HTTP/1.1" 200 512 12ms',
    '1.2.3.4 - - [18/Sep/2026:10:00:01] "GET /api/report HTTP/1.1" 500 80 2400ms',
    '1.2.3.4 - - [18/Sep/2026:10:00:02] "POST /api/report HTTP/1.1" 200 64 1800ms',
    "garbage line",
]


def write_logs(tmp_path: Path, n: int) -> list[Path]:
    paths = []
    for i in range(n):
        p = tmp_path / f"app-{i}.log"
        p.write_text("\n".join(SAMPLE) + "\n", encoding="utf-8")
        paths.append(p)
    return paths


def test_worker_function(tmp_path):
    [path] = write_logs(tmp_path, 1)
    s = analyse_file(str(path))
    assert (s.requests, s.errors, s.slow["/api/report"]) == (3, 1, 2)


def test_parallel_matches_sequential(tmp_path):
    paths = write_logs(tmp_path, 12)
    seq = analyse(paths, jobs=1)
    par = analyse(paths, jobs=2)
    assert (par.requests, par.errors, par.slow) == (seq.requests, seq.errors, seq.slow)
    assert par.requests == 36

Tests that spawn processes are slower than ordinary unit tests — each pool start re-imports your package — so keep one or two of them and test the logic through the worker function directly. If you use pytest-xdist, process pools inside tests work fine, but avoid combining huge --jobs values with a parallel test runner on small CI machines.

Conclusion

When a CLI is CPU-bound in pure Python, processes are the portable way to use every core. Keep worker functions at module level, send small arguments and return small summaries, chunk many small tasks, guard the entry point, and keep a sequential path for --jobs 1. Default to the CPUs the process is allowed to use, watch memory, and prove with a test that parallel and sequential runs agree. For work that mostly waits rather than computes, a thread pool is simpler and just as fast.

Frequently asked questions

Related

• Up: Concurrency and async in Python CLIs
• Parallelising CLI work with thread pools
• Cancelling async tasks on Ctrl+C
• Profiling Python CLI startup time
• Processing large files and NDJSON streams