Threads

A thread is the smallest unit of scheduling in an operating system. Multiple threads within the same process share that process’s memory space. Python provides thread support through the threading module.

Relationship Between Processes and Threads

Process: The basic unit of resource allocation, with its own independent memory space.
Thread: The basic unit of execution and scheduling, more lightweight than a process, with lower creation and context-switching overhead.
Every process has at least one main thread and can create multiple child threads. All threads within the same process share memory, making communication convenient but requiring synchronization control.

Python’s GIL

CPython has a Global Interpreter Lock (GIL) that allows only one thread to execute Python bytecode at a time. This means:

Multiple threads cannot achieve true parallel computation (CPU-bound tasks) using multiple CPU cores.
However, multiple threads can achieve concurrency: when one thread is waiting for I/O, other threads can run (I/O-bound tasks).
CPU-bound tasks should use multiprocessing (multiple processes) or concurrent.futures.ProcessPoolExecutor instead.

Python 3.13 introduced an experimental GIL-free mode (--disable-gil). True multi-core parallelism may become available in the future.

Creating Threads

import threading
import time

def worker(name: str, delay: float) -> None:
    print(f"[{name}] Started")
    time.sleep(delay)
    print(f"[{name}] Finished")

# Method 1: Thread object
t1 = threading.Thread(target=worker, args=("Thread A", 1))
t2 = threading.Thread(target=worker, args=("Thread B", 2))

t1.start()
t2.start()

# join() waits for the thread to finish
t1.join()
t2.join()
print("All threads finished")

Subclassing Thread

import threading

class DownloadTask(threading.Thread):
    def __init__(self, url: str):
        super().__init__()
        self.url = url

    def run(self) -> None:
        """This method is executed when the thread starts."""
        import time
        print(f"Downloading: {self.url}")
        time.sleep(1)
        print(f"Completed: {self.url}")

tasks = [
    DownloadTask("https://example.com/file1.zip"),
    DownloadTask("https://example.com/file2.zip"),
]

for t in tasks:
    t.start()

for t in tasks:
    t.join()

Daemon Threads

A daemon thread exits automatically when the main thread ends, and does not prevent the program from exiting. Suitable for background tasks (log reporting, heartbeat detection, etc.):

import threading
import time

def heartbeat() -> None:
    while True:
        print("♥ Heartbeat")
        time.sleep(2)

# Set as a daemon thread; it terminates automatically when the main thread ends
t = threading.Thread(target=heartbeat, daemon=True)
t.start()

time.sleep(5)
print("Main thread ending, daemon thread will also exit")

Thread Safety: Lock

When multiple threads modify shared variables, a lock (Lock) must be used to prevent race conditions:

import threading

counter = 0
lock = threading.Lock()

def increment(n: int) -> None:
    global counter
    for _ in range(n):
        with lock:   # Equivalent to lock.acquire() ... lock.release()
            counter += 1

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

print(f"Final count: {counter}")   # 50000 (correct result)

Without a lock, counter may end up less than 50000 (lost updates).

RLock — Reentrant Lock

The same thread can acquire an RLock multiple times, avoiding deadlocks:

import threading

rlock = threading.RLock()

def outer():
    with rlock:
        print("Outer lock acquired")
        inner()   # Same thread acquires the lock again; RLock allows this, Lock would deadlock

def inner():
    with rlock:
        print("Inner lock acquired")

t = threading.Thread(target=outer)
t.start()
t.join()

Semaphore

Controls the maximum number of threads that can access a resource simultaneously (e.g., connection pools, concurrent request limits):

import threading
import time

# Allow at most 3 threads in at once
sem = threading.Semaphore(3)

def access_resource(name: str) -> None:
    with sem:
        print(f"{name} is using the resource")
        time.sleep(1)
        print(f"{name} released the resource")

threads = [threading.Thread(target=access_resource, args=(f"Thread {i}",)) for i in range(8)]
for t in threads:
    t.start()
for t in threads:
    t.join()

Event — Thread Signaling

import threading
import time

event = threading.Event()

def producer() -> None:
    print("Producer: preparing data...")
    time.sleep(2)
    event.set()          # Send signal
    print("Producer: data ready")

def consumer() -> None:
    print("Consumer: waiting for data...")
    event.wait()         # Block until signal is received
    print("Consumer: starting to process data")

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t2.start()
t1.start()
t1.join()
t2.join()

Queue — Thread-Safe Queue

queue.Queue has built-in locking and is the preferred way to safely pass data between threads:

import threading
import queue
import time

task_queue: queue.Queue = queue.Queue(maxsize=10)

def producer(q: queue.Queue, n: int) -> None:
    for i in range(n):
        item = f"Task {i}"
        q.put(item)   # Automatically blocks when the queue is full
        print(f"Produced: {item}")
        time.sleep(0.1)
    q.put(None)       # Send termination signal

def consumer(q: queue.Queue) -> None:
    while True:
        item = q.get()  # Automatically blocks when the queue is empty
        if item is None:
            break
        print(f"Consumed: {item}")
        q.task_done()

t1 = threading.Thread(target=producer, args=(task_queue, 5))
t2 = threading.Thread(target=consumer, args=(task_queue,))
t1.start()
t2.start()
t1.join()
t2.join()

ThreadPoolExecutor — Thread Pool (Recommended)

concurrent.futures.ThreadPoolExecutor is the higher-level, easier-to-use thread pool interface:

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch(url: str) -> str:
    time.sleep(0.5)   # Simulate I/O
    return f"{url} → 200 OK"

urls = [f"https://api.example.com/item/{i}" for i in range(10)]

with ThreadPoolExecutor(max_workers=4) as executor:
    # map: preserves order, concise
    results = list(executor.map(fetch, urls))
    print(results)

# submit + as_completed: process whichever finishes first
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result()
            print(result)
        except Exception as e:
            print(f"{url} failed: {e}")

Thread-Local Storage (threading.local)

Each thread has its own independent copy that does not interfere with others:

import threading

local_data = threading.local()

def worker(value: int) -> None:
    local_data.value = value   # Each thread stores independently
    import time
    time.sleep(0.1)
    print(f"Thread {threading.current_thread().name}: {local_data.value}")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

Last updated on 2026-06-23

Processes Coroutines