Threading

Creating threads

import threading
import time


def worker(name, seconds):
    print(f"{name} starting")
    time.sleep(seconds)                  # stands in for waiting on a network or disk
    print(f"{name} finished after {seconds}s")


start = time.perf_counter()

threads = [
    threading.Thread(target=worker, args=("A", 2)),
    threading.Thread(target=worker, args=("B", 2)),
    threading.Thread(target=worker, args=("C", 2)),
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()                        # wait for it to finish

print(f"total {time.perf_counter() - start:.2f}s")

A starting
B starting
C starting
A finished after 2s
B finished after 2s
C finished after 2s
total 2.00s

Six seconds of waiting completed in two, because all three threads waited at the same time. That is what threads are for.

The essential methods

import threading
import time


def task():
    time.sleep(0.5)


t = threading.Thread(target=task, name="worker-1", daemon=False)

print(t.is_alive())          # False - not started
t.start()                     # begin running
print(t.is_alive())          # True
print(t.name, t.daemon)
t.join(timeout=2)             # wait, with an optional limit
print(t.is_alive())           # False

print(threading.current_thread().name)     # MainThread
print(threading.active_count())

• start() begins the thread. Calling run() directly just runs the function in the current thread.
• join() blocks until the thread ends.
• A daemon thread does not keep the program alive; it is killed abruptly at exit.
• A thread cannot be restarted once it has finished.

A thread subclass

import threading
import time


class Downloader(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None
        self.error = None

    def run(self):
        try:
            time.sleep(0.5)                     # pretend to fetch
            self.result = f"content of {self.url}"
        except Exception as error:              # a thread must catch its own
            self.error = error


workers = [Downloader(f"page-{i}") for i in range(3)]

for w in workers:
    w.start()
for w in workers:
    w.join()

for w in workers:
    print(w.result if w.error is None else f"failed: {w.error}")

An exception inside a thread does not reach the main thread. It prints a traceback and the thread dies quietly. Always catch exceptions inside the thread and store them, or use ThreadPoolExecutor, which re-raises them for you.

Race conditions

import threading

counter = 0


def increment(times):
    global counter
    for _ in range(times):
        counter += 1          # read, add, write - three separate steps


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

print(counter)                # usually less than 500000

counter += 1 is not one operation. A thread can be interrupted between reading the value and writing the new one, so two threads read the same number and both write the same result. One increment is lost.

Locks

import threading

counter = 0
lock = threading.Lock()


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


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

print(counter)                    # exactly 500000

import threading

lock = threading.Lock()

# Preferred: released even if the block raises
with lock:
    pass

# Manual equivalent
lock.acquire()
try:
    pass
finally:
    lock.release()

if lock.acquire(timeout=1):       # give up rather than block forever
    try:
        pass
    finally:
        lock.release()

Keep the critical section small

import threading

lock = threading.Lock()
results = []


def slow_computation(n):
    return n * n


def worker(n):
    value = slow_computation(n)       # do the work OUTSIDE the lock
    with lock:
        results.append(value)         # only the shared update is protected

Deadlock

import threading
import time

lock_a = threading.Lock()
lock_b = threading.Lock()


def task_one():
    with lock_a:
        time.sleep(0.1)
        with lock_b:                  # waits for B, which task_two holds
            print("task one done")


def task_two():
    with lock_b:
        time.sleep(0.1)
        with lock_a:                  # waits for A, which task_one holds
            print("task two done")


# Running both together can hang forever.
# The fix: always acquire locks in the SAME order everywhere.


def task_two_fixed():
    with lock_a:
        with lock_b:
            print("task two done")

Deadlock needs two threads each holding what the other wants. Acquiring locks in one consistent order across the whole program makes it impossible.

Other synchronisation tools

import threading
import time

# Event: one thread signals, others wait
ready = threading.Event()


def waiter():
    print("waiting for the signal")
    ready.wait()
    print("received")


def signaller():
    time.sleep(0.5)
    ready.set()


threading.Thread(target=waiter).start()
threading.Thread(target=signaller).start()
time.sleep(1)

import threading

# Semaphore: allow at most N threads at once
limit = threading.Semaphore(3)


def limited_task(n):
    with limit:
        print(f"task {n} running")


# RLock: the same thread may acquire it repeatedly
rlock = threading.RLock()


def outer():
    with rlock:
        inner()


def inner():
    with rlock:            # a plain Lock would deadlock here
        print("nested acquisition is fine")


outer()

Queues: the safer pattern

import queue
import threading
import time

work = queue.Queue()
results = queue.Queue()


def worker(name):
    while True:
        item = work.get()
        if item is None:                 # the shutdown signal
            work.task_done()
            break
        time.sleep(0.1)
        results.put((name, item, item * item))
        work.task_done()


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

for n in range(10):
    work.put(n)

work.join()                              # wait until every item is processed

for _ in workers:
    work.put(None)                       # stop each worker
for w in workers:
    w.join()

while not results.empty():
    print(results.get())

queue.Queue is thread safe on its own. Passing work through a queue instead of sharing variables removes most of the need for locks, and it is the pattern to reach for first.

ThreadPoolExecutor

import time
from concurrent.futures import ThreadPoolExecutor, as_completed


def fetch(name):
    time.sleep(0.5)
    if name == "bad":
        raise ValueError("could not fetch")
    return f"content of {name}"


names = ["a", "b", "bad", "c"]

with ThreadPoolExecutor(max_workers=4) as pool:
    futures = {pool.submit(fetch, name): name for name in names}

    for future in as_completed(futures):
        name = futures[future]
        try:
            print(f"{name}: {future.result()}")
        except Exception as error:
            print(f"{name}: failed - {error}")

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as pool:
    for result in pool.map(str.upper, ["a", "b", "c"]):
        print(result)                   # results in input order

The executor manages thread creation, reuse and shutdown, and it propagates exceptions when you call result(). Prefer it to creating threads by hand in almost all cases.

Thread local data

import threading

local = threading.local()


def worker(name):
    local.name = name                    # separate per thread
    print(f"{threading.current_thread().name}: {local.name}")


threading.Thread(target=worker, args=("first",)).start()
threading.Thread(target=worker, args=("second",)).start()

Where threads help and where they do not

import time
import threading


def io_bound(seconds):
    time.sleep(seconds)                  # releases the GIL while waiting


def cpu_bound(n):
    return sum(i * i for i in range(n))  # holds the GIL


def timed(label, target, args, count=4):
    start = time.perf_counter()
    threads = [threading.Thread(target=target, args=args) for _ in range(count)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"{label}: {time.perf_counter() - start:.2f}s")


timed("io  ", io_bound, (1,))            # about 1s - real gain
timed("cpu ", cpu_bound, (2_000_000,))   # no faster than running them one by one

The reason is the global interpreter lock, which the GIL note explains in full.

Common mistakes

• Calling run() instead of start(), so nothing runs concurrently.
• Forgetting join() and reading results before they exist.
• Sharing mutable state without a lock.
• Holding a lock while doing slow work.
• Acquiring locks in different orders in different places.
• Expecting an exception in a thread to reach the main thread.
• Using threads for CPU bound work.

Best practices

• Use ThreadPoolExecutor rather than creating threads by hand.
• Pass work through a queue.Queue instead of sharing variables.
• Protect every shared mutable object with a lock, and keep the critical section tiny.
• Acquire multiple locks in one consistent order.
• Catch exceptions inside the thread, or use futures.
• Use threads for waiting and processes for computing.

Practice

1. Download four simulated resources sequentially and then with threads, and compare the times.
2. Demonstrate a race condition on a shared counter, then fix it with a lock.
3. Build a producer and consumer pipeline using queue.Queue.
4. Use ThreadPoolExecutor with as_completed and handle a task that raises.
5. Create a deadlock with two locks, then fix it by ordering the acquisitions.

Conclusion

Threads let a program wait for many things at once, which is exactly what input and output need. Use an executor, pass work through queues, protect shared state with locks, and do not expect threads to make computation faster in Python.