Threads — Python Threading and Concurrency

S8.0 What This Teaches

• Creating and joining threads with the threading module
• Daemon threads and thread lifecycle
• Race conditions and shared mutable state
• Lock and RLock for mutual exclusion
• Thread-safe communication with queue.Queue
• The Global Interpreter Lock (GIL) and when threads help
• When to prefer multiprocessing or concurrent.futures

S8.1 Creating Threads

# Threads - creating and joining threads.
import threading
import time

def worker(name: str, delay: float):
    time.sleep(delay)
    print(f"{name} done after {delay}s")

t1 = threading.Thread(target=worker, args=("A", 0.5))
t2 = threading.Thread(target=worker, args=("B", 0.2))

t1.start()
t2.start()

t1.join()
t2.join()
print("all done")

S8.2 Daemon Threads

import threading, time

def heartbeat():
    while True:
        print("ping")
        time.sleep(1)

monitor = threading.Thread(target=heartbeat, daemon=True)
monitor.start()

time.sleep(2.5)
print("main exiting")   # daemon thread dies here automatically

S8.3 Race Conditions

import threading

counter = 0

def increment(n: int):
    global counter
    for _ in range(n):
        counter += 1   # read-modify-write is NOT atomic

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)   # expected 500_000, but often less due to lost updates

S8.4 Lock and RLock

import threading

counter = 0
lock = threading.Lock()

def safe_increment(n: int):
    global counter
    for _ in range(n):
        with lock:         # acquire on enter, release on exit
            counter += 1

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

print(counter)   # always 500_000

S8.5 Thread-Safe Queue

import threading
import queue

def producer(q: queue.Queue, items: list):
    for item in items:
        q.put(item)
    q.put(None)   # sentinel to signal completion

def consumer(q: queue.Queue):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"processed {item}")
        q.task_done()

q = queue.Queue()
p = threading.Thread(target=producer, args=(q, [1, 2, 3, 4, 5]))
c = threading.Thread(target=consumer, args=(q,))

p.start()
c.start()
p.join()
c.join()

S8.6 The Global Interpreter Lock

• Threads do NOT run in parallel for CPU-bound work (no true parallelism)
• Threads DO run concurrently for I/O-bound work - I/O releases the GIL
• Use multiprocessing or concurrent.futures.ProcessPoolExecutor for CPU-bound parallelism

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def cpu_work(n):
    return sum(range(n))

# ThreadPoolExecutor - good for I/O-bound tasks
with ThreadPoolExecutor(max_workers=4) as ex:
    results = list(ex.map(lambda url: fetch(url), urls))

# ProcessPoolExecutor - good for CPU-bound tasks (bypasses GIL)
with ProcessPoolExecutor(max_workers=4) as ex:
    results = list(ex.map(cpu_work, [10_000_000] * 4))

S8.7 Example - All Together

# Threads - worker pool with a Queue and Lock for safe result collection.

import threading
import queue

results = []
results_lock = threading.Lock()

def worker(task_queue: queue.Queue):
    while True:
        item = task_queue.get()
        if item is None:
            task_queue.put(None)   # pass sentinel to next worker
            break
        result = item * item
        with results_lock:
            results.append(result)
        task_queue.task_done()

tasks = queue.Queue()
for n in range(1, 11):
    tasks.put(n)
tasks.put(None)   # one sentinel triggers chain shutdown

workers = [threading.Thread(target=worker, args=(tasks,)) for _ in range(3)]
for w in workers: w.start()
for w in workers: w.join()

print(sorted(results))   # [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

S8.8 Exercise

S8.9 Common Mistakes

t = threading.Thread(target=worker)
t.start()
print("done")   # main exits before worker finishes!

# Fix: always join threads you care about
t.start()
t.join()
print("done")

# Dangerous: sharing a list without locking
shared = []
def bad_append(x): shared.append(x)   # list.append is thread-safe by accident; dict access is not

# Correct: use Queue
q = queue.Queue()
def safe_put(x): q.put(x)

S8.10 Key Terms