Threads, Processes & the GIL¶

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time, requests

# ── ThreadPoolExecutor: I/O-bound work ────────────────────────────────────────
urls = [f"https://httpbin.org/delay/1?n={i}" for i in range(10)]

def download(url: str) -> str:
    """Simulate a blocking HTTP GET."""
    time.sleep(1)   # in production: requests.get(url).text
    return f"content of {url[-1]}"

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=10) as pool:
    results = list(pool.map(download, urls))
print(f"Threads: {time.perf_counter()-start:.1f}s")   # ~1.0s (parallel I/O waits)

# Sequential would take ~10s
# Threads overlap the waiting (GIL released during sleep/I/O)

# ── ProcessPoolExecutor: CPU-bound work ──────────────────────────────────────
def is_prime(n: int) -> bool:
    """CPU-intensive check."""
    if n < 2: return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0: return False
    return True

numbers = list(range(10**6, 10**6 + 100))   # 100 numbers near 1 million

start = time.perf_counter()
with ProcessPoolExecutor(max_workers=4) as pool:  # 4 real CPU cores
    primes = [n for n, ok in zip(numbers, pool.map(is_prime, numbers)) if ok]
elapsed = time.perf_counter() - start
print(f"Processes: {elapsed:.2f}s  found {len(primes)} primes")

# With threads this would be ~4x slower (GIL prevents parallel CPU work)
# With processes: real parallelism across 4 cores

# ── submit() for individual tasks + futures ────────────────────────────────
with ThreadPoolExecutor(max_workers=5) as pool:
    futures = {pool.submit(download, url): url for url in urls[:5]}
    for future in futures:
        url = futures[future]
        try:
            result = future.result(timeout=5)
        except Exception as e:
            print(f"  {url} failed: {e}")

import threading
from collections import defaultdict
from queue import Queue

# ── Race condition — counter += 1 is NOT atomic ──────────────────────────────
counter = 0
def increment(n):
    global counter
    for _ in range(n):
        counter += 1    # read, add, write — 3 ops, interruptable!

threads = [threading.Thread(target=increment, args=(10_000,)) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
# counter may be < 100_000 due to lost updates (race condition)

# ── Fix with a Lock ───────────────────────────────────────────────────────────
counter = 0
lock = threading.Lock()

def safe_increment(n):
    global counter
    for _ in range(n):
        with lock:          # acquires lock, releases on exit (even on exception)
            counter += 1    # now atomic

threads = [threading.Thread(target=safe_increment, args=(10_000,)) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
# counter is exactly 100_000

# ── Thread-safe producer/consumer with Queue ──────────────────────────────────
# queue.Queue is designed for thread communication — fully thread-safe
job_queue = Queue()

def producer():
    for i in range(20):
        job_queue.put(f"job_{i}")
    job_queue.put(None)   # sentinel to signal done

def consumer():
    while True:
        job = job_queue.get()
        if job is None:
            break
        print(f"Processing {job}")
        job_queue.task_done()

prod_thread = threading.Thread(target=producer)
cons_thread = threading.Thread(target=consumer)
prod_thread.start(); cons_thread.start()
prod_thread.join();  cons_thread.join()

# ── threading.local() — per-thread storage ────────────────────────────────────
local_data = threading.local()
def worker():
    local_data.value = threading.current_thread().name  # each thread has own .value
    print(f"{local_data.value}: {local_data.value}")