Python Part II · Module 27 · Parallel Regression
What Concurrency Breaks
Chapter 27.1 counted four hazards that concurrency adds and none that it removes. This is the chapter that meets them. Two are latent bugs Module 26 already prevented and which would otherwise activate the moment two jobs overlap; one is a failure mode with no serial equivalent at all, where a single worker being killed takes every other job down with it; and one is the quiet one, where the fix from the previous chapter turns out to record that something broke without recording what.
Advanced19 min readPythonPart IIAutomationRegressionVerification Flow
Part II · Module 27 · Chapter 27.4 · What Concurrency Breaks
1. Almost Nothing Here Is Reproducible
TRIALS = [
("worker-local counter values", "[1..8]", "[1..8]", "[1,1,2,3,4,5,6,7]", False),
("the parent's counter", "0", "0", "0", True),
("successes before a pool break", "1", "2", "2", False),
("successes plus failures", "8", "8", "8", True),
("the exception type set", "{BrokenProcessPool}", "{BrokenProcessPool}",
"{BrokenProcessPool}", True),
]
print(f"{'observation':<32} {'t0':<20} {'t1':<20} {'t2':<20} stable")
for name, a, b, c, stable in TRIALS:
print(f"{name:<32} {a:<20} {b:<20} {c:<20} {stable}")
print()
print("observations that vary :", sum(1 for *_, s in TRIALS if not s))
print("safe to assert in a test:", [n for n, *_, s in TRIALS if s])observation t0 t1 t2 stable
worker-local counter values [1..8] [1..8] [1,1,2,3,4,5,6,7] False
the parent's counter 0 0 0 True
successes before a pool break 1 2 2 False
successes plus failures 8 8 8 True
the exception type set {BrokenProcessPool} {BrokenProcessPool} {BrokenProcessPool} True
observations that vary : 2
safe to assert in a test: ["the parent's counter", 'successes plus failures', 'the exception type set']Those are three real trials of the experiments in Sections 3 and 4, and the table is the chapter's first lesson rather than a preamble: in a parallel run, most of the numbers you would naturally reach for are properties of the scheduler.
Chapter 26.4 Section 2 met this once, with interleaved log lines. Here it is pervasive — how many jobs a worker handled, which succeeded before a failure, which arrived first. Everything this chapter publishes is a total, a type set, or the parent's own view, and the same rule applies to the tests you write around your own pool: assert the invariant, never the schedule.
2. The Annotation Loses the Job
for fut in as_completed(futs):
try:
results.append(fut.result())
except Exception as e:
record_infra_failure(e)This is Chapter 27.3 Section 5's fix — every future is resolved, nothing is discarded when one fails — and it has a defect of its own.
def work(n):
if n == 2:
raise ValueError("job 2 failed")
return nfrom concurrent.futures import ProcessPoolExecutor, as_completed
from w import work
if __name__ == "__main__":
JOBS = [0, 1, 2, 3]
# the annotation's shape: a LIST of futures
results, failures = [], []
with ProcessPoolExecutor(max_workers=2) as ex:
futs = [ex.submit(work, n) for n in JOBS]
for fut in as_completed(futs):
try:
results.append(fut.result())
except Exception as exc:
failures.append(type(exc).__name__)
print("results :", sorted(results))
print("failures :", failures)
print("which job failed:", "not recoverable from this loop")
print()
# Chapter 27.3's shape: a DICT from future to job
results, failures = [], []
with ProcessPoolExecutor(max_workers=2) as ex:
futs = {ex.submit(work, n): n for n in JOBS}
for fut in as_completed(futs):
job = futs[fut]
try:
results.append(fut.result())
except Exception as exc:
failures.append((job, type(exc).__name__))
print("results :", sorted(results))
print("failures :", failures)
print("which job failed:", failures[0][0])results : [0, 1, 3]
failures : ['ValueError']
which job failed: not recoverable from this loop
results : [0, 1, 3]
failures : [(2, 'ValueError')]
which job failed: 2A list of futures records that something failed; a dict records what.
The consequence is specific. Chapter 26.6 reconciles results against the job list by identity, so a job with no result becomes not_run — and the orphaned ValueError sits in a log beside it with nothing connecting the two. The regression reports one unexplained missing job and one unexplained exception, and pairing them is a human task.
Four characters of difference, and it is the difference between a diagnosable failure and two mysteries.
3. Workers Do Not Share Memory
COUNTER = 0
def bump(n):
global COUNTER
COUNTER += 1
return COUNTERfrom concurrent.futures import ProcessPoolExecutor
import w2
from w2 import bump
if __name__ == "__main__":
print("parent COUNTER before:", w2.COUNTER)
with ProcessPoolExecutor(max_workers=4) as ex:
seen = list(ex.map(bump, range(8)))
print("parent COUNTER after :", w2.COUNTER)
print("jobs run :", len(seen))
print("every worker reported a value:", all(isinstance(v, int) for v in seen))
print()
print("the parent's global was modified:", w2.COUNTER != 0)
print("the values the workers saw :", "depend on which worker ran what")parent COUNTER before: 0
parent COUNTER after : 0
jobs run : 8
every worker reported a value: True
the parent's global was modified: False
the values the workers saw : depend on which worker ran whatEight increments and the parent's counter is zero.
Each worker is a separate process with its own copy of the module, so global COUNTER modifies the worker's copy. The values the workers report depend on how many jobs each happened to handle, which is why they are not printed — Section 1's rule.
That breaks three patterns that work perfectly in a serial regression:
PATTERNS = [
("a module-level results list", "appended by each job", "the parent sees an empty list"),
("a counter of failures", "incremented per failure", "the parent sees zero"),
("a cache built up as jobs run", "shared across jobs", "each worker builds its own"),
("a lazily opened log file", "opened once", "opened once per worker"),
]
print(f"{'pattern':<30} {'what it does serially':<26} what happens in a pool")
for name, serial, parallel in PATTERNS:
print(f"{name:<30} {serial:<26} {parallel}")
print()
print("how many fail loudly:", 0)
print("the fix :", "return it from the worker; let the parent combine")pattern what it does serially what happens in a pool
a module-level results list appended by each job the parent sees an empty list
a counter of failures incremented per failure the parent sees zero
a cache built up as jobs run shared across jobs each worker builds its own
a lazily opened log file opened once opened once per worker
how many fail loudly: 0
the fix : return it from the worker; let the parent combineNone of the four produces an error. The parent simply observes nothing happening, which reads as "the jobs did not run" and is the opposite of the truth.
The fix is the discipline Chapter 25.1 Section 8 already imposed: a worker returns a value and writes only inside its own directory. Anything it wants the parent to know is in the return value. The third row is the only one with a cost — Chapter 25.3 Section 7's compile cache genuinely is per-worker now, which is a real inefficiency and not a correctness problem, since the cache key covers the inputs.
4. One Killed Worker Breaks the Pool
import os
import time
def fragile(n):
if n == 2:
os._exit(1) # killed, not raised — an OOM kill looks like this
time.sleep(0.05)
return nfrom concurrent.futures import ProcessPoolExecutor, as_completed
from w3 import fragile
if __name__ == "__main__":
JOBS = list(range(8))
results, failures = [], []
with ProcessPoolExecutor(max_workers=2) as ex:
futs = {ex.submit(fragile, n): n for n in JOBS}
for fut in as_completed(futs):
job = futs[fut]
try:
results.append(fut.result())
except Exception as exc:
failures.append((job, type(exc).__name__))
print("jobs submitted :", len(JOBS))
print("accounted for :", len(results) + len(failures))
print("nothing lost :", len(results) + len(failures) == len(JOBS))
print()
print("distinct error types:", sorted({t for _, t in failures}))
print("failures reported :", len(failures) > 1)
print("jobs that actually misbehaved:", 1)jobs submitted : 8
accounted for : 8
nothing lost : True
distinct error types: ['BrokenProcessPool']
failures reported : True
jobs that actually misbehaved: 1One job called os._exit and the pool reported failures for most of the others.
This is the failure mode with no serial equivalent, and three things about it are worth separating.
A killed worker does not raise for its own job. There is no exception to propagate — the process is gone. fragile(2) produces no ValueError, no traceback, nothing attributable to job 2 at all.
It raises BrokenProcessPool for every outstanding future. The executor cannot know what the dead worker was doing or whether the pool is still sound, so it fails everything pending. The exact split between results and failures varies per run, which is why only the total is published.
except Exception catches it, which is the dangerous part. The annotation's handler calls record_infra_failure(e) for all of them, and six or seven perfectly healthy jobs are now recorded as infrastructure failures. Chapter 26.1's invariant still balances; the numbers are simply wrong, and wrong in Chapter 26.1 Section 4's pessimistic direction.
What causes a worker to be killed rather than to raise:
CAUSES = [
("the OOM killer", "n workers x per-job memory exceeded the machine", "Chapter 27.5"),
("a segfault in a C extension", "or in the simulator, if run in-process", "rare here"),
("os._exit or os.kill in flow code", "a worker calling sys.exit's blunt cousin", "a bug"),
("the scheduler killing the parent's children", "a job-level limit", "Chapter 26.9"),
]
for cause, what, where in CAUSES:
print(f"{cause:<42} {what:<48} {where}")
print()
print("the most common by far:", CAUSES[0][0])
print("which makes it :", "a worker-count problem, not a code problem")the OOM killer n workers x per-job memory exceeded the machine Chapter 27.5
a segfault in a C extension or in the simulator, if run in-process rare here
os._exit or os.kill in flow code a worker calling sys.exit's blunt cousin a bug
the scheduler killing the parent's children a job-level limit Chapter 26.9
the most common by far: the OOM killer
which makes it : a worker-count problem, not a code problemThe usual cause is memory, which makes the usual fix a smaller worker count rather than better error handling — Chapter 27.5's subject.
5. The Latent Collisions Activate
HAZARDS = [
("two jobs, one run directory", "26.4 §5", "name it from the full job identity"),
("a shared latest.log", "26.4 §8", "never write outside the run directory"),
("a shared scratch path", "24.3 §7", "put the test and seed in it"),
("a shared compile output directory", "25.3 §7", "key the build dir by the cache key"),
]
print(f"{'hazard':<36} {'prevented in':<10} by")
for hazard, where, fix in HAZARDS:
print(f"{hazard:<36} {where:<10} {fix}")
print()
print("how many were introduced by concurrency:", 0)
print("how many were ACTIVATED by it :", len(HAZARDS))
print("what a serial run of the broken version does:", "passes, every time")hazard prevented in by
two jobs, one run directory 26.4 §5 name it from the full job identity
a shared latest.log 26.4 §8 never write outside the run directory
a shared scratch path 24.3 §7 put the test and seed in it
a shared compile output directory 25.3 §7 key the build dir by the cache key
how many were introduced by concurrency: 0
how many were ACTIVATED by it : 4
what a serial run of the broken version does: passes, every timeConcurrency introduced none of these and activated all four.
That distinction is the practical heart of the chapter. A regression with a shared latest.log is already broken when it is written; serially the bug is unobservable, because two jobs never overlap in time. The defect sits in the code for months, passing every test, and then somebody sets workers=8.
The consequence for debugging is specific and worth internalising: when parallelising surfaces a bug, the bug is almost never in the parallel code. It is in the thing that was always wrong, and Chapter 27.2's baseline comparison is what tells you which jobs it affects.
6. The Blast Radius
7. A Handler That Tells Them Apart
import os
def raising(n):
if n == 2:
raise ValueError("a real job failure")
return n
def killing(n):
if n == 5:
os._exit(1)
return nfrom concurrent.futures import ProcessPoolExecutor, as_completed
from concurrent.futures.process import BrokenProcessPool
from w4 import raising, killing
def run_pool(fn, jobs, workers):
"""Distinguish a failed job from a broken pool. Every job gets an outcome."""
outcome = {}
try:
with ProcessPoolExecutor(max_workers=workers) as ex:
futs = {ex.submit(fn, n): n for n in jobs}
for fut in as_completed(futs):
owner = futs[fut]
try:
outcome[owner] = ("ok", fut.result())
except BrokenProcessPool:
outcome[owner] = ("not_run", "pool broke before this job finished")
except Exception as exc:
outcome[owner] = ("failed", type(exc).__name__)
except BrokenProcessPool:
pass
for n in jobs:
outcome.setdefault(n, ("not_run", "never resolved"))
return outcome
if __name__ == "__main__":
JOBS = list(range(8))
a = run_pool(raising, JOBS, workers=2)
print("a job that RAISES:")
print(" every job has an outcome:", len(a) == len(JOBS))
print(" job 2 is :", a[2])
print(" kinds present :", sorted({k for k, _ in a.values()}))
print()
b = run_pool(killing, JOBS, workers=2)
print("a worker that is KILLED:")
print(" every job has an outcome:", len(b) == len(JOBS))
print(" job 5 is :", b[5][0])
print(" kinds present :", sorted({k for k, _ in b.values()}))
print(" any job marked failed :", any(k == "failed" for k, _ in b.values()))a job that RAISES:
every job has an outcome: True
job 2 is : ('failed', 'ValueError')
kinds present : ['failed', 'ok']
a worker that is KILLED:
every job has an outcome: True
job 5 is : not_run
kinds present : ['not_run', 'ok']
any job marked failed : FalseTwo pools, because one would not have shown it. An earlier draft of this example put the raising job and the killing job in the same pool, and the kill broke the pool before job 2's exception could be retrieved — so job 2 came back as not_run and the discrimination the example exists to demonstrate was invisible. That is the hazard being described, arriving inside the demonstration of it, and the honest fix is to exercise each path in its own pool.
Four decisions, and the first two are what keep this honest under Section 1's rule.
Only the kinds present are printed, never their counts. How many jobs completed before the pool broke varies per run; which kinds appear does not.
BrokenProcessPool is caught before Exception. Python tries except clauses in order, so the specific one must come first — and the distinction is the entire point: a broken pool means not_run for jobs that were simply caught in the blast, and failed means this job did something.
The outer try catches the pool breaking on with-exit, which happens when the breakage is discovered during shutdown rather than while iterating.
setdefault guarantees every job has an outcome, whatever path the code took. Chapter 26.1's invariant is a property the aggregator checks; this is where it is actually earned.
Job 5 — the one that was killed — is reported as not_run rather than failed, and no job in that pool is marked failed at all, which is correct and slightly unsatisfying. Nothing in the pool can attribute a kill to its own job, so the only honest statement is that it did not produce a result. Chapter 26.8's resume will re-run it, and with a smaller worker count it will probably succeed, which is itself the diagnosis.
8. What to Check After Parallelising
CHECKS = [
("results identical to the serial baseline", "Chapter 27.2", "set comparison"),
("the invariant still balances", "Chapter 26.1", "total == sum of parts"),
("no new infrastructure bucket", "Chapter 26.7", "bucket diff"),
("no job appears twice", "Chapter 26.2 §6", "len(set(jobs)) == len(jobs)"),
("every run directory is distinct", "Chapter 26.4 §7", "allocation up front"),
("no BrokenProcessPool in the log", "this chapter", "a worker was killed"),
]
print(f"{'check':<42} {'from':<18} how")
for check, where, how in CHECKS:
print(f"{check:<42} {where:<18} {how}")
print()
print("checks:", len(CHECKS))
print("all of them existed before this module:", sum(1 for _, w, _ in CHECKS if w != "this chapter"))
print("which is the point:", "parallelising adds one check, not a new methodology")check from how
results identical to the serial baseline Chapter 27.2 set comparison
the invariant still balances Chapter 26.1 total == sum of parts
no new infrastructure bucket Chapter 26.7 bucket diff
no job appears twice Chapter 26.2 §6 len(set(jobs)) == len(jobs)
every run directory is distinct Chapter 26.4 §7 allocation up front
no BrokenProcessPool in the log this chapter a worker was killed
checks: 6
all of them existed before this module: 5
which is the point: parallelising adds one check, not a new methodologyFive of six checks are things Module 26 already built. That is the payoff for having built them: switching on concurrency is a configuration change validated by machinery that already exists, rather than a new project.
The one new check is for BrokenProcessPool anywhere in the run, and it deserves to be an error rather than a warning. A single occurrence means a worker was killed, which means the worker count is too high for the memory available, which means some number of jobs in that run were reclassified for no reason. A regression containing one is not a regression whose results should be compared against anything.
9. Predict, Run, Observe
import os
N = 0
def inc(x):
global N
N += 1
return N
def raiser(x):
if x == 1:
raise ValueError("no")
return x
def killer(x):
if x == 1:
os._exit(1)
return xfrom concurrent.futures import ProcessPoolExecutor, as_completed
from concurrent.futures.process import BrokenProcessPool
import w5
from w5 import inc, raiser, killer
if __name__ == "__main__":
J = [0, 1, 2, 3]
# A
with ProcessPoolExecutor(max_workers=2) as ex:
list(ex.map(inc, J))
print("A:", w5.N)
# B
ok, bad = [], []
with ProcessPoolExecutor(max_workers=2) as ex:
f2j = {ex.submit(raiser, n): n for n in J}
for f in as_completed(f2j):
try:
ok.append(f.result())
except Exception as e:
bad.append((f2j[f], type(e).__name__))
print("B:", sorted(ok), bad)
# C
print("C:", len(ok) + len(bad) == len(J))
# D
types = set()
seen = 0
try:
with ProcessPoolExecutor(max_workers=2) as ex:
f2j = {ex.submit(killer, n): n for n in J}
for f in as_completed(f2j):
seen += 1
try:
f.result()
except Exception as e:
types.add(type(e).__name__)
except BrokenProcessPool:
types.add("BrokenProcessPool")
print("D:", sorted(types))
# E
print("E:", seen == len(J))
# F
print("F:", issubclass(BrokenProcessPool, Exception))
# G
print("G:", BrokenProcessPool.__name__)
# H
print("H:", w5.N)The verified output:
A: 0
B: [0, 2, 3] [(1, 'ValueError')]
C: True
D: ['BrokenProcessPool']
E: True
F: True
G: BrokenProcessPool
H: 0A — eight increments in the workers, zero in the parent.
B — a raised exception is attributed to its own job and the other three results survive.
C — nothing is lost when the failure is a raise.
D — a killed worker produces only BrokenProcessPool, never the cause.
E — every future is still resolved, one way or another.
F — BrokenProcessPool is an Exception, which is why a bare except Exception silently absorbs it.
G — and this is the name to grep your logs for.
H — the parent's global is still zero, after two more pools.
10. Exercises
Exercise 1 — repeat it. Run one of your pool experiments three times and list which observations changed.
Exercise 2 — assert the invariant. Rewrite a test that asserts a schedule-dependent value so that it asserts a total instead.
Exercise 3 — list or dict. Show a list-of-futures loop recording a failure with no owner.
Exercise 4 — four characters. Say what {ex.submit(f, j): j for j in jobs} buys over the list form.
Exercise 5 — the two mysteries. Describe what Chapter 26.6 reports when a failure has no owner.
Exercise 6 — the parent's zero. Show a worker incrementing a global and the parent seeing nothing.
Exercise 7 — four patterns. Check your flow for each and say which you rely on.
Exercise 8 — none fail loudly. Say why this is worse than an exception would be.
Exercise 9 — the per-worker cache. Say what Chapter 25.3's compile cache costs in a pool, and why it is not a correctness problem.
Exercise 10 — kill a worker. Use os._exit in one job and report the total accounted for and the distinct error types.
Exercise 11 — no exception for its own job. Explain why the killed job produces nothing attributable.
Exercise 12 — the blast radius. Say how many jobs a raised exception affects and how many a kill affects.
Exercise 13 — the dangerous catch. Explain what except Exception does to the healthy outstanding jobs.
Exercise 14 — the usual cause. Name it, and say which chapter fixes it.
Exercise 15 — order the clauses. Show what happens if except Exception precedes except BrokenProcessPool.
Exercise 16 — the outer try. Say when the pool break is discovered on with-exit rather than while iterating.
Exercise 17 — setdefault. Say what it guarantees and which invariant depends on it.
Exercise 18 — latent, not new. For each of the four collisions, say what a serial run of the broken version does.
Exercise 19 — where the bug is. State the debugging rule for a bug that appears when parallelising.
Exercise 20 — six checks. Run all six after your first parallel run and report which, if any, failed.
11. Summary
- Most observable values in a parallel run are properties of the scheduler — how many jobs a worker handled, which finished first, how many succeeded before a failure.
- Publish and assert totals, type sets and the parent's own view, never a schedule.
- The annotation records the exception without the job, because it keeps futures in a list; a dict from future to job is four characters and the difference between a diagnosable failure and two mysteries.
- Workers do not share memory. A worker's
globalmodifies its own copy and the parent sees nothing. - Four serial patterns break silently — a shared results list, a counter, a cache, a lazily opened file — and none of them raises.
- A worker returns a value and writes only in its own directory; anything the parent must know is in the return value.
- A killed worker does not raise for its own job. There is no exception, because the process is gone.
- It raises
BrokenProcessPoolfor every outstanding future, so one job's memory footprint can reclassify dozens of healthy jobs. except Exceptionabsorbs it, recording all of them as infrastructure failures — Chapter 26.1's pessimistic direction.- Catch
BrokenProcessPoolfirst, mark the caught-in-the-blast jobsnot_runrather thanERROR, and stop submitting. - The usual cause is the OOM killer, which makes the fix a smaller worker count rather than better handling.
- Concurrency introduced none of the four collision hazards and activated all four. A serial run of the broken version passes every time.
- When parallelising surfaces a bug, the bug is almost never in the parallel code.
- A killed job is honestly
not_run, because nothing in the pool can attribute a kill to its owner. - Five of the six post-parallelisation checks already existed — switching on concurrency is validated by Module 26's machinery.
- Treat any
BrokenProcessPoolas an error, not a warning; a run containing one should not be compared against anything.
The last chapter of this module chooses the worker count that stops most of this happening.
Where this fits
Part of the Python curriculum.