Concurrency with Threads and Processes
concurrent.futures as the common interface, processes and the pickling constraint, sharing state safely, and the deadlocks that come from getting either wrong.
concurrent.futures as the common interface, processes and the pickling constraint, sharing state safely, and the deadlocks that come from getting either wrong.
The parallel version of your image pipeline works on your laptop and hangs on the build machine. No error, no output, no CPU usage — the pool is waiting for workers, and the workers are waiting to send back results that will not fit in the pipe until someone reads them. Nobody reads them, because the main thread is waiting for the workers.
Concurrency is where correct-looking code fails in ways that depend on timing and machine. By the end of this lesson you will use one interface for both threads and processes, know what crossing a process boundary costs, share state without corrupting it, and recognise the deadlocks before you write them.
concurrent.futures gives threads and processes the same API,
which means switching is one word once you have measured which
you need:
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
with ThreadPoolExecutor(max_workers=16) as pool: # waiting
results = list(pool.map(fetch, urls))
with ProcessPoolExecutor(max_workers=8) as pool: # computing
digests = list(pool.map(checksum, paths))map keeps input order and re-raises the first exception when
you reach that result — which means an exception in item three
surfaces when you iterate to item three, not when it happens.
For results as they finish, and for handling failures per item:
from concurrent.futures import as_completed
with ThreadPoolExecutor(max_workers=16) as pool:
futures = {pool.submit(fetch, url): url for url in urls}
for future in as_completed(futures):
url = futures[future]
try:
results.append(future.result())
except httpx.HTTPError as error:
log.warning("failed %s: %s", url, error)The dictionary mapping future back to input is the standard
idiom, because a Future on its own does not remember what
produced it.
A future's exception is stored, not raised. It surfaces when
you call .result(). Submitting work and never collecting the
results means failures vanish entirely, which is one of the
quietest bugs available.
Start around four times the core count.
Then tune by measuring. The real limit is almost never your machine — it is the remote service's connection limit or its rate limiter.
os.cpu_count(), or one less.
There is no benefit past the number of cores, because there is nothing left to run on.
More threads is not better. Each costs a stack, each adds scheduling, and past a point you are only queueing more concurrent requests at a service that will start rate-limiting you. If the far end has a connection limit, that number is your ceiling regardless of what your machine could manage.
Threads share memory. Processes do not.
Pickle
The object is serialised in the parent. Anything unpicklable fails here.
Copy through a pipe
Bytes on the wire. This is the cost people forget.
Unpickle
The worker rebuilds it, and now two copies exist.
Then the whole thing runs again, backwards, for the result.
with ProcessPoolExecutor() as pool:
result = pool.submit(process, huge_dataframe).result()For a small computation on a large object, that round trip costs more than the work — and the parallel version is slower than the serial one it replaced.
Two consequences to design around.
Send small, work big. Pass a path and let the worker read the file; do not pass the file's contents.
Not everything can be pickled. Lambdas, local functions, open files, sockets, database connections and most objects holding a lock all fail:
pool.map(lambda p: resize(p, 800), paths)
# PicklingError: Can't pickle <function <lambda>>The fix is a module-level function, with functools.partial for
the extra arguments:
from functools import partial
def resize_one(path: Path, width: int) -> Path: ...
pool.map(partial(resize_one, width=800), paths)Bad — several threads updating a plain dictionary.
results: dict[str, int] = {}
def process(path: Path) -> None:
size = expensive_measure(path)
results[path.name] = size
results["total"] = results.get("total", 0) + sizeGood — return values and combine them in one place.
def process(path: Path) -> tuple[str, int]:
return path.name, expensive_measure(path)
with ThreadPoolExecutor(max_workers=16) as pool:
results = dict(pool.map(process, paths))
total = sum(results.values())The individual assignment happens to be atomic; the
read-add-write on "total" is not, so under load the total is
quietly low — no exception, a plausible number, and a bug that
does not reproduce at small scale. It is the counter += 1 race
from the previous lesson, in the shape people actually write it.
The good version has no shared mutable state at all. Workers return values, one thread combines them, and there is nothing to synchronise. Prefer returning results to mutating shared state — it is faster than locking and it removes the class of bug rather than defending against it.
When you genuinely must share:
import threading
lock = threading.Lock()
with lock:
shared_total += size # one place, held brieflyfrom queue import Queue
queue: Queue[Path] = Queue() # designed for this; no lock neededqueue.Queue is the tool for producer–consumer work between
threads. Across processes, multiprocessing.Queue,
Value and Array exist, and a Manager offers shared
dictionaries and lists at the cost of a proxy round trip per
operation — which is frequently slower than passing results
back.
Two classic shapes, and the opening scenario is the second.
Lock ordering. Two threads take two locks in opposite orders, each holding one and waiting for the other:
# thread A: with lock_a: with lock_b: ...
# thread B: with lock_b: with lock_a: ...The fix is a rule: always acquire locks in the same order, everywhere, and write that order down. Or use a timeout so a deadlock becomes a loud failure rather than a silent hang:
if not lock.acquire(timeout=5):
raise TimeoutError("could not acquire the index lock")Pipe buffers. A worker writing a large result blocks until
someone reads it; the parent will read only after the work
finishes. Both wait. Using the executor's own map and
as_completed avoids this, because they drain results
continuously — hand-rolled multiprocessing.Process with a
Queue you read after join() is where people meet it.
A third worth naming: nesting pools. Work submitted to a pool that itself submits to the same pool can occupy every worker with tasks waiting on tasks that have nowhere to run. Keep pools flat.
Shutdown is where concurrent programs leak.
with on an executor calls shutdown(wait=True), which blocks
until every submitted task finishes. For cancelling queued work
on Ctrl-C:
pool.shutdown(wait=False, cancel_futures=True)That cancels tasks not yet started; anything already running runs to completion, because there is no safe way to interrupt a thread mid-work. This is why long tasks should check a flag:
stop = threading.Event()
def process(path: Path) -> None:
for chunk in read_chunks(path):
if stop.is_set():
return
handle(chunk)threading.Event is the right tool for "please stop" — it is
thread-safe and readable from anywhere. And non-daemon threads
keep the interpreter alive at exit, so a forgotten background
thread is a program that will not quit.
ONE INTERFACE
ThreadPoolExecutor waiting (network, disk, subprocess)
ProcessPoolExecutor computing
same API - switching is one word
pool.map(f, items) ordered; raises when you reach it
pool.submit(f, x) -> Future
as_completed(futures) results as they finish
{pool.submit(f, x): x ...} map a future back to its input
a Future STORES its exception - never collecting results
means failures vanish
WORKERS
threads ~4x cores to start; the remote service is the real limit
processes os.cpu_count()
PROCESS BOUNDARY
everything is pickled, copied, unpickled
send a PATH, not the file contents
cannot pickle: lambdas, local functions, files, sockets,
connections, most lock-holding objects
module-level function + functools.partial
if __name__ == "__main__": is REQUIRED, or workers re-import
your module and spawn more workers
SHARING STATE
prefer RETURNING results and combining in one thread
x += 1 is read-add-write: a race, whatever the GIL does
threading.Lock for a real shared counter
queue.Queue for producer/consumer - no lock needed
Manager dict/list: works, and pays a round trip per operation
DEADLOCK
same lock order everywhere, written down
lock.acquire(timeout=5) so a hang becomes an error
pipe buffers: use map/as_completed, which drain continuously
never nest pools
SHUTDOWN
`with` -> shutdown(wait=True)
shutdown(wait=False, cancel_futures=True) on Ctrl-C
running tasks cannot be interrupted - poll a threading.Event
a non-daemon thread keeps the process aliveYou can now run work concurrently with one interface, choose the executor from the diagnosis rather than from habit, keep state out of the shared space where most concurrency bugs live, and put timeouts where a hang would otherwise be invisible.
Next is Asyncio from the Event Loop Up, the third model. Threads let the operating system decide when to switch; asyncio makes the switches explicit and visible in the source, which scales to thousands of concurrent operations and introduces its own single spectacular failure — one blocking call that stops everything.
Before you move on, take the CPU-bound and I/O-bound loops from the previous lesson and run each through both executors — four combinations, timed. Two of them will be faster, one will be about the same, and one will be slower. Producing that table yourself replaces a rule you have to remember with a result you have seen.