Running Jobs Concurrently

Part II · Module 27 · Chapter 27.3 · Running Jobs Concurrently

1. Processes, Not Threads

import os

CHOICES = [
    ("threads", "shared memory", "blocked by the GIL for Python work", "fine for waiting"),
    ("processes", "separate memory", "true parallelism", "the right choice here"),
]

print(f"{'model':<11} {'memory':<16} {'CPU work':<34} verdict")
for name, mem, cpu, verdict in CHOICES:
    print(f"{name:<11} {mem:<16} {cpu:<34} {verdict}")

print()
print("what a regression worker actually does:")
print("   calls subprocess.run and waits for a simulator")
print("   parses a log")
print("   writes a record")
print()
print("so threads would mostly work:", True)
print("and processes are still better:", True)
print("why:", "a crash in one worker cannot take the parent with it")

model       memory           CPU work                           verdict
threads     shared memory    blocked by the GIL for Python work fine for waiting
processes   separate memory  true parallelism                   the right choice here

what a regression worker actually does:
   calls subprocess.run and waits for a simulator
   parses a log
   writes a record

so threads would mostly work: True
and processes are still better: True
why: a crash in one worker cannot take the parent with it

The honest version of the threads-versus-processes argument for this workload: a regression worker spends almost all its time blocked in subprocess.run waiting for a simulator, and a thread blocked on a subprocess releases the GIL. Threads would mostly work.

Processes are still the right choice, for reasons that are not about speed. A worker that segfaults, exhausts memory or calls os._exit takes its own process down and nothing else; in a thread pool it takes the parent and the whole regression. Isolation is the same argument Chapter 26.4 made about directories, one level up.

2. The Worker Function

import os
from pathlib import Path


def run_job(job):
    """One job. Must be importable and its arguments picklable."""
    test, seed = job
    run_dir = Path("runs") / f"{test}_{seed}"
    run_dir.mkdir(parents=True, exist_ok=True)
    (run_dir / "sim.log").write_text(f"ran {test} seed {seed}\n", encoding="utf-8")
    status = "FAIL" if seed % 4 == 3 else "PASS"
    return {"test": test, "seed": seed, "status": status,
            "run_dir": str(run_dir), "pid_differs": os.getpid()}

from worker import run_job

JOBS = [("axi_smoke", 1), ("axi_burst", 3)]
for job in JOBS:
    r = run_job(job)
    print(f"{r['test']:<11} seed {r['seed']}  {r['status']:<5} {r['run_dir']}")

print()
print("it takes one argument      :", "a job tuple")
print("it returns a plain dict    :", True)
print("it writes only in its dir  :", "Chapter 26.4")
print("it does not print or exit  :", "Chapter 25.1 §8")

axi_smoke   seed 1  PASS  runs/axi_smoke_1
axi_burst   seed 3  FAIL  runs/axi_burst_3

it takes one argument      : a job tuple
it returns a plain dict    : True
it writes only in its dir  : Chapter 26.4
it does not print or exit  : Chapter 25.1 §8

Four properties, and every one of them is a rule from an earlier chapter that now becomes load-bearing.

One argument, because ex.map passes one. A real runner needs the simulator, the waves flag and the timeout too, and those arrive via functools.partial or by putting them in the job — not as extra positional parameters.

A plain, picklable return value. Chapter 25.1's frozen dataclass is picklable too, provided it is defined at module scope; the dict here keeps the example focused.

It writes only inside its own run directory — Chapter 26.4, which is what makes concurrency safe at all.

It does not print, exit or raise for a failing test — Chapter 25.1 Section 8. In a pool those prohibitions get sharper: a print from sixteen workers interleaves into unreadable mush, and sys.exit in a worker is a worker that simply vanishes.

3. The Annotation

with ProcessPoolExecutor(max_workers=n) as ex:
    for res in ex.map(run_job, jobs):
        results.append(res)

def square(n):
    return (n, n * n)

from concurrent.futures import ProcessPoolExecutor
from worker import square

JOBS = list(range(8))

if __name__ == "__main__":
    results = []
    with ProcessPoolExecutor(max_workers=4) as ex:
        for res in ex.map(square, JOBS):
            results.append(res)

    print("submitted:", JOBS)
    print("received :", [n for n, _ in results])
    print("in submission order:", [n for n, _ in results] == JOBS)
    print("count    :", len(results))

submitted: [0, 1, 2, 3, 4, 5, 6, 7]
received : [0, 1, 2, 3, 4, 5, 6, 7]
in submission order: True
count    : 8

ex.map yields in submission order, not completion order, which is a genuinely useful property: the results line up with the job list regardless of which worker finished first.

It is also worth not relying on. Chapter 27.2's comparison is a set comparison precisely so that ordering is irrelevant, and Chapter 26.6 reconciles by job identity rather than by position. The ordering guarantee is a convenience; the code should not need it.

The with block is doing real work. On exit it waits for every outstanding job and shuts the pool down, so a result cannot be missed by falling off the end of the loop — and an exception inside the block still tears the pool down rather than leaving sixteen orphaned processes.

4. The Main Guard Is Not Optional

import multiprocessing

print("start method on this platform:", multiprocessing.get_start_method())
print()
print("what 'spawn' means:")
print("   each worker starts a fresh interpreter")
print("   which IMPORTS the module that defined the work")
print("   so anything at module scope runs again, in every worker")
print()
print("a pool created at module scope therefore:")
print("   creates a pool in the parent")
print("   each worker re-imports and creates a pool")
print("   each of those workers re-imports and creates a pool")
print()
print("the guard that stops it:", 'if __name__ == "__main__":')
print("why it works:", "__name__ is not '__main__' in a re-imported module")

start method on this platform: spawn

what 'spawn' means:
   each worker starts a fresh interpreter
   which IMPORTS the module that defined the work
   so anything at module scope runs again, in every worker

a pool created at module scope therefore:
   creates a pool in the parent
   each worker re-imports and creates a pool
   each of those workers re-imports and creates a pool

the guard that stops it: if __name__ == "__main__":
why it works: __name__ is not '__main__' in a re-imported module

if __name__ == "__main__": around the pool is a correctness requirement on any spawn platform, which since Python 3.8 means macOS and Windows by default.

Without it, the failure is not a tidy exception. Each worker re-imports the module, reaches the pool-creation line, and creates its own pool — recursively, until the machine runs out of processes. It is the most dramatic failure in this module and it does not happen on Linux with the fork start method, which is how it reaches production: it works on the developer's machine and does not work on the compute farm, or the reverse.

The same re-import is why the worker function must live in an importable module. A function defined inside if __name__ == "__main__": cannot be found by a worker that re-imports the file, because that branch does not run there.

5. ex.map Loses Results

def boom(n):
    if n == 3:
        raise ValueError(f"job {n} exploded")
    return (n, n * n)

from concurrent.futures import ProcessPoolExecutor
from worker import boom

JOBS = list(range(8))

if __name__ == "__main__":
    collected = []
    try:
        with ProcessPoolExecutor(max_workers=4) as ex:
            for res in ex.map(boom, JOBS):
                collected.append(res[0])
    except ValueError as exc:
        print("raised  :", exc)

    print("submitted:", len(JOBS))
    print("kept     :", len(collected), collected)
    print("lost     :", len(JOBS) - len(collected))
    print()
    print("of the lost ones, how many actually failed:", 1)
    print("how many succeeded and were discarded     :", len(JOBS) - len(collected) - 1)

raised  : job 3 exploded
submitted: 8
kept     : 3 [0, 1, 2]
lost     : 5

of the lost ones, how many actually failed: 1
how many succeeded and were discarded     : 4

Eight jobs, one failure, five results lost — four of which succeeded.

ex.map returns an iterator that re-raises a worker's exception at the point the failed item would have been yielded. The loop stops there, the with block tears the pool down, and everything after job 3 is discarded regardless of whether it worked.

In a regression that is catastrophic and silent in a very specific way. Under Chapter 26.1's invariant those four jobs have no result, so they are counted as not_run — and they are not: they ran, they passed, and the accounting now says four jobs went missing when one job raised. The summary balances, which is worse, because it looks correct.

The fix is Chapter 27.4's annotation and it is worth seeing now:

def boom(n):
    if n == 3:
        raise ValueError(f"job {n} exploded")
    return (n, n * n)

from concurrent.futures import ProcessPoolExecutor, as_completed
from worker import boom

JOBS = list(range(8))

if __name__ == "__main__":
    good, failed = [], []
    with ProcessPoolExecutor(max_workers=4) as ex:
        futures = {ex.submit(boom, n): n for n in JOBS}
        for fut in as_completed(futures):
            job = futures[fut]
            try:
                good.append(fut.result()[0])
            except Exception as exc:
                failed.append((job, type(exc).__name__))

    print("submitted      :", len(JOBS))
    print("succeeded      :", len(good))
    print("failed         :", failed)
    print("accounted for  :", len(good) + len(failed))
    print("nothing lost   :", len(good) + len(failed) == len(JOBS))
    print()
    print("succeeded jobs (sorted):", sorted(good))

submitted      : 8
succeeded      : 7
failed         : [(3, 'ValueError')]
accounted for  : 8
nothing lost   : True

succeeded jobs (sorted): [0, 1, 2, 4, 5, 6, 7]

All eight accounted for, seven results and one captured failure.

Two things about that loop. as_completed yields futures in completion order, which depends on scheduling — so the successes are printed sorted, and nowhere in this chapter is a completion order published as a list. Chapter 26.4 Section 2 established what publishing a scheduling-dependent value costs.

And futures is a dict from future to job, which is what lets the except clause say which job failed. A bare list of futures records the exception and loses its owner, which under Chapter 26.6's reconciliation is an unexplained not_run.

6. What Crosses the Process Boundary

import pickle


def module_level(x):
    return x


class Job:
    def __init__(self, test):
        self.test = test


print("a module-level function  :", type(pickle.dumps(module_level)).__name__)

try:
    pickle.dumps(lambda x: x)
except Exception as exc:
    print("a lambda                 :", type(exc).__name__)

def outer():
    def inner(x):
        return x
    return inner


try:
    pickle.dumps(outer())
except Exception as exc:
    print("a nested function        :", type(exc).__name__)
    print("   its qualified name    :", outer().__qualname__)

print("a plain dict             :", type(pickle.dumps({"a": 1})).__name__)
print("a tuple of strings/ints  :", type(pickle.dumps(("axi", 1))).__name__)

try:
    pickle.dumps(open(__file__))
except Exception as exc:
    print("an open file handle      :", type(exc).__name__)

a module-level function  : bytes
a lambda                 : PicklingError
a nested function        : AttributeError
   its qualified name    : outer.<locals>.inner
a plain dict             : bytes
a tuple of strings/ints  : bytes
an open file handle      : TypeError

The function and every argument cross a process boundary by pickle, which rules out three things people reach for.

A lambda or a genuinely nested function cannot be pickled, because pickling a function stores its qualified name and re-imports it in the worker — and outer.<locals>.inner is not a name any import can reach — and note the two failures arrive as different exception types, PicklingError for the lambda and AttributeError for the nested function, so a handler that catches only pickle.PicklingError misses half of them. Note that nested means lexically inside another function: a def inside an if or a try at module scope is still a module-level name and pickles perfectly well, which is an easy way to convince yourself the restriction is looser than it is. functools.partial over a module-level function is picklable and is the usual way to bind extra arguments.

An open file handle cannot cross either. A worker must open its own; passing a path is the fix, which Chapter 23.7 Section 3 already required for a different reason.

Only the exception types are printed above: the PicklingError message embeds a memory address, which would differ on every run and must never appear in published output.

7. The Shape of a Pool

8. Choosing the Worker Count

import os

CORES = 64
MEMORY_GB = 256
PER_JOB_GB = 8
LICENCES = 12
DISK_MB_PER_SEC = 400
PER_JOB_MB_PER_SEC = 25

limits = {
    "cores": CORES,
    "memory": MEMORY_GB // PER_JOB_GB,
    "licences": LICENCES,
    "disk bandwidth": DISK_MB_PER_SEC // PER_JOB_MB_PER_SEC,
}

for name, value in limits.items():
    print(f"{name:<16} allows {value:>3} workers")

binding = min(limits, key=limits.get)
print()
print("workers           :", min(limits.values()))
print("binding constraint:", binding)
print("cores available   :", CORES, "— not the limit")
print()
print("the optimistic answer would have been:", CORES)
print("over-subscription factor if used      :", f"{CORES / min(limits.values()):.1f}x")

cores            allows  64 workers
memory           allows  32 workers
licences         allows  12 workers
disk bandwidth   allows  16 workers

workers           : 12
binding constraint: licences
cores available   : 64 — not the limit

the optimistic answer would have been: 64
over-subscription factor if used      : 5.3x

Sixty-four cores and twelve workers, because the licence pool is the binding constraint — which is the usual answer and never the one os.cpu_count() gives.

max_workers=os.cpu_count() is the optimistic default and it is wrong in both directions here. Running sixty-four jobs against twelve licences means fifty-two jobs waiting for or failing to acquire one; running them against thirty-two jobs' worth of memory means the machine swaps or the kernel starts killing processes.

Report the binding constraint, not just the number. workers=12 tells somebody what to type; workers=12, limited by licences tells them what to buy. Chapter 27.5 is this calculation in full, including what to do when the pool is exhausted mid-run.

All the figures above are configured inputs, not measurements — the per-job memory and disk rates are the numbers you would get from Chapter 25.7's soak with resource monitoring, and they are a property of your design rather than of Python.

9. The Runner, Assembled

from pathlib import Path


def run_job(job):
    """One job, in its own directory. Returns a plain dict; never raises for a failing test."""
    test, seed = job
    run_dir = Path("runs") / f"{test}_{seed}"
    run_dir.mkdir(parents=True, exist_ok=True)
    (run_dir / "sim.log").write_text(f"ran {test} {seed}\n", encoding="utf-8")
    if seed == 99:
        raise RuntimeError("the worker itself broke")
    return {"test": test, "seed": seed,
            "status": "FAIL" if seed % 4 == 3 else "PASS",
            "run_dir": str(run_dir)}

from collections import Counter
from concurrent.futures import ProcessPoolExecutor, as_completed
from worker import run_job

JOBS = [("axi_smoke", s) for s in (1, 2, 3, 4)] + [("axi_burst", 99)]


def run_all(jobs, workers=1):
    """Collect a result for every job, including the ones whose worker raised."""
    results, infra = {}, {}
    with ProcessPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(run_job, job): job for job in jobs}
        for fut in as_completed(futures):
            job = futures[fut]
            try:
                results[job] = fut.result()
            except Exception as exc:
                infra[job] = {"test": job[0], "seed": job[1], "status": "ERROR",
                              "reason": f"worker raised {type(exc).__name__}"}
    results.update(infra)
    return results


if __name__ == "__main__":
    out = run_all(JOBS, workers=3)

    for job in sorted(out, key=lambda j: (j[0], j[1])):
        r = out[job]
        print(f"{r['test']:<11} {r['seed']:>3}  {r['status']:<6} {r.get('reason', '')}")

    counts = Counter(r["status"] for r in out.values())
    print()
    print("jobs submitted :", len(JOBS))
    print("results        :", len(out))
    print("accounted for  :", len(out) == len(JOBS))
    print("counts         :", dict(sorted(counts.items())))

axi_burst    99  ERROR  worker raised RuntimeError
axi_smoke     1  PASS   
axi_smoke     2  PASS   
axi_smoke     3  FAIL   
axi_smoke     4  PASS   

jobs submitted : 5
results        : 5
accounted for  : True
counts         : {'ERROR': 1, 'FAIL': 1, 'PASS': 3}

Five decisions, and the second is the one that keeps Chapter 26.1 intact.

The results are a dict keyed by job, so a result can always be attributed, and the output is sorted for display rather than relying on completion order.

A worker that raises becomes status="ERROR", not a missing result. Chapter 23.9 Section 7's classification: the worker broke, the design was never exercised, and the job must still appear in the summary. Without this the job is not_run and the cause is lost.

workers=1 is the default, which makes this the same function Chapter 27.2's baseline came from. One code path, one parameter, and the comparison tests exactly one change.

run_job lives in an importable module, because a spawn-platform worker re-imports it.

The with block waits for everything, so run_all cannot return before every future has been resolved one way or the other.

What this does not yet do is Chapter 27.4's subject: a worker killed by the kernel rather than raising, a job whose run directory collided, a pool that exhausted the licence server halfway through. Those are the failures concurrency adds rather than the ones it inherits.

10. Predict, Run, Observe

def sq(n):
    return n * n


def bang(n):
    if n == 2:
        raise ValueError("no")
    return n

import pickle
from concurrent.futures import ProcessPoolExecutor, as_completed
from worker import sq, bang

if __name__ == "__main__":
    J = [0, 1, 2, 3]

    # A
    with ProcessPoolExecutor(max_workers=2) as ex:
        print("A:", list(ex.map(sq, J)))

    # B
    got = []
    try:
        with ProcessPoolExecutor(max_workers=2) as ex:
            for r in ex.map(bang, J):
                got.append(r)
    except ValueError:
        print("B:", got)

    # C
    ok, bad = [], []
    with ProcessPoolExecutor(max_workers=2) as ex:
        futs = {ex.submit(bang, n): n for n in J}
        for f in as_completed(futs):
            try:
                ok.append(f.result())
            except ValueError:
                bad.append(futs[f])
    print("C:", sorted(ok), bad)

    # D
    print("D:", len(ok) + len(bad) == len(J))

    # E
    print("E:", type(pickle.dumps(sq)).__name__)

    # F
    try:
        pickle.dumps(lambda x: x)
    except Exception as exc:
        print("F:", type(exc).__name__)

    # G
    import multiprocessing
    print("G:", multiprocessing.get_start_method() in ("spawn", "fork", "forkserver"))

    # H
    print("H:", __name__)

The verified output:

A: [0, 1, 4, 9]
B: [0, 1]
C: [0, 1, 3] [2]
D: True
E: bytes
F: PicklingError
G: True
H: __main__

A — map yields in submission order, all four.

B — one failure and only two results survive. Section 5.

C — as_completed with a per-future try keeps three results and names the failing job.

D — and nothing is unaccounted for.

E — a module-level function pickles.

F — a lambda does not.

G — the start method is one of the three; which one decides whether the missing guard is fatal.

H — inside the guard, __name__ is "__main__"; in a re-imported worker it is not, which is the whole mechanism.

11. Exercises

Exercise 1 — threads or processes. Say what a regression worker spends its time doing, and why processes are still preferred.

Exercise 2 — the isolation argument. Relate the choice to Chapter 26.4's directories.

Exercise 3 — four properties. Check your own worker function against all four and fix any that fail.

Exercise 4 — extra arguments. Bind a simulator and a timeout to a one-argument worker using functools.partial.

Exercise 5 — submission order. Confirm ex.map preserves it, then say why your code should not depend on it.

Exercise 6 — the with block. Say what it guarantees on exit and what would leak without it.

Exercise 7 — the start method. Report yours, and say whether a missing main guard would be fatal on it.

Exercise 8 — the recursion. Describe exactly what happens without the guard on a spawn platform.

Exercise 9 — works here, not there. Explain how this bug reaches production.

Exercise 10 — importable. Say why a worker defined inside the guard cannot be used.

Exercise 11 — count the loss. Reproduce the ex.map failure and report submitted, kept and lost.

Exercise 12 — the silent part. Say what Chapter 26.1's invariant reports for the four discarded successes.

Exercise 13 — as_completed. Rewrite it and confirm nothing is lost.

Exercise 14 — future to job. Say what a bare list of futures loses when one raises.

Exercise 15 — do not publish the order. Explain why completion order must not appear in a report or a test.

Exercise 16 — picklability. Try a lambda, a nested function and an open file. Record the exception types only.

Exercise 17 — why types only. Say what is in the PicklingError message that must not be published.

Exercise 18 — four limits. Compute the worker count your machine, memory, licences and disk each allow.

Exercise 19 — the binding one. Report it by name, and say what that tells somebody that the number does not.

Exercise 20 — one code path. Confirm your run_all(workers=1) produces Chapter 27.2's baseline.

12. Summary

• A regression worker is mostly blocked on a subprocess, so threads would mostly work — and processes are still right, because a worker crash cannot take the parent down.
• The worker takes one argument, returns a picklable value, writes only in its own directory, and never prints or exits.
• ex.map yields in submission order, which is convenient and should not be depended on; Chapter 27.2 compares sets and Chapter 26.6 reconciles by identity.
• The with block waits for every outstanding job and tears the pool down on an exception.
• if __name__ == "__main__": is a correctness requirement on a spawn platform, not a style point — without it each worker re-imports and creates its own pool, recursively.
• It works on fork and fails on spawn, which is how it reaches production.
• The worker function must live in an importable module, for the same reason.
• ex.map raises on the first failing job and discards every result after it — eight jobs, one failure, four successes thrown away.
• Those four are then counted as not_run and the summary balances, which makes it look correct.
• as_completed with a per-future try accounts for all of them, which is Chapter 27.4's annotation.
• Key the futures by job, or an exception is recorded with no owner.
• Never publish a completion order — it is scheduling-dependent, like Chapter 26.4 Section 2's interleaving.
• The function and every argument cross by pickle, so no lambdas, no nested functions, no open file handles.
• Print exception types, not pickling messages, which embed a memory address.
• A worker that raises becomes status="ERROR", never a missing result.
• Choose the worker count from the binding constraint — usually licences, never os.cpu_count() — and report which constraint it was.
• Default workers=1, so the baseline and the parallel run come from one code path.

The next chapter is about everything that goes wrong once this is running.