Python Concurrency

Python has three concurrency models — threads, processes and asyncio — and the right one depends on whether your program is waiting (I/O-bound) or computing (CPU-bound). Output below is from CPython 3.13 on a 4-core machine; timings vary run to run.

Threads for blocking I/O

time.sleep stands in for a blocking network call here; like real I/O, it releases the GIL while it waits.

import threading
import time

def fetch(url: str, results: dict) -> None:
    time.sleep(0.5)                       # blocking I/O stand-in
    results[url] = 200

urls = ["/a", "/b", "/c", "/d"]
results: dict[str, int] = {}
start = time.perf_counter()

threads = [threading.Thread(target=fetch, args=(u, results)) for u in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()                              # wait for every thread to finish

print(f"elapsed {time.perf_counter() - start:.1f}s")
print(dict(sorted(results.items())))

$ python threads_io.py
elapsed 0.5s
{'/a': 200, '/b': 200, '/c': 200, '/d': 200}

Four half-second waits took half a second in total: while one thread sleeps, the others run. join() is the synchronisation point — without it, print would run before the threads had filled results. The dict is sorted before printing because threads finish in no guaranteed order.

The GIL: threads don’t speed up CPU work

Now the same four tasks are pure computation, run on threads and then on processes.

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def count_down(n: int) -> int:
    while n > 0:
        n -= 1
    return n

def timed(executor_cls) -> float:
    start = time.perf_counter()
    with executor_cls(max_workers=4) as ex:
        list(ex.map(count_down, [5_000_000] * 4))
    return time.perf_counter() - start

if __name__ == "__main__":
    start = time.perf_counter()
    for _ in range(4):
        count_down(5_000_000)
    print(f"sequential: {time.perf_counter() - start:.2f}s")
    print(f"threads:    {timed(ThreadPoolExecutor):.2f}s")
    print(f"processes:  {timed(ProcessPoolExecutor):.2f}s")

$ python gil.py
sequential: 0.72s
threads:    0.87s
processes:  0.23s

Threads were no faster than running the work sequentially — in this run slightly slower, from the cost of switching between them. The GIL lets only one thread execute bytecode at a time, so they just take turns. Processes each have their own interpreter and GIL, so four cores did four tasks at once. Your numbers will differ, and threads can land slightly either side of sequential depending on the interpreter version and scheduling. What holds generally is the shape: for CPU-bound work under the GIL, threads get nowhere near a 4x speedup on four cores, and processes do.

Race conditions and Lock

counter += 1 is a read, an add and a write. This version makes the gap between read and write explicit so the race shows up every time.

import threading
import time

counter = 0

def increment(times: int) -> None:
    global counter
    for _ in range(times):
        current = counter
        time.sleep(0)                  # let another thread run mid-update
        counter = current + 1

threads = [threading.Thread(target=increment, args=(100,)) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter == 1000, counter < 1000)

$ python race.py
False True

Two threads read the same value, both add one, and both write it back — one increment is lost. The exact total changes every run, so the program prints only the comparison. Wrap the read-modify-write in a lock and the result is exact:

import threading
import time

counter = 0
lock = threading.Lock()

def increment(times: int) -> None:
    global counter
    for _ in range(times):
        with lock:                     # only one thread inside at a time
            current = counter
            time.sleep(0)
            counter = current + 1

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

$ python race_fixed.py
1000

Other threads still get scheduled during sleep(0), but they block at with lock: until the holder finishes its update.

Producer and consumer with queue.Queue

Instead of sharing a variable, threads can hand work to each other through a thread-safe queue.

import queue
import threading

tasks: queue.Queue[int | None] = queue.Queue()
results: queue.Queue[int] = queue.Queue()

def worker() -> None:
    while (item := tasks.get()) is not None:   # None means "stop"
        results.put(item * item)

workers = [threading.Thread(target=worker) for _ in range(3)]
for w in workers:
    w.start()
for n in range(1, 7):
    tasks.put(n)
for _ in workers:
    tasks.put(None)                            # one stop signal per worker
for w in workers:
    w.join()

print(sorted(results.get() for _ in range(results.qsize())))

$ python producer_consumer.py
[1, 4, 9, 16, 25, 36]

tasks.get() blocks until an item arrives, so idle workers wait without spinning. Each worker needs its own None sentinel to exit its loop; join() then guarantees every result is in the queue before we read it.

concurrent.futures: one API for threads and processes

submit returns a Future; as_completed yields futures as they finish, and .result() re-raises a worker’s exception in the caller.

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def download(name: str, seconds: float) -> str:
    time.sleep(seconds)
    if name == "bad.csv":
        raise ConnectionError(f"{name}: connection reset")
    return f"{name} ok"

jobs = {"big.csv": 0.6, "small.csv": 0.1, "bad.csv": 0.3}
with ThreadPoolExecutor(max_workers=3) as ex:
    futures = {ex.submit(download, n, s): n for n, s in jobs.items()}
    for future in as_completed(futures):
        try:
            print(future.result())
        except ConnectionError as e:
            print("failed:", e)

$ python futures.py
small.csv ok
failed: bad.csv: connection reset
big.csv ok

Results arrive in completion order, not submission order. The exception raised inside a worker thread was stored on its future and re-raised by .result(), where an ordinary try catches it. Swap in ProcessPoolExecutor (under an if __name__ == "__main__": guard) and the code is otherwise unchanged.

Calling blocking code from asyncio

asyncio.to_thread runs a blocking function in a worker thread so the event loop stays responsive.

import asyncio
import time

def blocking_read(name: str) -> str:
    time.sleep(0.3)                    # a library with no async API
    return f"{name} read"

async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        asyncio.to_thread(blocking_read, "a.txt"),
        asyncio.to_thread(blocking_read, "b.txt"),
    )
    print(results, f"{time.perf_counter() - start:.1f}s")

asyncio.run(main())

$ python to_thread.py
['a.txt read', 'b.txt read'] 0.3s

Calling blocking_read directly inside a coroutine would freeze the whole loop for 0.3s per call; pushed onto threads, the two calls overlap.

Choosing the right tool

Practice

1. Rewrite threads_io.py with ThreadPoolExecutor.map so no manual start/join is needed.

import time
from concurrent.futures import ThreadPoolExecutor

def fetch(url: str) -> tuple[str, int]:
    time.sleep(0.5)
    return url, 200

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as ex:
    results = dict(ex.map(fetch, ["/a", "/b", "/c", "/d"]))
print(f"elapsed {time.perf_counter() - start:.1f}s")
print(results)

$ python pool_map.py
elapsed 0.5s
{'/a': 200, '/b': 200, '/c': 200, '/d': 200}

2. Run ThreadPoolExecutor(max_workers=2) over the same four URLs and predict the elapsed time before you run it.

import time
from concurrent.futures import ThreadPoolExecutor

def fetch(url: str) -> int:
    time.sleep(0.5)
    return 200

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=2) as ex:
    list(ex.map(fetch, ["/a", "/b", "/c", "/d"]))
print(f"elapsed {time.perf_counter() - start:.1f}s")

$ python two_workers.py
elapsed 1.0s

3. Make the producer/consumer example record which worker handled how many items, using threading.current_thread().name and a Counter guarded by a Lock, and print only the total.

import queue
import threading
from collections import Counter

tasks: queue.Queue[int | None] = queue.Queue()
handled: Counter[str] = Counter()
lock = threading.Lock()

def worker() -> None:
    while (item := tasks.get()) is not None:
        with lock:
            handled[threading.current_thread().name] += 1

workers = [threading.Thread(target=worker, name=f"w{i}") for i in range(3)]
for w in workers:
    w.start()
for n in range(30):
    tasks.put(n)
for _ in workers:
    tasks.put(None)
for w in workers:
    w.join()
print(sum(handled.values()), set(handled) <= {"w0", "w1", "w2"})

$ python per_worker.py
30 True

4. Use a threading.Event so a worker thread waits for a “go” signal from the main thread before it starts.

import threading

go = threading.Event()
log: list[str] = []

def worker() -> None:
    log.append("worker waiting")
    go.wait()                       # blocks until set()
    log.append("worker running")

t = threading.Thread(target=worker)
t.start()
while "worker waiting" not in log:
    pass                            # spin until the worker is really waiting
log.append("main sets go")
go.set()
t.join()
print(log)

$ python event.py
['worker waiting', 'main sets go', 'worker running']