Concurrency all 3 written

Threads and what the GIL actually blocks, real parallelism with processes, and async for waiting on thousands of things at once.

41 Threads, and what the GIL actually stops

Threads let one program wait for several things at once. In Python they are excellent at that and nearly useless for arithmetic, and the reason is a single lock inside the interpreter. Rather than explain it, this page times it.

A thread, at its simplest

A Thread is a callable plus its arguments. start() runs it alongside you; join() waits for it to finish. Nothing more.

start, join, and the one-shot rule

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27import threading, time

def greet(name, pause):
    time.sleep(pause)
    print(f"  {name} finished after {pause}s")

worker = threading.Thread(target=greet, args=("worker", 0.05))

print("before start, alive:", worker.is_alive())
worker.start()
print("after start, alive: ", worker.is_alive())
worker.join()                      # block until it is done
print("after join, alive:  ", worker.is_alive())

# A Thread object is single-use. Starting it again is an error, not a restart.
try:
    worker.start()
except RuntimeError as exc:
    print("restarting it:", exc)

print()
print("the thread you are already on:", threading.current_thread().name)
print("threads are non-daemon by default:", threading.Thread(target=greet).daemon)

test("a fresh thread is not alive yet", threading.Thread(target=greet).is_alive(), False)
test("join() means it has really finished", worker.is_alive(), False)
test("the main thread is called MainThread", threading.current_thread().name, "MainThread")

What it printed, when the build ran it

before start, alive: False
after start, alive:  True
  worker finished after 0.05s
after join, alive:   False
restarting it: threads can only be started once

the thread you are already on: MainThread
threads are non-daemon by default: False
ok   a fresh thread is not alive yet
ok   join() means it has really finished
ok   the main thread is called MainThread

join() is not optional. Without it your program can reach the end of main while threads are still working. Non-daemon threads keep the process alive until they finish, so you get the right answer late and in the wrong order; daemon threads are killed mid-sentence instead. Either way you have lost control of when your results exist.

Work that waits: threads earn their keep

Three requests, each taking a tenth of a second, where nearly all of that time is spent waiting for someone else. Run one after another, that is three tenths. Run them together, it is one.

Three waits at once

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31import threading, time

def fetch(url, out, index):
    time.sleep(0.1)                # stands in for a network round trip
    out[index] = len(url)

urls = ["https://a.example", "https://bb.example", "https://ccc.example"]

# Serially.
start = time.perf_counter()
serial_results = [0] * len(urls)
for i, url in enumerate(urls):
    fetch(url, serial_results, i)
serial = time.perf_counter() - start

# Across threads.
start = time.perf_counter()
threaded_results = [0] * len(urls)
threads = [threading.Thread(target=fetch, args=(u, threaded_results, i))
           for i, u in enumerate(urls)]
for t in threads: t.start()
for t in threads: t.join()
threaded = time.perf_counter() - start

print(f"serial:   {serial:.2f}s")
print(f"threaded: {threaded:.2f}s")
print(f"speedup:  {serial / threaded:.1f}x")

test("both orderings compute the same thing", threaded_results, serial_results)
test("three 0.1s waits, done serially, take about 0.3s", 0.28 < serial < 0.45, True)
test("done together, they take about one of them", threaded < serial / 2, True)

What it printed, when the build ran it

serial:   0.31s
threaded: 0.10s
speedup:  3.0x
ok   both orderings compute the same thing
ok   three 0.1s waits, done serially, take about 0.3s
ok   done together, they take about one of them

That is the entire case for threads, and it is a strong one: whenever the time is spent waiting — network, disk, a database, a subprocess — a thread that is waiting costs almost nothing, and you can wait on hundreds at once.

Work that computes: threads do nothing

Now the same shape of experiment with arithmetic instead of waiting. Two threads, two cores available, half the work each.

The same experiment, with computation

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25import threading, time

def burn(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

N = 3_000_000

start = time.perf_counter()
burn(N)
serial = time.perf_counter() - start

start = time.perf_counter()
halves = [threading.Thread(target=burn, args=(N // 2,)) for _ in range(2)]
for t in halves: t.start()
for t in halves: t.join()
threaded = time.perf_counter() - start

print(f"serial:    {serial:.2f}s")
print(f"2 threads: {threaded:.2f}s")
print(f"speedup:   {serial / threaded:.2f}x    (two cores, half the work each: 2.0x was the hope)")

test("two threads got nowhere near twice the throughput", serial / threaded < 1.6, True)

What it printed, when the build ran it

serial:    0.26s
2 threads: 0.21s
speedup:   1.22x    (two cores, half the work each: 2.0x was the hope)
ok   two threads got nowhere near twice the throughput

Nothing was wrong with that code. The reason is the global interpreter lock: a single lock that must be held to execute Python bytecode, so only one thread is ever running Python at a time. Threads take turns. On work that waits, a thread releases the lock while it waits and the turns overlap usefully; on work that computes, the turns are the whole story and there is nothing to overlap.

Asking this interpreter, rather than assuming

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17import sys, sysconfig

# Since 3.13 the GIL is a build option, so this is worth asking rather than assuming.
print("Python:", ".".join(map(str, sys.version_info[:3])))
print("Py_GIL_DISABLED in this build's config:", sysconfig.get_config_var("Py_GIL_DISABLED"))
print("sys._is_gil_enabled():", sys._is_gil_enabled())
print()
if sys._is_gil_enabled():
    print("This page's timings were produced with the GIL on.")
else:
    print("Free-threaded build: the CPU-bound timing above would look different.")

gil_on = sys._is_gil_enabled()
free_threaded = sysconfig.get_config_var("Py_GIL_DISABLED") == 1

test("the two answers agree", gil_on, not free_threaded)
test("the interpreter that built this page had the GIL enabled", gil_on, True)

What it printed, when the build ran it

Python: 3.14.7
Py_GIL_DISABLED in this build's config: 0
sys._is_gil_enabled(): True

This page's timings were produced with the GIL on.
ok   the two answers agree
ok   the interpreter that built this page had the GIL enabled

The GIL is being removed, slowly. Python 3.13 added a free-threaded build (PEP 703) where the lock genuinely does not exist and CPU-bound threads scale. It is an opt-in build, not the default download, and extension modules have to be adapted for it. Until the day it is the default, assume the timings above.

The GIL does not make your code safe

This is the expensive misunderstanding. The GIL protects the interpreter's internal state. It says nothing about yours. Any update you make in more than one step can be interrupted between the steps — and "read it, compute, write it back" is three steps.

A lost update, reproduced

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22import threading, time

counter = 0

def bump(times):
    global counter
    for _ in range(times):
        current = counter        # read
        time.sleep(0)            # yield to another thread, mid-update
        counter = current + 1    # write back what is now stale

threads = [threading.Thread(target=bump, args=(2000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()

expected = 4 * 2000
print("expected:", expected)
print("actual:  ", counter)
print("lost:    ", expected - counter)

test("four threads each counted to 2000", expected, 8000)
test("and most of those increments vanished", counter < expected, True)

What it printed, when the build ran it

expected: 8000
actual:   2000
lost:     6000
ok   four threads each counted to 2000
ok   and most of those increments vanished

The fix is a lock of your own, around the whole update rather than any part of it. While one thread holds it, the others wait at the with.

The same code, with a lock

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22import threading, time

counter = 0
lock = threading.Lock()

def bump(times):
    global counter
    for _ in range(times):
        with lock:               # nobody else can be between these lines
            current = counter
            time.sleep(0)
            counter = current + 1

threads = [threading.Thread(target=bump, args=(2000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()

print("with a lock:", counter)
print("expected:   ", 4 * 2000)

test("every increment survived", counter, 8000)
test("the lock is released again afterwards", lock.locked(), False)

What it printed, when the build ran it

with a lock: 8000
expected:    8000
ok   every increment survived
ok   the lock is released again afterwards

Do not reason about which operations are "atomic". On the interpreter that built this page, four threads running counter += 1 a million times lost nothing at all — the race the textbooks demonstrate did not reproduce. That is not a guarantee; it is an accident of how this version compiles that statement and how often it switches threads. Code that is correct because of an implementation detail breaks on the next release. If two threads touch the same thing, lock it.

`queue.Queue`: the shape that avoids most locks

Most threaded work is "a pile of jobs, several workers". A Queue is already thread-safe, so handing work through it removes almost all of your own locking.

Producer, queue, three consumers

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38import queue, threading, time

jobs = queue.Queue()
results = queue.Queue()

def worker(name):
    while True:
        item = jobs.get()
        if item is None:              # the agreed stop signal
            jobs.task_done()
            return
        time.sleep(0.01)              # the work
        results.put((name, item * item))
        jobs.task_done()

workers = [threading.Thread(target=worker, args=(f"w{i}",)) for i in range(3)]
for t in workers: t.start()

for n in range(9):
    jobs.put(n)
for _ in workers:                     # one stop signal per worker
    jobs.put(None)

jobs.join()                           # wait until every task_done() has landed
for t in workers: t.join()

collected = []
while not results.empty():
    collected.append(results.get())

squares = sorted(value for _, value in collected)
print("jobs done:", len(collected))
print("squares:  ", squares)
print("workers that got some:", len({name for name, _ in collected}))

test("every job was handled exactly once", squares, [n * n for n in range(9)])
test("no job was handled twice", len(collected), 9)
test("the queue is empty at the end", jobs.empty(), True)

What it printed, when the build ran it

jobs done: 9
squares:   [0, 1, 4, 9, 16, 25, 36, 49, 64]
workers that got some: 3
ok   every job was handled exactly once
ok   no job was handled twice
ok   the queue is empty at the end

An exception in a thread does not stop your program

This one silently ruins results. When a thread raises, the traceback is printed and that thread ends. The main thread never hears about it and carries on as if the work was done.

A thread dies; the program does not notice

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20import threading

shared = {}

def load(key):
    if key == "bad":
        raise ValueError(f"cannot load {key!r}")
    shared[key] = key.upper()

threads = [threading.Thread(target=load, args=(k,)) for k in ("a", "bad", "c")]
for t in threads: t.start()
for t in threads: t.join()

print()
print("main thread reached the end normally.")
print("shared:", shared)
print("but 'bad' is missing, and nothing here failed.")

test("the surviving threads did their work", sorted(shared), ["a", "c"])
test("the failed one left no trace in the result", "bad" in shared, False)

What it printed, when the build ran it

main thread reached the end normally.
shared: {'a': 'A', 'c': 'C'}
but 'bad' is missing, and nothing here failed.
ok   the surviving threads did their work
ok   the failed one left no trace in the result
Exception in thread Thread-2 (load):
Traceback (most recent call last):
  File "<python>/lib/python3.14/threading.py", line 1082, in _bootstrap_inner
    self._context.run(self.run)
    ~~~~~~~~~~~~~~~~~^^^^^^^^^^
  File "<python>/lib/python3.14/threading.py", line 1024, in run
    self._target(*self._args, **self._kwargs)
    ~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "example.py", line 7, in load
    raise ValueError(f"cannot load {key!r}")
ValueError: cannot load 'bad'

The traceback above was printed by threading.excepthook, on stderr. Nothing checked it, and in a real program it scrolls past. ThreadPoolExecutor is the answer: it hands you a Future per job, and asking a failed future for its result re-raises the exception on your thread, where you can handle it.

ThreadPoolExecutor, where failures come back to you

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31from concurrent.futures import ThreadPoolExecutor
import time

def load(key):
    time.sleep(0.02)
    if key == "bad":
        raise ValueError(f"cannot load {key!r}")
    return key.upper()

loaded, failed = {}, {}

with ThreadPoolExecutor(max_workers=3) as pool:
    futures = {pool.submit(load, k): k for k in ("a", "bad", "c")}
    for future, key in futures.items():
        try:
            loaded[key] = future.result()      # re-raises here, on this thread
        except ValueError as exc:
            failed[key] = str(exc)

print("loaded:", loaded)
print("failed:", failed)

# pool.map is the shorter form -- but it raises on the first failure, so use it
# only when any failure should abort the batch.
with ThreadPoolExecutor(max_workers=3) as pool:
    fine = list(pool.map(load, ["x", "y", "z"]))
print("map, when nothing fails:", fine)

test("the successes came back", loaded, {"a": "A", "c": "C"})
test("and the failure was caught, not printed", failed, {"bad": "cannot load 'bad'"})
test("map returns results in the order you submitted", fine, ["X", "Y", "Z"])

What it printed, when the build ran it

loaded: {'a': 'A', 'c': 'C'}
failed: {'bad': "cannot load 'bad'"}
map, when nothing fails: ['X', 'Y', 'Z']
ok   the successes came back
ok   and the failure was caught, not printed
ok   map returns results in the order you submitted

Prefer the pool. ThreadPoolExecutor reuses a fixed number of threads rather than creating one per job, gives you results and exceptions as values, and shuts down when the with block ends. Reach for a bare Thread only when you need something long-lived that a pool does not model.

Where this leaves you

Threads are for waiting. If your program is slow because it is computing, threads cannot help while the GIL exists — that is the next lesson. If it is slow because it is waiting on thousands of things, threads work but get unwieldy, and async was added for exactly that shape.

To reproduce this yourself

shown, not run

1
2
3
4
5
6
7
8# One line of shell, to see the difference for yourself.
# Real work, timed three ways -- this is what the panels above measured.

python3 -X importtime -c "pass"        # what interpreter start-up costs
python3 -m timeit -s "import time" "time.sleep(0)"

# And on a free-threaded build, the CPU-bound panel above tells a different story:
python3.14t -c "import sys; print(sys._is_gil_enabled())"   # -> False

Shell commands, not Python: there is no terminal in this page to run them in. Every claim they would demonstrate is measured by the panels above, with the real interpreter.

Threads overlap waiting, not computing. That is measured above, both ways.
The GIL means one thread runs Python bytecode at a time. Ask sys._is_gil_enabled() rather than assuming — it is a build option now.
The GIL does not protect your invariants. Any read-modify-write needs a Lock.
Never rely on an operation being atomic. It is not a promise the language makes.
queue.Queue is thread-safe, and the job-queue shape removes most of your locking.
A thread that raises prints to stderr and dies quietly. Use ThreadPoolExecutor and check future.result() so failures reach you.

42 multiprocessing: real parallelism, real costs

One interpreter runs one thread of Python at a time. Two interpreters run two. That is the whole idea: if the work is arithmetic rather than waiting, give it to separate processes, each with its own interpreter and its own lock.

The experiment that failed with threads

Lesson 41 split arithmetic across two threads and got nothing for it. Here is the same split across four processes.

Four processes, four quarters of the work

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29from concurrent.futures import ProcessPoolExecutor
import os, time

def burn(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

# Everything with an effect goes inside this guard. The next panel shows why.
if __name__ == "__main__":
    N, workers = 12_000_000, 4

    start = time.perf_counter()
    whole = burn(N)
    serial = time.perf_counter() - start

    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=workers) as pool:
        parts = list(pool.map(burn, [N // workers] * workers))
    parallel = time.perf_counter() - start

    print(f"one process:    {serial:.2f}s")
    print(f"{workers} processes:    {parallel:.2f}s")
    print(f"speedup:        {serial / parallel:.2f}x   on {os.cpu_count()} cores")

    test("four processes really were faster", parallel < serial, True)
    test("and by more than measurement noise", serial / parallel > 1.5, True)
    test("each worker did a quarter of the range", len(parts), 4)

What it printed, when the build ran it

one process:    0.90s
4 processes:    0.56s
speedup:        1.59x   on 16 cores
ok   four processes really were faster
ok   and by more than measurement noise
ok   each worker did a quarter of the range

Not 4x, and it never is: the processes have to be started, the arguments sent to them and the results sent back. What you get is real parallelism minus a fixed cost, and the rest of this lesson is about that cost, because it is large enough to turn a speed-up into a slow-down.

The guard is not a style rule

A new process has to get your code somehow. On macOS and Windows the default is spawn: a fresh interpreter starts and imports your main module to find the function it was asked to run. Anything at the top level of that module therefore runs again, in every child.

What a spawned child sees

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29import multiprocessing as mp
import os

# Top level. Under spawn this line runs AGAIN in every child, with the child's pid.
STARTED_AS = os.getpid()

def who(parent_pid):
    return {
        "__name__": __name__,           # "__mp_main__" in a spawned child
        "different process": os.getpid() != parent_pid,
        "STARTED_AS still matches me": os.getpid() == STARTED_AS,
    }

if __name__ == "__main__":
    print("start method here:", mp.get_start_method())
    print("also available:   ", sorted(mp.get_all_start_methods()))
    print()
    with mp.Pool(2) as pool:
        seen = pool.map(who, [os.getpid()] * 4)
    for row in seen:
        print("  ", row)

    test("the child's module is not called __main__",
         {row["__name__"] for row in seen}, {"__mp_main__"})
    test("every task really ran elsewhere",
         all(row["different process"] for row in seen), True)
    test("but the top-level constant was recomputed there, so it agrees with the child",
         all(row["STARTED_AS still matches me"] for row in seen), True)
    test("this process is still __main__", __name__, "__main__")

What it printed, when the build ran it

start method here: spawn
also available:    ['fork', 'forkserver', 'spawn']

   {'__name__': '__mp_main__', 'different process': True, 'STARTED_AS still matches me': True}
   {'__name__': '__mp_main__', 'different process': True, 'STARTED_AS still matches me': True}
   {'__name__': '__mp_main__', 'different process': True, 'STARTED_AS still matches me': True}
   {'__name__': '__mp_main__', 'different process': True, 'STARTED_AS still matches me': True}
ok   the child's module is not called __main__
ok   every task really ran elsewhere
ok   but the top-level constant was recomputed there, so it agrees with the child
ok   this process is still __main__

Look at the last line of each row. STARTED_AS was set at the top of the module, once, in this process — and in every child it holds the child's pid, because the child ran that line too. That is the whole hazard of spawn in one value.

So without the guard, the child re-imports your module, reaches the line that creates a pool, and tries to start more processes while it is still being imported. Python catches that and raises — and then the pool replaces the worker that just died, forever. It does not crash; it never stops.

The missing guard, captured from a real run

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34import subprocess, sys, textwrap
from pathlib import Path

# A script that forgets the guard.
Path("no_guard.py").write_text(textwrap.dedent('''
    import multiprocessing as mp

    def double(n):
        return n * 2

    print(mp.Pool(2).map(double, [1, 2, 3]))
''').lstrip(), encoding="utf-8")

if __name__ == "__main__":
    try:
        subprocess.run([sys.executable, "no_guard.py"], capture_output=True,
                       text=True, timeout=4, check=True)
        finished = True
    except subprocess.TimeoutExpired as exc:
        finished = False
        complaint = (exc.stderr or "")
        if isinstance(complaint, bytes):
            complaint = complaint.decode()

    print("did it finish?", finished)
    print()
    print("what it said, once (it said this over and over):")
    at = complaint.index("RuntimeError:")
    for line in complaint[at:].splitlines()[:9]:
        print("  ", line)

    test("the broken script never terminated", finished, False)
    test("and the reason names the idiom it is missing",
         "if __name__ == '__main__'" in complaint, True)

What it printed, when the build ran it

did it finish? False

what it said, once (it said this over and over):
   RuntimeError: 
           An attempt has been made to start a new process before the
           current process has finished its bootstrapping phase.
   
           This probably means that you are not using fork to start your
           child processes and you have forgotten to use the proper idiom
           in the main module:
   
               if __name__ == '__main__':
ok   the broken script never terminated
ok   and the reason names the idiom it is missing

Put everything under the guard, not just the pool. Module-level printing, file writing, argument parsing and database connections all run again in every child under spawn. The rule that makes this painless: at module level define things; do things inside if __name__ == "__main__": or inside a function it calls.

Nothing is shared

Threads share memory, which is why lesson 41 needed locks. Processes do not share anything, which removes that problem and replaces it with a different one: a child cannot see your changes, and you cannot see its.

Separate memory, in both directions

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19import multiprocessing as mp

cache = {}          # the parent's

def remember(n):
    cache[n] = n * n              # whichever child runs this has its OWN cache
    return len(cache)

if __name__ == "__main__":
    with mp.Pool(2) as pool:
        sizes = pool.map(remember, [1, 2, 3, 4])

    print("each result is the size of that child's cache:", sizes)
    print("the parent's cache after all of that:", cache)
    print()
    print("no lock was needed, and no result came back except the return value.")

    test("the parent's dict was never touched", cache, {})
    test("the children did fill their own", max(sizes) >= 1, True)

What it printed, when the build ran it

each result is the size of that child's cache: [1, 2, 3, 4]
the parent's cache after all of that: {}

no lock was needed, and no result came back except the return value.
ok   the parent's dict was never touched
ok   the children did fill their own

Everything crossing the boundary must pickle

Arguments go to the child and results come back as bytes, using pickle. Anything pickle cannot represent cannot make the trip — and the two commonest culprits are a lambda and an open file.

What can and cannot cross

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22import pickle
from functools import partial

def scale(factor, n):        # module level, so it has a name to pickle by
    return n * factor

triple = partial(scale, 3)   # partials pickle, as long as the function does

print("a named function:", pickle.loads(pickle.dumps(scale))(2, 5))
print("a partial:       ", pickle.loads(pickle.dumps(triple))(5))

for label, value in [("a lambda", lambda n: n),
                     ("a nested function", (lambda: (lambda: 1))()),
                     ("an open file", open("probe.txt", "w"))]:
    try:
        pickle.dumps(value)
        print(f"{label}: pickled")
    except Exception as exc:
        print(f"{label}: {type(exc).__name__}")

test("a module-level function survives the round trip", pickle.loads(pickle.dumps(scale))(3, 4), 12)
test("so does a partial of one", pickle.loads(pickle.dumps(triple))(7), 21)

What it printed, when the build ran it

a named function: 10
a partial:        15
a lambda: PicklingError
a nested function: PicklingError
an open file: TypeError
ok   a module-level function survives the round trip
ok   so does a partial of one

Which is why passing a lambda to a pool fails, and why the fix is a module-level function or a partial of one.

The same job, failing and then working

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24import multiprocessing as mp
from functools import partial

def scale(factor, n):
    return n * factor

if __name__ == "__main__":
    # The failure, caught rather than allowed to print a wall of traceback.
    try:
        with mp.Pool(2) as pool:
            pool.map(lambda n: n * 3, [1, 2, 3])
        failure = "none"
    except Exception as exc:
        failure = type(exc).__name__

    print("passing a lambda:", failure)

    # The fix.
    with mp.Pool(2) as pool:
        tripled = pool.map(partial(scale, 3), [1, 2, 3])
    print("passing a partial:", tripled)

    test("a lambda cannot be sent to a worker", failure, "PicklingError")
    test("a partial of a named function can", tripled, [3, 6, 9])

What it printed, when the build ran it

passing a lambda: PicklingError
passing a partial: [3, 6, 9]
ok   a lambda cannot be sent to a worker
ok   a partial of a named function can

The cost, measured

This is the number that decides whether processes are the right tool. Squaring two thousand integers is not enough work to be worth sending anywhere.

When processes are the wrong answer

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24from concurrent.futures import ProcessPoolExecutor
import time

def square(n):
    return n * n

if __name__ == "__main__":
    data = list(range(2000))

    start = time.perf_counter()
    serial = [square(n) for n in data]
    serial_ms = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=2) as pool:
        parallel = list(pool.map(square, data))
    parallel_ms = (time.perf_counter() - start) * 1000

    print(f"a comprehension:  {serial_ms:.2f} ms")
    print(f"two processes:    {parallel_ms:.2f} ms")
    print(f"the pool was {parallel_ms / serial_ms:.0f}x SLOWER")

    test("both got the right answer", parallel, serial)
    test("and the pool was far slower", parallel_ms > serial_ms * 10, True)

What it printed, when the build ran it

a comprehension:  0.12 ms
two processes:    441.02 ms
the pool was 3806x SLOWER
ok   both got the right answer
ok   and the pool was far slower

The reason is per-item overhead: map sends each item separately by default, so two thousand items means two thousand round trips. chunksize batches them, and it is usually the difference between unusable and fine.

chunksize, which is not a micro-optimisation

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25from concurrent.futures import ProcessPoolExecutor
import time

def square(n):
    return n * n

if __name__ == "__main__":
    data = list(range(20_000))

    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=2) as pool:
        one_at_a_time = list(pool.map(square, data))
    per_item = time.perf_counter() - start

    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=2) as pool:
        batched = list(pool.map(square, data, chunksize=2000))
    chunked = time.perf_counter() - start

    print(f"chunksize=1 (default):  {per_item:.2f}s")
    print(f"chunksize=2000:         {chunked:.2f}s")
    print(f"the only change:        how many messages were sent")

    test("the answers are identical", batched, one_at_a_time)
    test("batching was dramatically faster", chunked < per_item / 5, True)

What it printed, when the build ran it

chunksize=1 (default):  3.62s
chunksize=2000:         0.14s
the only change:        how many messages were sent
ok   the answers are identical
ok   batching was dramatically faster

Sharing state anyway, when you must

A Manager puts the shared object in its own server process and hands the children proxies. It works, and every read and write is a round trip — so use it for coordination, never in a loop.

Manager, for coordination rather than throughput

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20import multiprocessing as mp

def bump(args):
    counter, lock = args
    with lock:                       # the same race as with threads, same fix
        counter.value += 1
    return counter.value

if __name__ == "__main__":
    with mp.Manager() as manager:
        counter = manager.Value("i", 0)
        lock = manager.Lock()
        with mp.Pool(2) as pool:
            pool.map(bump, [(counter, lock)] * 8)
        final = counter.value

    print("eight increments across two processes:", final)
    print("the proxy object was:", "a stand-in for a value living in a third process")

    test("nothing was lost", final, 8)

What it printed, when the build ran it

eight increments across two processes: 8
the proxy object was: a stand-in for a value living in a third process
ok   nothing was lost

The better answer is usually no shared state. Give each worker its own slice, return a result, and combine in the parent. That is what pool.map already does, it needs no locks, and it is the reason the first panel on this page is four lines long.

Choosing between the three

Your program is slow because…	Use	Why
it waits on a few dozen things (files, requests, a database)	`ThreadPoolExecutor`	Waiting releases the GIL, and threads need no change to the code being called.
it waits on thousands of things at once	`async`	No per-task thread stack. Next lesson.
it computes, and the chunks are big	`ProcessPoolExecutor`	Separate interpreters, separate locks, real cores. Measured above.
it computes, and the chunks are tiny	neither	The overhead is thousands of times the work. Make the chunks bigger or leave it serial.

Processes give real parallelism because each has its own interpreter and its own GIL.
if __name__ == "__main__": is required under spawn. Without it the program does not crash — it spawns forever, as captured above.
Nothing is shared. Arguments and results are pickled, so lambdas and open files cannot cross.
The fixed cost is large. Measure before believing a pool helped; for small items set chunksize.
Prefer returning results over sharing state. Manager works but every access is a message.

43 async and await, and the loop underneath

Threads let the operating system interleave waiting. async lets you do it, in one thread, at the points you mark with await — which is cheaper, and the reason one program can wait on thousands of things at once.

An async function does not run when you call it

This is the first surprise and the source of most async bugs. Calling a coroutine function builds a coroutine object. Something has to drive it.

Call, then drive

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25import asyncio, inspect

async def fetch(name):
    await asyncio.sleep(0.01)
    return f"{name} loaded"

# Calling it does not run it.
pending = fetch("config")
kind = type(pending).__name__
before = inspect.getcoroutinestate(pending)
print("calling it returned:", kind)
print("is it a coroutine?  ", inspect.iscoroutine(pending))
print("what state is it in?", before)

# asyncio.run() starts an event loop, drives the coroutine to completion, and closes.
returned = asyncio.run(pending)
print()
print("after asyncio.run:  ", inspect.getcoroutinestate(pending))
print("it returned:        ", returned)

test("an async def is still a plain function object", inspect.isfunction(fetch), True)
test("but calling it yields a coroutine", kind, "coroutine")
test("which had not started running", before, "CORO_CREATED")
test("asyncio.run drove it and gave back the value", returned, "config loaded")
test("and left it finished", inspect.getcoroutinestate(pending), "CORO_CLOSED")

What it printed, when the build ran it

calling it returned: coroutine
is it a coroutine?   True
what state is it in? CORO_CREATED

after asyncio.run:   CORO_CLOSED
it returned:         config loaded
ok   an async def is still a plain function object
ok   but calling it yields a coroutine
ok   which had not started running
ok   asyncio.run drove it and gave back the value
ok   and left it finished

The mistake that makes async pointless

await means "suspend me until this finishes". Two awaits on consecutive lines are therefore strictly sequential — which is correct code that does nothing concurrently, and looks identical to code that does.

await in a row is not concurrency

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26import asyncio, time

async def call(name, seconds):
    await asyncio.sleep(seconds)
    return name

async def one_after_another():
    start = time.perf_counter()
    a = await call("a", 0.05)          # waits here...
    b = await call("b", 0.05)          # ...and only then starts this
    return [a, b], time.perf_counter() - start

async def at_the_same_time():
    start = time.perf_counter()
    both = await asyncio.gather(call("a", 0.05), call("b", 0.05))
    return both, time.perf_counter() - start

seq_values, seq_time = asyncio.run(one_after_another())
par_values, par_time = asyncio.run(at_the_same_time())

print(f"two awaits in a row: {seq_time:.2f}s  -> {seq_values}")
print(f"gather:              {par_time:.2f}s  -> {par_values}")

test("both produce the same answer", seq_values, par_values)
test("sequential awaits took both waits", 0.09 < seq_time < 0.2, True)
test("gather took one of them", par_time < seq_time / 1.5, True)

What it printed, when the build ran it

two awaits in a row: 0.10s  -> ['a', 'b']
gather:              0.05s  -> ['a', 'b']
ok   both produce the same answer
ok   sequential awaits took both waits
ok   gather took one of them

The rule. await on its own line means "I need this before I continue". Concurrency comes from starting things before awaiting them — gather, or a TaskGroup, or create_task. If your async code is no faster than the blocking version, this is nearly always why.

`TaskGroup`, which is what you should reach for

gather is older and looser. A TaskGroup (3.11+) starts tasks, waits for all of them at the end of the with block, and — the part that matters — cancels the rest if one fails, so you never leave tasks running after an error.

TaskGroup: start inside, collect after

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22import asyncio, time

async def load(key):
    await asyncio.sleep(0.02)
    return key.upper()

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as group:
        tasks = {key: group.create_task(load(key)) for key in ("a", "b", "c")}
        # Inside the block the tasks are running. They are awaited on exit.
        started = [t.done() for t in tasks.values()]
    return {k: t.result() for k, t in tasks.items()}, started, time.perf_counter() - start

results, started, elapsed = asyncio.run(main())
print("results:", results)
print("were they finished inside the block?", started)
print(f"three 0.02s loads took {elapsed:.2f}s")

test("every task produced its value", results, {"a": "A", "b": "B", "c": "C"})
test("they had not finished before the block ended", started, [False, False, False])
test("and they overlapped", elapsed < 0.05, True)

What it printed, when the build ran it

results: {'a': 'A', 'b': 'B', 'c': 'C'}
were they finished inside the block? [False, False, False]
three 0.02s loads took 0.02s
ok   every task produced its value
ok   they had not finished before the block ended
ok   and they overlapped

When one of them fails

A TaskGroup raises an ExceptionGroup, because more than one task could have failed. except* is the syntax for catching inside one.

ExceptionGroup and except*

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27import asyncio

async def ok(n):
    await asyncio.sleep(0.02)
    return n

async def boom(n):
    await asyncio.sleep(0.01)
    raise ValueError(f"task {n} failed")

async def main():
    caught = []
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(ok(1))
            group.create_task(boom(2))
            group.create_task(boom(3))
    except* ValueError as group_of_errors:
        caught = [str(e) for e in group_of_errors.exceptions]
    return caught

caught = asyncio.run(main())
print("caught:", caught)
print("both failures survived, rather than only the first")

test("both failing tasks were reported", len(caught), 2)
test("with their own messages", sorted(caught), ["task 2 failed", "task 3 failed"])

What it printed, when the build ran it

caught: ['task 2 failed', 'task 3 failed']
both failures survived, rather than only the first
ok   both failing tasks were reported
ok   with their own messages

gather behaves differently, and the difference bites: by default the first exception propagates and the other results are lost. return_exceptions=True hands them back as values instead.

gather, and the results you can lose

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25import asyncio

async def ok(n):
    return n

async def boom(n):
    raise ValueError(f"no {n}")

async def main():
    try:
        await asyncio.gather(ok(1), boom(2), ok(3))
        raised = None
    except ValueError as exc:
        raised = str(exc)

    mixed = await asyncio.gather(ok(1), boom(2), ok(3), return_exceptions=True)
    return raised, mixed

raised, mixed = asyncio.run(main())
print("default gather raised:", raised)
print("with return_exceptions:", [type(m).__name__ if isinstance(m, Exception) else m for m in mixed])

test("the first failure propagated and the rest were discarded", raised, "no 2")
test("return_exceptions keeps the successes", [m for m in mixed if not isinstance(m, Exception)], [1, 3])
test("and gives you the failure as a value", type(mixed[1]).__name__, "ValueError")

What it printed, when the build ran it

default gather raised: no 2
with return_exceptions: [1, 'ValueError', 3]
ok   the first failure propagated and the rest were discarded
ok   return_exceptions keeps the successes
ok   and gives you the failure as a value

Forgetting `await`

There is no error for this. The coroutine is created, never driven, and discarded — your program does less than you wrote and says so only in a warning on stderr.

The missing await, and the warning it produces

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17import asyncio

log = []

async def record(event):
    log.append(event)

async def main():
    record("saved")          # no await: this never runs
    await record("really saved")
    return log

print("log:", asyncio.run(main()))
print()
print("one of those two calls did nothing at all.")

test("only the awaited call had an effect", log, ["really saved"])

What it printed, when the build ran it

log: ['really saved']

one of those two calls did nothing at all.
ok   only the awaited call had an effect
example.py:9: RuntimeWarning: coroutine 'record' was never awaited
  record("saved")          # no await: this never runs
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

That RuntimeWarning in the output is the only thing standing between you and a silent no-op. It is worth turning warnings into errors while you develop async code.

The one hard rule: never block the loop

Everything is on one thread. A call that does not await holds that thread, and every other task stops — including ones that were nearly finished.

One blocking call, three stalled tasks

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26import asyncio, time

async def blocking(n):
    time.sleep(0.05)              # WRONG: nothing else can run during this
    return n

async def cooperative(n):
    await asyncio.sleep(0.05)     # right: yields the thread while waiting
    return n

async def main():
    start = time.perf_counter()
    await asyncio.gather(*(blocking(i) for i in range(3)))
    blocked = time.perf_counter() - start

    start = time.perf_counter()
    await asyncio.gather(*(cooperative(i) for i in range(3)))
    return blocked, time.perf_counter() - start

blocked, yielded = asyncio.run(main())
print(f"three blocking sleeps, gathered:     {blocked:.2f}s")
print(f"three awaited sleeps, gathered:      {yielded:.2f}s")
print("the first three ran one after another, despite the gather.")

test("gathering blocking calls gained nothing", blocked > 0.13, True)
test("gathering awaited ones overlapped them", yielded < 0.09, True)

What it printed, when the build ran it

three blocking sleeps, gathered:     0.16s
three awaited sleeps, gathered:      0.05s
the first three ran one after another, despite the gather.
ok   gathering blocking calls gained nothing
ok   gathering awaited ones overlapped them

Real code still has blocking calls in it — a database driver, a file read, any library that was not written for async. asyncio.to_thread is the escape hatch: it runs the call on a worker thread and gives you something to await.

to_thread, for code you cannot change

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17import asyncio, threading, time

def read_config():                 # an ordinary blocking function
    time.sleep(0.05)
    return threading.current_thread() is threading.main_thread()

async def main():
    start = time.perf_counter()
    on_main = await asyncio.gather(*(asyncio.to_thread(read_config) for _ in range(4)))
    return on_main, time.perf_counter() - start

on_main, elapsed = asyncio.run(main())
print("did any of them run on the main thread?", any(on_main))
print(f"four 0.05s blocking calls: {elapsed:.2f}s")

test("to_thread moved the work off the loop's thread", any(on_main), False)
test("so the four overlapped", elapsed < 0.12, True)

What it printed, when the build ran it

did any of them run on the main thread? False
four 0.05s blocking calls: 0.06s
ok   to_thread moved the work off the loop's thread
ok   so the four overlapped

Limiting how much runs at once

Being able to start ten thousand requests is not permission to. A Semaphore is the standard way to cap it, and the cap is observable.

Semaphore, as a real limit

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26import asyncio

live = 0
peak = 0

async def fetch(n, gate):
    global live, peak
    async with gate:              # waits here if the cap is reached
        live += 1
        peak = max(peak, live)
        await asyncio.sleep(0.02)
        live -= 1
    return n

async def main():
    gate = asyncio.Semaphore(3)
    return await asyncio.gather(*(fetch(i, gate) for i in range(20)))

results = asyncio.run(main())
print("all finished:", len(results))
print("most that were ever in flight:", peak)
print("none left running:", live)

test("every request completed", sorted(results), list(range(20)))
test("never more than three at a time", peak, 3)
test("and the counter came back to zero", live, 0)

What it printed, when the build ran it

all finished: 20
most that were ever in flight: 3
none left running: 0
ok   every request completed
ok   never more than three at a time
ok   and the counter came back to zero

live += 1 needs no lock here. Lesson 41 needed one for exactly this code because a thread can be interrupted anywhere. A coroutine can only be suspended at an await, and there is none between the read and the write — so the update cannot be interleaved. Cooperative scheduling is what buys that, and it is the main thing async is easier at than threads.

Cancellation and timeouts

asyncio.timeout, and cleaning up on cancellation

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27import asyncio

cleaned_up = []

async def long_job():
    try:
        await asyncio.sleep(10)
        return "finished"
    except asyncio.CancelledError:
        cleaned_up.append("released the connection")
        raise                       # always re-raise: cancellation is not an error to swallow

async def main():
    try:
        async with asyncio.timeout(0.05):
            await long_job()
        timed_out = False
    except TimeoutError:
        timed_out = True
    return timed_out

timed_out = asyncio.run(main())
print("timed out:", timed_out)
print("cleanup that ran:", cleaned_up)

test("the timeout fired", timed_out, True)
test("and the job got a chance to clean up", cleaned_up, ["released the connection"])

What it printed, when the build ran it

timed out: True
cleanup that ran: ['released the connection']
ok   the timeout fired
ok   and the job got a chance to clean up

What it is actually for

Two thousand things waiting at once, in one thread, with no thread stacks to pay for.

Two thousand at once

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19import asyncio, time

async def request(n):
    await asyncio.sleep(0.1)         # stands in for a network round trip
    return n

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(request(i)) for i in range(2000)]
    return time.perf_counter() - start, sum(t.result() for t in tasks)

elapsed, total = asyncio.run(main())
print(f"2000 requests, each taking 0.1s: {elapsed:.2f}s")
print(f"the same work in a plain loop:   {2000 * 0.1:.0f}s")
print("sum of the results:", total)

test("all two thousand completed", total, sum(range(2000)))
test("in about the time one of them takes", elapsed < 1.0, True)

What it printed, when the build ran it

2000 requests, each taking 0.1s: 0.12s
the same work in a plain loop:   200s
sum of the results: 1999000
ok   all two thousand completed
ok   in about the time one of them takes

Where it does not help

Async is concurrency, not parallelism. There is one thread, so arithmetic gets no faster — and unlike threads, there is not even a second thread to interleave with.

Async on arithmetic

ran at build time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24import asyncio, time

async def burn(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

async def main():
    N = 3_000_000
    start = time.perf_counter()
    await burn(N)
    alone = time.perf_counter() - start

    start = time.perf_counter()
    await asyncio.gather(burn(N // 2), burn(N // 2))
    return alone, time.perf_counter() - start

alone, gathered = asyncio.run(main())
print(f"one coroutine:  {alone:.2f}s")
print(f"two gathered:   {gathered:.2f}s")
print("there is no await inside burn(), so there is nowhere to interleave.")

test("gathering CPU work bought nothing", gathered > alone * 0.7, True)

What it printed, when the build ran it

one coroutine:  0.21s
two gathered:   0.22s
there is no await inside burn(), so there is nowhere to interleave.
ok   gathering CPU work bought nothing

Two errors you will meet

await is only legal inside an async def. This one is caught before your program starts.

await outside a coroutine

raises SyntaxError

1
2def load():
    await fetch()

The traceback it really produced, when the build ran it

  File "example.py", line 2
    await fetch()
    ^^^^^^^^^^^^^
SyntaxError: 'await' outside async function

And asyncio.run creates a loop, so it cannot be called while one is running. Inside a coroutine you await instead.

asyncio.run inside a running loop

raises RuntimeError

1
2
3
4
5
6
7
8
9import asyncio

async def inner():
    return 1

async def outer():
    return asyncio.run(inner())     # should be: return await inner()

asyncio.run(outer())

The traceback it really produced, when the build ran it

Traceback (most recent call last):
  File "example.py", line 9, in <module>
    asyncio.run(outer())
    ~~~~~~~~~~~^^^^^^^^^
  File "<python>/lib/python3.14/asyncio/runners.py", line 205, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "<python>/lib/python3.14/asyncio/runners.py", line 128, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "<python>/lib/python3.14/asyncio/base_events.py", line 720, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File "example.py", line 7, in outer
    return asyncio.run(inner())     # should be: return await inner()
           ~~~~~~~~~~~^^^^^^^^^
  File "<python>/lib/python3.14/asyncio/runners.py", line 201, in run
    raise RuntimeError(
        "asyncio.run() cannot be called from a running event loop")
RuntimeError: asyncio.run() cannot be called from a running event loop
<sys>:0: RuntimeWarning: coroutine 'inner' was never awaited

Calling an async def returns a coroutine. asyncio.run drives it; one loop per program, at the top.
Consecutive awaits are sequential. Concurrency comes from TaskGroup, create_task or gather.
Prefer TaskGroup: it cancels siblings on failure and raises an ExceptionGroup you catch with except*.
Never block the loop. Wrap unavoidable blocking calls in asyncio.to_thread.
A forgotten await is not an error — only a warning on stderr. Watch for it.
Cap concurrency with Semaphore; bound time with asyncio.timeout and re-raise CancelledError.
Async is for waiting on many things. For arithmetic, use processes.

41 Threads, and what the GIL actually stops

A thread, at its simplest

Work that waits: threads earn their keep

Work that computes: threads do nothing

The GIL does not make your code safe

queue.Queue: the shape that avoids most locks

An exception in a thread does not stop your program

Where this leaves you

42 multiprocessing: real parallelism, real costs

The experiment that failed with threads

The guard is not a style rule

Nothing is shared

Everything crossing the boundary must pickle

The cost, measured

Sharing state anyway, when you must

Choosing between the three

43 async and await, and the loop underneath

An async function does not run when you call it

The mistake that makes async pointless

TaskGroup, which is what you should reach for

When one of them fails

Forgetting await

The one hard rule: never block the loop

Limiting how much runs at once

Cancellation and timeouts

What it is actually for

Where it does not help

Two errors you will meet

New writing, when it lands.

`queue.Queue`: the shape that avoids most locks

`TaskGroup`, which is what you should reach for

Forgetting `await`