Python multithreading: threads, locks, semaphores, events and queues

A thread is a separate flow of execution inside one program. With several threads, a program can wait for a download, read a file and answer a user at the same time, instead of doing those one after another. Python's threading module makes that easy to start and surprisingly easy to get wrong. This post explains how threads work, the problems they create (races and deadlocks), and the tools Python provides to control them. All outputs come from running the code on a real interpreter (Python 3.9). Outputs that depend on timing are described rather than quoted.

Why threads, and the GIL

A program often spends time waiting: for the network, a disk, a database, a sleep. A thread that is waiting does not use the processor, so other threads can run. That is where threads help: I/O-bound work.

Python's standard interpreter has a Global Interpreter Lock (GIL): only one thread runs Python bytecode at a time. So threads do not speed up CPU-heavy calculations such as number crunching. For that, use the multiprocessing module or concurrent.futures.ProcessPoolExecutor, which use separate processes. Rule of thumb: threads for waiting, processes for computing.

Proof that threads overlap waiting:

import threading, time
t0 = time.perf_counter()
ts = [threading.Thread(target=time.sleep, args=(0.2,)) for _ in range(5)]
[t.start() for t in ts]; [t.join() for t in ts]
print(time.perf_counter() - t0 < 0.5)        # True: five 0.2 s sleeps took about 0.2 s, not 1 s

Doing the same five sleeps one after another took at least a second.

Creating and joining threads

import threading

def work(n, out):
    out.append(n * n)

out = []
ts = [threading.Thread(target=work, args=(i, out)) for i in range(5)]
for t in ts: t.start()          # begin running
for t in ts: t.join()           # wait until each has finished
print(sorted(out))              # [0, 1, 4, 9, 16]

• Thread(target=function, args=(...)) creates the thread; nothing runs until start().
• join() makes the caller wait for the thread to finish. Without it, the main program might finish first (non-daemon threads keep the program alive anyway, but you would read the results too early).
• Call start() only once per thread. A function you pass runs in the new thread; its return value is discarded, so share results through a list, a queue, or use a thread pool.
• The order in which threads run is not guaranteed. That is why the example sorts the output.

Identifying a thread

t = threading.Thread(target=lambda: None, name="worker-1")
print(t.name, t.is_alive(), t.ident, t.daemon)     # worker-1 False None False
t.start(); t.join()
print(t.is_alive(), t.ident is not None)            # False True
print(threading.current_thread().name, threading.main_thread() is threading.current_thread())
# MainThread True

ident is a number the system gives the thread once it starts (None before). threading.current_thread() tells you which thread runs your code, which is helpful in logs.

Exceptions in threads

An unhandled exception in a thread prints a traceback to the error stream and ends only that thread; the main program carries on. Catch exceptions inside the thread function, or collect them, if you need to react.

Daemon threads

A daemon thread is a background helper that does not keep the program alive. When all non-daemon threads finish, the program exits and daemon threads are stopped abruptly.

import time
def forever():
    while True: time.sleep(0.01)
d = threading.Thread(target=forever, daemon=True)
d.start()
print("daemon:", d.daemon, d.is_alive())      # daemon: True True

Use daemon threads for housekeeping that can be dropped at exit (a heartbeat, a cache cleaner), never for work that must complete or release resources, such as writing a file.

Race conditions

When several threads read and change the same variable, results can be wrong because a thread can be paused between reading and writing. This is a race condition:

counter = 0
def inc_unsafe(n):
    global counter
    for _ in range(n):
        v = counter; time.sleep(0); counter = v + 1    # read, give other threads a turn, write
ts = [threading.Thread(target=inc_unsafe, args=(2000,)) for _ in range(4)]
[t.start() for t in ts]; [t.join() for t in ts]
print(counter)      # far less than 8000 (in the test run it was 2000); the number differs from run to run

Four threads each added 2,000 times, so the answer should be 8,000. Many updates were lost because threads overwrote each other's results. (The sleep(0) makes the problem easy to see; without it, the bug is still there but appears rarely, which is worse.)

Lock: one thread at a time

A Lock lets only one thread into a protected section. Use it with with, so it is always released:

counter = 0
lock = threading.Lock()
def inc_safe(n):
    global counter
    for _ in range(n):
        with lock:
            v = counter; time.sleep(0); counter = v + 1
ts = [threading.Thread(target=inc_safe, args=(2000,)) for _ in range(4)]
[t.start() for t in ts]; [t.join() for t in ts]
print(counter)      # 8000

Keep the locked section short, and never hold a lock while waiting for something slow. A plain Lock is not re-entrant: a thread that tries to acquire a lock it already holds waits for itself forever (a deadlock).

l = threading.Lock()
l.acquire()
print(l.acquire(blocking=False))      # False: it would have blocked
l.release()

RLock: re-entrant lock

An RLock can be acquired again by the thread that already holds it, as long as it is released the same number of times. It suits recursive code and methods that call each other under the same lock.

r = threading.RLock()
def nested():
    with r:
        with r:
            return "rlock ok"
print(nested())        # rlock ok

Difference in one line: Lock — one acquire at a time, even for the same thread; RLock — the owning thread may acquire repeatedly.

A classic deadlock: thread 1 holds lock A and waits for B while thread 2 holds B and waits for A. Avoid it by always acquiring multiple locks in the same order, or by using timeouts.

Semaphore: limit concurrency

A Semaphore(n) lets up to n threads in at once. It suits a limited resource, such as 3 database connections.

sem = threading.BoundedSemaphore(2)
active = peak = 0
glock = threading.Lock()
def use():
    global active, peak
    with sem:
        with glock:
            active += 1; peak = max(peak, active)
        time.sleep(0.02)
        with glock:
            active -= 1
ts = [threading.Thread(target=use) for _ in range(6)]
[t.start() for t in ts]; [t.join() for t in ts]
print("peak concurrent:", peak)       # peak concurrent: 2

Six threads ran, never more than two at once. A BoundedSemaphore also catches a bug that a plain Semaphore hides, releasing more times than acquired:

try:
    threading.BoundedSemaphore(1).release()
except ValueError as e:
    print(e)        # Semaphore released too many times

Event: a signal between threads

An Event is a flag one thread sets and others wait for.

ev = threading.Event(); res = []
def waiter():
    ev.wait(); res.append("go")
t = threading.Thread(target=waiter); t.start()
time.sleep(0.02)
print(res, ev.is_set())     # [] False   (still waiting)
ev.set(); t.join()
print(res, ev.is_set())     # ['go'] True

Condition: wait for something to become true

A Condition combines a lock with waiting and notifying. A consumer waits until a producer adds something:

cond = threading.Condition(); items = []
def consumer():
    with cond:
        while not items: cond.wait()
        print("consumed", items.pop())
def producer():
    time.sleep(0.02)
    with cond:
        items.append("x"); cond.notify()
a = threading.Thread(target=consumer); b = threading.Thread(target=producer)
a.start(); b.start(); a.join(); b.join()       # consumed x

Always wait in a while loop that rechecks the condition.

Queue: the safest way to share data

queue.Queue is already thread-safe, so a producer and consumer need no locks of their own:

import queue
q = queue.Queue(); got = []
def prod():
    for i in range(5): q.put(i)
    q.put(None)                      # a sentinel meaning "no more work"
def cons():
    while True:
        x = q.get()
        if x is None: break
        got.append(x * 10)
a = threading.Thread(target=prod); b = threading.Thread(target=cons)
a.start(); b.start(); a.join(); b.join()
print(got)         # [0, 10, 20, 30, 40]

Prefer queues to shared variables whenever you can: pass messages instead of sharing state.

Thread pools

Creating threads by hand gets tedious. ThreadPoolExecutor manages a pool and gives results back:

from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=3) as ex:
    print(list(ex.map(lambda x: x * x, range(6))))      # [0, 1, 4, 9, 16, 25]

submit() returns a future whose .result() waits for the value, and re-raises any exception from the worker, which handles error reporting neatly. A threading.Timer(0.05, func) runs a function once after a delay.

Good practice with locks

1. Use with lock: rather than manual acquire/release.
2. Keep critical sections small.
3. Don't call unknown code while holding a lock.
4. Lock in a fixed order if you need several.
5. Prefer queues, immutable data and thread pools over shared mutable variables.
6. Make threads stoppable with an Event rather than killing them.
7. Test with more threads and bigger loops than you expect; races hide in small tests.

Common mistakes

• Expecting threads to speed up calculations (the GIL).
• Reading results before join().
• Passing target=work() instead of target=work (that calls the function immediately).
• Sharing a variable without a lock.
• Forgetting the comma in args=(x,) for a one-argument function.
• Holding a lock during slow I/O.

Watch the lessons

Watch the video lessons: Episode 74: Multithreading, Part 1, Episode 80: Multi-Tasking and Episode 81: Synchronisation. Related reading: logging is thread-safe and useful for tracing threads, and the Python cheat sheet covers the language basics used here.