Multiprocessing & Threading: Beating the GIL

Async handles concurrency for I/O-bound work: one thread, many awaits. But there are two other cases. The first is a library that only has a blocking API and won't be rewritten. The second is a numerical calculation that needs every core on your laptop. Async helps with neither.

Python has two heavier concurrency tools: threads (several flows of execution in one process, sharing memory) and processes (several OS processes, each with its own memory). Both use the same API, concurrent.futures, so switching between them is a one-line change. The hard part is knowing which one to choose.

The answer starts with the GIL.

1. The GIL: The Starting Point

CPython's Global Interpreter Lock is a mutex inside the interpreter. At any instant, exactly one thread is executing Python bytecode. Threads still exist, still run, and still help, but only when they are blocked on I/O or running C extension code that releases the GIL (NumPy, hashlib, file I/O, network calls).

The consequences:

• Threads for I/O: yes. A thread waiting on a socket has released the GIL, so another thread runs.
• Threads for CPU: no. A pure-Python loop holds the GIL. Ten threads on ten cores still do only as much work as one core. Sometimes they do less, because passing the lock between threads takes time.
• Processes for CPU: yes. Each Python process has its own GIL. Eight processes can use eight cores.

3.13 note: PEP 703 added an experimental free-threaded build of CPython (no GIL). It is optional, it needs a separate interpreter binary, and many C extensions don't support it yet. The standard build still has the GIL, so the decisions below still apply.

2. The Decision Matrix (Memorise This)

Three signs that you made the wrong choice:

1. "I added threads and the CPU work isn't any faster" → GIL. Use processes.
2. "I added processes and now my objects don't share state" → processes have separate memory. Use a Manager, Queue, or shared_memory.
3. "I added async and the requests library still blocks everything" → requests is sync. Use httpx.AsyncClient, or run requests through asyncio.to_thread.

3. threading.Thread: The Low-Level API

You will mostly use ThreadPoolExecutor (Section 7), but it is worth seeing the basic tools once.

import threading
import time

def worker(name, delay):
    print(f"{name}: starting")
    time.sleep(delay)
    print(f"{name}: done")

t1 = threading.Thread(target=worker, args=("a", 1.0))
t2 = threading.Thread(target=worker, args=("b", 0.5))

t1.start(); t2.start()
t1.join(); t2.join()                         # wait for both to finish
print("all done")

start() schedules the thread, and the OS decides when it really runs. join() blocks until the thread finishes. Without join, the main thread can finish while workers are still running. This is usually fine, but sometimes it causes serious bugs.

Daemon vs non-daemon: a daemon thread (Thread(target=..., daemon=True)) is killed when the main thread exits. Non-daemon threads keep the program alive until they finish. Use daemon threads for background workers like "do something every minute". Use non-daemon threads for real work that the program depends on.

4. Locks, RLocks, Semaphores

When several threads write to the same data structure without synchronisation, the data can become corrupted. Locks prevent this.

import threading

counter = 0
lock = threading.Lock()

def increment(n):
    global counter
    for _ in range(n):
        with lock:                           # only one thread inside at a time
            counter += 1                     # the unsafe critical section

threads = [threading.Thread(target=increment, args=(100_000,)) for _ in range(8)]
for t in threads: t.start()
for t in threads: t.join()

print(counter)                               # 800_000 (without the lock you'd see a smaller, random number)

The primitives:

• Lock: mutual exclusion. One holder at a time. Cannot be re-acquired by the same thread (deadlock).
• RLock: re-entrant lock. The same thread can acquire multiple times (releases must match). Use when a function holding the lock calls another that also needs it.
• Semaphore(n): allows up to n holders at the same time. The threaded equivalent of asyncio.Semaphore. Use to cap concurrent HTTP requests, DB connections, file handles.
• Event: a flag for "something happened". Threads wait() until another thread set()s it.
• Condition: a lock plus a wait/notify protocol, used for producer/consumer signalling. You can almost always use queue.Queue (next section) instead, which is easier and correct by default.

Always use with lock:. If a lock is never released (because of an early return, an exception, or anything else), every other thread that needs it waits forever (a deadlock).

5. queue.Queue: Thread-Safe Data Passing

Locks are easy to use incorrectly. A queue is hard to use incorrectly. For "one thread produces, another consumes," try queue.Queue first.

import queue, threading, time

q = queue.Queue(maxsize=10)

def producer():
    for i in range(20):
        q.put(i)                             # blocks if queue is full
        print(f"produced {i}")
    q.put(None)                              # sentinel: tells consumer to stop

def consumer():
    while True:
        item = q.get()                       # blocks until an item is available
        if item is None:
            q.task_done()
            return
        print(f"  consumed {item}")
        q.task_done()

threading.Thread(target=producer).start()
threading.Thread(target=consumer).start()

Queue is thread-safe: every put and get is atomic. The producer can add items quickly from one thread while the consumer removes them from another, and the data stays correct. The maxsize limit gives you backpressure: a fast producer must wait for a slow consumer, so the queue can't grow until the program runs out of memory.

Variants: LifoQueue (stack), PriorityQueue (priority-ordered tuples). Same API.

6. Thread-Local Data

Sometimes you want each thread to have its own copy of a "global" value, such as a database connection, a request context, or a per-thread cache. threading.local() stores these.

import threading

ctx = threading.local()

def init_thread(name):
    ctx.user = name                          # only this thread sees it

def use_thread():
    print(f"hello, {ctx.user}")              # works only after init_thread ran in the same thread

threading.Thread(target=lambda: (init_thread("alice"), use_thread())).start()
threading.Thread(target=lambda: (init_thread("bob"), use_thread())).start()

Each thread gets its own attribute namespace on the local() object. Read it on a thread that hasn't set it and you get AttributeError. Older versions of Flask used this for the per-request context. Modern code, including current Flask, uses contextvars instead, which also works with async.

7. ThreadPoolExecutor: The Preferred API

In modern code, you will rarely create Thread objects by hand. concurrent.futures.ThreadPoolExecutor is the high-level pool: submit work, get futures back, let the pool manage the threads.

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch(url):
    time.sleep(0.5)                          # pretend this is a network call
    return f"got {url}"

urls = [f"https://x/{i}" for i in range(20)]

with ThreadPoolExecutor(max_workers=8) as pool:
    # Option A: map gives results in input order
    for result in pool.map(fetch, urls):
        print(result)

    # Option B: submit + as_completed gives results as each one finishes
    futures = {pool.submit(fetch, u): u for u in urls}
    for fut in as_completed(futures):
        url = futures[fut]
        try:
            print(url, "->", fut.result())
        except Exception as e:
            print(url, "FAILED:", e)

The three methods you will use most:

• pool.submit(fn, *args): returns a Future. Call .result() to block until it is done (or to raise the function's exception). The other methods are built on this one.
• pool.map(fn, iterable): a lazy iterator over the results, in input order. Use it when order matters and you don't need error handling for each item.
• as_completed(futures): yields futures as they finish. Best when items take different times and you want to start handling fast ones immediately.

The with block joins all workers before it exits, so no threads are left running.

8. ProcessPoolExecutor: Same API, Separate Processes

from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    # a pure-Python CPU loop, which threads would not run in parallel
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(heavy, [10_000_000] * 8))
    print(results)

The API is deliberately the same as ThreadPoolExecutor. Replace ThreadPoolExecutor with ProcessPoolExecutor and you switch from "shared memory, limited by the GIL" to "separate processes, true parallelism." The cost is serialisation: arguments and return values are pickled so they can move between processes.

if __name__ == "__main__": is required on Windows. We'll explain why in Section 11.

9. multiprocessing.Pool: The Older API

Before concurrent.futures (Python 3.2), multiprocessing.Pool was the standard way. It is still used, and you will see it in older codebases. The shape is similar:

from multiprocessing import Pool

if __name__ == "__main__":
    with Pool(processes=4) as p:
        print(p.map(heavy, [10_000_000] * 8))
        print(p.apply(heavy, (1_000_000,)))                  # one call, blocks until done
        print(p.apply_async(heavy, (1_000_000,)).get())      # one call, async (.get() waits)
        # imap_unordered: stream results as they finish, like as_completed
        for r in p.imap_unordered(heavy, [10_000_000] * 8):
            print(r)

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

heavy = _AutoMock('heavy')

Reasons you might still use it:

• imap_unordered with a chunksize is often the fastest way to stream a big batch through a process pool.
• Existing code uses it and rewriting isn't justified.

For new code, prefer ProcessPoolExecutor. It can do the same work, it is easier to use, and it works with asyncio.run_in_executor.

10. Sharing Data Between Processes

Threads share memory automatically. Processes do not. There are three ways to share data between them:

multiprocessing.Queue: message passing

from multiprocessing import Process, Queue

def worker(q):
    q.put("hello from worker")

if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    print(q.get())                           # "hello from worker"
    p.join()

This is different from queue.Queue: it is process-safe (it uses a pipe and locks internally). Items are pickled on put and unpickled on get.

multiprocessing.Manager: proxied mutable state

from multiprocessing import Manager, Process

def append(shared_list, item):
    shared_list.append(item)

if __name__ == "__main__":
    with Manager() as m:
        shared = m.list()                    # proxy to a list living in a manager process
        ps = [Process(target=append, args=(shared, i)) for i in range(5)]
        for p in ps: p.start()
        for p in ps: p.join()
        print(list(shared))                  # [0, 1, 2, 3, 4] (the order can vary)

A Manager starts a separate server process that owns the real objects, and every other process gets proxies. Every method call serialises the arguments, sends them to the server process, and sends the results back. This is convenient, but slow. Don't use it for objects that your code accesses very often.

multiprocessing.shared_memory: zero-copy bytes (3.8+)

from multiprocessing import shared_memory, Process
import numpy as np

def increment(name, shape, dtype):
    shm = shared_memory.SharedMemory(name=name)
    arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    arr += 1
    shm.close()

if __name__ == "__main__":
    arr = np.zeros((1_000_000,), dtype=np.int64)
    shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
    backed = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
    backed[:] = arr

    procs = [Process(target=increment, args=(shm.name, arr.shape, arr.dtype))
             for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()

    print(backed[:5])                        # [4 4 4 4 4]
    shm.close()
    shm.unlink()                             # free the shared memory block

A raw memory block visible to every process by name. With NumPy, this lets you share gigabytes of data between processes for the cost of an mmap, with no pickling and no copying. It is the fastest option, and also the one most likely to corrupt data if you don't add your own locks. Use it only when you have measured that pickling is the bottleneck.

11. if __name__ == "__main__": and the Windows Spawn Trap

from multiprocessing import Process

def hi(): print("hi from child")

# DANGER on Windows
p = Process(target=hi)
p.start()

On macOS (since 3.8) and Windows, multiprocessing uses the spawn start method: the child process launches a fresh Python interpreter and re-imports your script. If the top level of your script also starts processes, the child tries to start them again while it is still importing, and Python stops it with RuntimeError: An attempt has been made to start a new process before the current process has finished its bootstrapping phase.

The fix is the boilerplate every multiprocessing example has:

if __name__ == "__main__":
    p = Process(target=hi)
    p.start()
    p.join()

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def Process(*_a, **_kw):
    print('-> Process() called')
    return _AutoMock('Process()')
hi = _AutoMock('hi')

When the child re-imports the script, __name__ is "__mp_main__", not "__main__", so the block doesn't run again. Everything inside that guard is "code that runs only in the parent."

On Linux, the default was fork (no re-import) until Python 3.14, which changed it to forkserver. Like spawn, forkserver re-imports your script, so the guard is needed on Linux too from 3.14. Portable code should follow the spawn rules everywhere. The guard costs nothing.

12. Map-Reduce Pattern

This is the most common use of a process pool: split the work, run the map step in parallel, and combine the results.

from concurrent.futures import ProcessPoolExecutor

def word_counts(text_chunk):
    counts = {}
    for word in text_chunk.split():
        counts[word] = counts.get(word, 0) + 1
    return counts

def merge(a, b):
    out = dict(a)
    for k, v in b.items():
        out[k] = out.get(k, 0) + v
    return out

def parallel_word_count(chunks):
    with ProcessPoolExecutor() as pool:
        per_chunk = pool.map(word_counts, chunks)
    total = {}
    for d in per_chunk:
        total = merge(total, d)
    return total

if __name__ == "__main__":
    chunks = ["the quick brown fox"] * 8
    print(parallel_word_count(chunks))

Map runs in parallel across workers. Reduce (merge) runs in the main process. It is usually fast, because each map result is already a summary. MapReduce and Spark use the same structure, but across a cluster of machines.

13. A Benchmark to Remember

The standard demo: a CPU-bound function, run serially, then with a thread pool, then with a process pool.

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def heavy(n):
    return sum(i * i for i in range(n))

WORK = [10_000_000] * 8

def serial():
    return [heavy(n) for n in WORK]

def threads():
    with ThreadPoolExecutor(max_workers=8) as pool:
        return list(pool.map(heavy, WORK))

def processes():
    with ProcessPoolExecutor(max_workers=8) as pool:
        return list(pool.map(heavy, WORK))

if __name__ == "__main__":
    for label, fn in [("serial", serial), ("threads", threads), ("processes", processes)]:
        t0 = time.perf_counter()
        fn()
        print(f"{label:>10}: {time.perf_counter()-t0:.2f}s")

On a typical 8-core laptop:

    serial: ~6.0s
   threads: ~6.0s   (no faster, because the GIL runs one thread at a time)
 processes: ~0.9s   (~6.7× faster, limited by core count and pickling cost)

Threads gave no speedup. Processes gave almost linear scaling. The GIL causes that difference. If your workload looks like this, multiprocessing is the only solution in the standard library.

(If heavy were numpy.sum(numpy.arange(n)**2) instead, the loop would run in C and release the GIL, so threads would scale too. Knowing which kind of code you have is a large part of optimisation.)

Common Mistakes

1. Threading CPU-bound code and expecting speedup

This is the most common misunderstanding. Pure-Python loops do not run in parallel with threads. Profile first. If the hot loop is Python bytecode, switch to processes or move the loop into NumPy/C.

2. Forgetting if __name__ == "__main__": on Windows/macOS

Each child crashes with a RuntimeError about the "bootstrapping phase" instead of running your code. The guard is required in any module that calls Process, Pool, or ProcessPoolExecutor.

3. Mutating "shared" state in subprocesses

from concurrent.futures import ProcessPoolExecutor

counter = 0
def bump(_): global counter; counter += 1

if __name__ == "__main__":
    with ProcessPoolExecutor() as p:
        list(p.map(bump, range(10)))

    print(counter)                           # 0 (each child process has its own `counter`)

Each subprocess gets a copy of the parent's state at spawn time. Changes in a child are not copied back to the parent. Use Manager, Queue, or shared_memory. Or avoid shared state completely: return values from the workers and combine them in the parent.

4. Deadlocks from inconsistent lock ordering

Thread 1 holds A, wants B. Thread 2 holds B, wants A. Both wait forever. The fix: choose one global order (for example, by id(lock)) and always acquire locks in that order. Even better, design the code so that no thread needs two locks at once.

5. Pickle errors on closures, lambdas, local functions

def outer():
    def inner(x): return x * 2
    with ProcessPoolExecutor() as p:
        list(p.map(inner, range(10)))        # PicklingError: can't pickle local function

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def ProcessPoolExecutor(*_a, **_kw):
    print('-> ProcessPoolExecutor() called')
    return _AutoMock('ProcessPoolExecutor()')

Process pools pickle the function to ship it to workers. Pickling needs the function to be importable at module top level. Move inner to the top level of the module, or use multiprocess (a third-party fork that uses dill). Module-level functions are the better solution.

6. Pool overhead larger than the work

Submitting 100,000 tiny tasks to a process pool spends more time pickling than computing. Two fixes: batch the work (each task does 1000 items instead of 1) or pass chunksize=N to pool.map so the pool batches for you.

7. Not joining/closing the pool

Always with ThreadPoolExecutor(...) as pool:. The with block calls shutdown(wait=True) on exit: the workers finish their tasks, and the threads stop. Without it, workers can outlive the main thread and prevent the program from exiting.

🎯 Your Turn: parallel_map(fn, items, mode="thread")

Write a function that parallelises a map across either threads or processes, picked by a mode argument. Then benchmark it on a CPU-bound function and check that only process mode makes it faster.

Requirements:

• Signature: parallel_map(fn, items, mode="thread", max_workers=None).
• mode="thread" uses ThreadPoolExecutor, mode="process" uses ProcessPoolExecutor, and anything else raises ValueError.
• Returns results in input order (same as pool.map).
• If max_workers is None, let the executor choose (it defaults to a reasonable value).
• Include a benchmark block (under if __name__ == "__main__":) that times a CPU-bound function in serial, thread, and process mode.

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def parallel_map(fn, items, mode="thread", max_workers=None):
    # TODO 1: pick the executor class based on `mode`
    # TODO 2: open the executor in a `with` block
    # TODO 3: return list(executor.map(fn, items))
    ...

def cpu_heavy(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    work = [2_000_000] * 8
    # TODO 4: time serial baseline
    # TODO 5: time parallel_map(cpu_heavy, work, mode="thread")
    # TODO 6: time parallel_map(cpu_heavy, work, mode="process")
    # Print the three timings.

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time


def parallel_map(fn, items, mode="thread", max_workers=None):
    """Parallelise `fn` over `items` using threads or processes.

    Results come back in input order, like the built-in `map`.
    """
    executor_cls = {
        "thread": ThreadPoolExecutor,
        "process": ProcessPoolExecutor,
    }.get(mode)
    if executor_cls is None:
        raise ValueError(f"mode must be 'thread' or 'process', got {mode!r}")

    with executor_cls(max_workers=max_workers) as pool:
        return list(pool.map(fn, items))


def cpu_heavy(n):
    """Pure-Python CPU loop. It does NOT release the GIL."""
    return sum(i * i for i in range(n))


def timed(label, fn):
    t0 = time.perf_counter()
    fn()
    elapsed = time.perf_counter() - t0
    print(f"{label:>14}: {elapsed:.2f}s")


if __name__ == "__main__":
    work = [2_000_000] * 8

    timed("serial",       lambda: [cpu_heavy(n) for n in work])
    timed("thread pool",  lambda: parallel_map(cpu_heavy, work, mode="thread"))
    timed("process pool", lambda: parallel_map(cpu_heavy, work, mode="process"))

        serial: 1.20s
   thread pool: 1.22s    <- no speedup, because the GIL runs one thread at a time
  process pool: 0.18s    <- ~6.5x faster, because each process has its own GIL

What You Learned

• The GIL: only one thread runs Python bytecode at a time in each process. Threads help with I/O and GIL-releasing C code, but they don't help with pure-Python CPU work.
• Decision matrix: I/O + async libs → asyncio, I/O + blocking libs → threads, CPU-bound → processes.
• threading.Thread, Lock, RLock, Semaphore: the low-level tools. Always acquire under with.
• queue.Queue: thread-safe producer/consumer. Almost always better than building your own with locks.
• threading.local(): per-thread state. contextvars is the async-friendly modern alternative.
• concurrent.futures.ThreadPoolExecutor: the modern, preferred thread API. submit returns a Future, map preserves order, and as_completed yields in completion order.
• concurrent.futures.ProcessPoolExecutor: the same API with separate processes and true parallelism. Arguments and results are pickled.
• multiprocessing.Pool: the older API, but imap_unordered is still useful. New code: prefer ProcessPoolExecutor.
• Sharing across processes: Queue (messages), Manager (proxied state, slow), shared_memory (zero-copy bytes, fast but risky).
• if __name__ == "__main__": is required with multiprocessing on Windows/macOS (spawn), and on Linux from Python 3.14 (forkserver). Without it, each child process re-imports the script and fails with a RuntimeError.
• Map-reduce: pool.map(map_fn, chunks) + a serial reduce. Many batch jobs have this structure.
• Pickling cost: function arguments and return values are serialised when they move between processes. Batch small tasks (chunksize=) and avoid large per-call payloads.
• Pool overhead can exceed benefit for tiny tasks. Measure before parallelising.

Next: Type Hints in Depth: Protocol, TypeVar, Generic, Self, ParamSpec, and how the modern Python type system catches bugs where your code meets other code.