Threads

A thread is a separate line of execution inside your program. All threads share the same objects, which makes them easy to start and easy to get wrong. On the default build of CPython they take turns running Python code, as the previous lesson explained, so use them for work that waits: network calls, disk access, slow APIs.

Starting and joining

threading.Thread takes a target function and its args. start() launches it, and join() waits until it has finished:

import threading
import time

results = {}


def fetch(name, delay):
    time.sleep(delay)  # stands in for a network call
    results[name] = len(name)


threads = [
    threading.Thread(target=fetch, args=(name, 0.1))
    for name in ["alpha", "beta", "gamma"]
]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

print(sorted(results.items()))

[('alpha', 5), ('beta', 4), ('gamma', 5)]

The three sleeps overlap, so the program takes about 0.1 seconds, not 0.3. Start every thread first, then join them all. Joining inside the first loop would wait for each thread before starting the next, which is just a slow sequential program.

The threads finish in an order you do not control, so the program prints sorted(...). Treat any order that comes out of threads as random unless you impose one.

A race condition, on purpose

Here is the classic bug. balance = balance + 1 looks like one step but is two: read balance, then write a new value. If two threads both read before either writes, one deposit is lost. That timing is rare, so a buggy program can pass its tests for months. To make it happen every time, this example holds both threads at a Barrier between the read and the write:

import threading

balance = 0
both_have_read = threading.Barrier(2)


def deposit(amount):
    global balance
    current = balance          # 1. read
    both_have_read.wait()      # force the unlucky schedule
    balance = current + amount  # 2. write


threads = [threading.Thread(target=deposit, args=(1,)) for _ in range(2)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

print("expected 2, got", balance)

expected 2, got 1

Both threads read 0, both write 1. Without the barrier the same interleaving can still happen, just not reliably. That is what a race condition is: the result depends on the order in which threads happen to run. The GIL does not prevent it, because a thread can be switched out between any two bytecode instructions.

Locks

A threading.Lock lets only one thread at a time into a block of code. Use it as a context manager, so it is released even if the block raises:

import threading

balance = 0
balance_lock = threading.Lock()


def deposit_many(times):
    global balance
    for _ in range(times):
        with balance_lock:
            balance = balance + 1


threads = [
    threading.Thread(target=deposit_many, args=(50_000,))
    for _ in range(4)
]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

print("expected 200000, got", balance)

expected 200000, got 200000

The read and the write now happen together, as one critical section. Every piece of code that touches balance must take the same lock. A lock that only some code paths use protects nothing.

Keep critical sections short and never call unknown code while holding a lock. If thread A holds lock 1 and waits for lock 2 while thread B holds lock 2 and waits for lock 1, both wait forever. That is a deadlock. Taking locks in one fixed order everywhere avoids it. On the free-threaded build, threads really run at the same time, so a missing lock shows up far sooner.

Re-entrant locks: RLock

A Lock does not know who holds it. If the thread that holds it tries to take it again, it waits for itself forever. A threading.RLock (re-entrant lock) counts how many times its owner has taken it, and is free again only after the same number of releases:

import threading

plain = threading.Lock()
with plain:
    print("Lock taken again:", plain.acquire(timeout=0.1))

reentrant = threading.RLock()
with reentrant:
    print("RLock taken again:", reentrant.acquire(timeout=0.1))
    reentrant.release()


class Account:
    def __init__(self):
        self.balance = 0
        self._lock = threading.RLock()

    def deposit(self, amount):
        with self._lock:
            self.balance += amount

    def deposit_twice(self, amount):
        with self._lock:  # holds the lock while deposit takes it again
            self.deposit(amount)
            self.deposit(amount)


account = Account()
account.deposit_twice(5)
print("balance:", account.balance)

Lock taken again: False
RLock taken again: True
balance: 10

acquire(timeout=0.1) gave up and returned False on the plain lock, instead of hanging. An RLock suits a class whose public methods lock and call each other. It is slightly slower than a Lock, and it hides the question of which method really owns the lock, so prefer a Lock and private helpers that assume it is held.

Queues

Often the cleanest design shares nothing at all: threads pass messages through a queue.Queue, which does its own locking. Stacks and queues introduced the data structure. Here, three worker threads take words from one queue and put results on another:

import queue
import threading

jobs = queue.Queue()
done = queue.Queue()
STOP = None


def worker():
    while (word := jobs.get()) is not STOP:
        done.put(word.upper())


workers = [threading.Thread(target=worker) for _ in range(3)]
for w in workers:
    w.start()
for word in ["pear", "fig", "kiwi", "plum", "lime"]:
    jobs.put(word)
for _ in workers:
    jobs.put(STOP)  # one stop signal per worker
for w in workers:
    w.join()

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

['FIG', 'KIWI', 'LIME', 'PEAR', 'PLUM']

jobs.get() blocks until an item arrives, so idle workers cost nothing. Each worker stops when it receives the STOP sentinel, so put one sentinel per worker. 3.13+Requires version 3.13 or later jobs.shutdown() is the alternative: after it, get on an empty queue raises queue.ShutDown in every waiting worker.

Signalling with Event

A threading.Event is a flag that threads can wait on. set() raises it, wait() blocks until it is raised, and wait(timeout) returns False if the time ran out first. That makes it the clean way to say “I am ready” and “please stop”, instead of a global variable that a loop polls:

import threading

ready = threading.Event()
stop = threading.Event()


def server():
    print("server: listening")
    ready.set()
    while not stop.wait(timeout=0.05):
        pass  # check for work here, then wait again
    print("server: shutting down")


thread = threading.Thread(target=server)
thread.start()
ready.wait()
print("main: server is ready")
stop.set()
thread.join()
print("main: done")

server: listening
main: server is ready
server: shutting down
main: done

The order is guaranteed: the main thread cannot pass ready.wait() until the server has printed and set the flag. stop.wait(timeout=0.05) also serves as the loop’s pause, and it returns as soon as stop is set, rather than after a fixed sleep.

Limiting with Semaphore

A threading.Semaphore(n) is a lock that n threads can hold at once, for limiting how many downloads or database connections run in parallel. A BoundedSemaphore also raises ValueError when it is released more often than acquired, which catches a common bug:

import threading
import time

slots = threading.BoundedSemaphore(2)  # at most two downloads at once
lock = threading.Lock()
active = peak = 0


def download(url):
    global active, peak
    with slots:
        with lock:
            active += 1
            peak = max(peak, active)
        time.sleep(0.05)  # the download itself
        with lock:
            active -= 1


threads = [threading.Thread(target=download, args=(n,)) for n in range(6)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
print("most at once:", peak)

try:
    slots.release()  # one release too many
except ValueError as error:
    print("ValueError:", error)

most at once: 2
ValueError: Semaphore released too many times

Six threads started, but no more than two were ever inside the with slots: block. Use BoundedSemaphore unless you have a reason not to. With a thread pool, max_workers does the same job for you.

Waiting for a state: Condition

A threading.Condition combines a lock with waiting. A thread holds the condition, checks some shared state, and if it is not ready, wait() releases the lock and sleeps until another thread calls notify_all(). wait_for(predicate) does the check-and-wait loop for you:

import threading

basket = []
changed = threading.Condition()


def picker(fruit):
    with changed:
        basket.append(fruit)
        changed.notify_all()  # wake everyone waiting on this condition


def packer():
    with changed:
        changed.wait_for(lambda: len(basket) >= 3)
        print("packing", sorted(basket))


threads = [threading.Thread(target=packer)]
threads += [threading.Thread(target=picker, args=(f,)) for f in ["fig", "kiwi", "lime"]]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

packing ['fig', 'kiwi', 'lime']

The packer may start first or last. Either way, it sleeps until the basket holds three items, and it reads the basket while holding the lock. queue.Queue is built on conditions, so reach for a queue first, and for a Condition when threads wait for a state that is not “an item arrived”.

Thread pools

Creating, starting and joining threads by hand gets repetitive. concurrent.futures.ThreadPoolExecutor keeps a pool of worker threads and hands them jobs. Its map works like the built-in map, and returns results in input order, whatever order the jobs finish in:

import time
from concurrent.futures import ThreadPoolExecutor

PAGES = {
    "/": "<h1>Home</h1>",
    "/about": "<p>About us</p>",
    "/contact": "<p>Write to us</p>",
}


def fetch(path):
    time.sleep(0.1)  # pretend to wait for the network
    return len(PAGES[path])


with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = pool.map(fetch, PAGES)
    for path, size in zip(PAGES, sizes):
        print(f"{path:10} {size:3} bytes")

/           13 bytes
/about      15 bytes
/contact    18 bytes

Leaving the with block waits for all jobs and shuts the pool down. From async code, asyncio.to_thread runs a blocking function on a thread like this one.

For more control, submit schedules one call and returns a future, an object that will hold the result. as_completed yields futures as they finish. future.result() returns the value, or raises the exception the call raised:

from concurrent.futures import ThreadPoolExecutor, as_completed


def parse_port(text):
    port = int(text)
    if not 0 < port < 65536:
        raise ValueError(f"port out of range: {port}")
    return port


inputs = ["8080", "http", "443", "70000"]
ok, failed = [], []
with ThreadPoolExecutor(max_workers=4) as pool:
    futures = {pool.submit(parse_port, text): text for text in inputs}
    for future in as_completed(futures):
        try:
            ok.append(future.result())
        except ValueError as error:
            failed.append(f"{futures[future]!r}: {error}")

print("ok:", sorted(ok))
for line in sorted(failed):
    print("failed", line)

ok: [443, 8080]
failed '70000': port out of range: 70000
failed 'http': invalid literal for int() with base 10: 'http'

The dictionary maps each future back to its input, a common idiom. An exception in a worker never disappears: it waits inside the future until you call result().

Sharing an iterator

3.15+Requires version 3.15 or later A generator must not be advanced by two threads at once. Python 3.15 adds threading.serialize_iterator, which wraps any iterator so that each item goes to exactly one caller:

import threading


def ticket_numbers(count):
    for number in range(1, count + 1):
        yield number


tickets = threading.serialize_iterator(ticket_numbers(300))
taken = {name: [] for name in ("ann", "bob", "cy")}


def take_tickets(name):
    for number in tickets:
        taken[name].append(number)


threads = [threading.Thread(target=take_tickets, args=(n,)) for n in taken]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

everything = sorted(n for numbers in taken.values() for n in numbers)
print(len(everything), everything == list(range(1, 301)))

300 True

How the 300 tickets are split between the threads changes from run to run. That none is lost or duplicated does not. The same release added the @threading.synchronized_iterator decorator and threading.concurrent_tee(). On older versions, use a queue.

When not to use threads

Threads share everything, so every shared object needs a reason to be safe. For CPU-bound pure Python on the default build they give no speed-up: use processes or interpreters. For thousands of concurrent connections, asyncio is cheaper than thousands of threads. And when a job is fast already, a thread only adds a place for bugs. Threads are the right tool for a handful to a few dozen blocking calls, through a ThreadPoolExecutor, with data passed in and results passed back.

Common pitfalls

Assuming the GIL makes your code thread-safe. It makes single bytecode steps safe, not your multi-step logic. x += 1, check-then-act and read-modify-write all need a lock.

Locking some accesses but not others. Every reader and writer of shared state must use the same lock.

Swallowed exceptions. An exception in a plain Thread prints a traceback and the thread ends, but the main thread carries on. With a pool, call result() so errors reach you.

Using threads for CPU-bound work. On the default build they take turns. Use processes instead.

  45  https://example.com/d
  80  https://example.com/b
 120  https://example.com/a
 300  https://example.com/c
 545  total

import time
from concurrent.futures import ThreadPoolExecutor

FAKE_WEB = {
    "https://example.com/a": b"A" * 120,
    "https://example.com/b": b"B" * 80,
    "https://example.com/c": b"C" * 300,
    "https://example.com/d": b"D" * 45,
}


def download(url):
    time.sleep(0.2)  # every request takes 0.2 s
    return url, len(FAKE_WEB[url])


with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(download, FAKE_WEB))

for url, size in sorted(results, key=lambda pair: pair[1]):
    print(f"{size:4}  {url}")
print(f"{sum(size for _, size in results):4}  total")

  45  https://example.com/d
  80  https://example.com/b
 120  https://example.com/a
 300  https://example.com/c
 545  total

import threading

seats_left = 1
sold = []
both_checked = threading.Barrier(2)


def buy(customer):
    global seats_left
    if seats_left > 0:
        both_checked.wait()  # simulate an unlucky thread switch
        seats_left -= 1
        sold.append(customer)


buyers = [threading.Thread(target=buy, args=(c,)) for c in ("ann", "bob")]
for t in buyers:
    t.start()
for t in buyers:
    t.join()

print(f"tickets sold: {len(sold)}, seats left: {seats_left}")

tickets sold: 2, seats left: -1

tickets sold: 1, seats left: 0

import threading
import time

seats_left = 1
sold = []
seats_lock = threading.Lock()


def buy(customer):
    global seats_left
    with seats_lock:
        if seats_left > 0:
            time.sleep(0.05)  # a slow step can no longer hurt us
            seats_left -= 1
            sold.append(customer)


buyers = [threading.Thread(target=buy, args=(c,)) for c in ("ann", "bob")]
for t in buyers:
    t.start()
for t in buyers:
    t.join()

print(f"tickets sold: {len(sold)}, seats left: {seats_left}")

tickets sold: 1, seats left: 0

from concurrent.futures import ThreadPoolExecutor


def inverse(x):
    return 1 / x


with ThreadPoolExecutor(max_workers=4) as pool:
    try:
        for result in pool.map(inverse, [4, 2, 0, 1]):
            print(result)
    except ZeroDivisionError as error:
        print("stopped:", error)

0.25
0.5
stopped: division by zero

Summary

• Thread(target=..., args=...), then start() all threads, then join() them all.
• A race condition is a result that depends on how threads interleave. Read-modify-write and check-then-act are the usual culprits, and the GIL does not prevent them.
• A Lock used as with lock: makes a block a critical section. Every access to the shared state must take it.
• RLock can be taken again by the thread that holds it. Lock cannot.
• queue.Queue passes work between threads without shared state. Send one sentinel per worker to stop them.
• Event signals readiness or a stop, BoundedSemaphore caps concurrency, and Condition waits for a state.
• ThreadPoolExecutor.map returns results in input order. submit with as_completed gives results as they finish, and result() re-raises a worker’s exception.
• 3.15+Requires version 3.15 or later threading.serialize_iterator lets several threads share one iterator safely.
• Next: processes, for CPU-bound work.