21 — Concurrency: Threading & Multiprocessing

# ── Production: benchmarking I/O-bound vs CPU-bound across threading/multiprocessing ──
# Demonstrates the GIL's impact with real timing, not theoretical claims.

import threading
import multiprocessing as mp
import time
import requests
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

# ── I/O-bound: threading WINS (GIL released during network waits) ──
def fetch_url(url: str) -> tuple[str, int]:
    """I/O-bound task — releases GIL during socket recv, allowing true concurrency."""
    try:
        resp = requests.get(url, timeout=5)
        return url, resp.status_code
    except Exception as e:
        return url, -1

def io_bound_benchmark(urls: list[str]):
    # Sequential — one at a time, total = sum of all latencies
    t0 = time.perf_counter()
    for url in urls:
        fetch_url(url)
    sequential = time.perf_counter() - t0

    # Threaded — overlapping I/O waits, total ≈ slowest single request
    t0 = time.perf_counter()
    with ThreadPoolExecutor(max_workers=10) as pool:
        list(pool.map(fetch_url, urls))
    threaded = time.perf_counter() - t0

    print(f"I/O-bound: sequential={sequential:.2f}s, threaded={threaded:.2f}s")
    print(f"  speedup: {sequential/threaded:.1f}x — threading helps (GIL released during I/O)")

# ── CPU-bound: multiprocessing WINS (GIL blocks true parallelism in threads) ──
def cpu_intensive(n: int) -> int:
    """Pure Python computation — GIL prevents parallel bytecode execution across threads."""
    total = 0
    for i in range(n):
        total += i * i
    return total

def cpu_bound_benchmark(n: int, workers: int):
    tasks = [n] * workers

    # Sequential
    t0 = time.perf_counter()
    for task in tasks:
        cpu_intensive(task)
    sequential = time.perf_counter() - t0

    # Threaded — NO speedup (GIL serializes bytecode execution)
    t0 = time.perf_counter()
    with ThreadPoolExecutor(max_workers=workers) as pool:
        list(pool.map(cpu_intensive, tasks))
    threaded = time.perf_counter() - t0

    # Multiprocessing — TRUE parallelism (separate interpreters, separate GILs)
    t0 = time.perf_counter()
    with ProcessPoolExecutor(max_workers=workers) as pool:
        list(pool.map(cpu_intensive, tasks))
    multiprocess = time.perf_counter() - t0

    print(f"CPU-bound: sequential={sequential:.2f}s, threaded={threaded:.2f}s, multiprocess={multiprocess:.2f}s")
    print(f"  threading speedup: {sequential/threaded:.1f}x (≈1.0 — GIL blocks parallelism)")
    print(f"  multiprocessing speedup: {sequential/multiprocess:.1f}x (≈workers — true parallelism)")

if __name__ == "__main__":   # REQUIRED for ProcessPoolExecutor (spawn re-imports module)
    cpu_bound_benchmark(n=5_000_000, workers=4)

# ── Race condition: demonstrating the lost-update problem ──
# counter += 1 is NOT atomic — it's LOAD + ADD + STORE, interruptible between any step.

import threading

counter = 0
lock = threading.Lock()

def increment_unsafe(n: int):
    """RACE CONDITION: counter += 1 compiles to multiple bytecodes, GIL can switch mid-way."""
    global counter
    for _ in range(n):
        counter += 1   # LOAD_FAST counter; LOAD_CONST 1; BINARY_OP +=; STORE_FAST counter
                       # GIL can release between LOAD and STORE → two threads read same old value

def increment_safe(n: int, lock: threading.Lock):
    """CORRECT: lock serializes the read-modify-write, no interleaving possible."""
    global counter
    for _ in range(n):
        with lock:           # acquire on entry, release on exit (even on exception)
            counter += 1     # only one thread can be inside this block at a time

# Demonstrate the race
counter = 0
threads = [threading.Thread(target=increment_unsafe, args=(100_000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Unsafe (expected 400000): {counter}")   # e.g. 387,341 — lost updates!

# Demonstrate the fix
counter = 0
lock = threading.Lock()
threads = [threading.Thread(target=increment_safe, args=(100_000, lock)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Safe (expected 400000):   {counter}")   # 400000 — always correct

import threading

counter = 0

def increment_unsafe():
    global counter
    for _ in range(100_000):
        counter += 1     # NOT atomic! read-modify-write across three bytecode ops, interruptible mid-way

threads = [threading.Thread(target=increment_unsafe) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)   # expected 400,000 — often prints something LESS, e.g. 391,842 — a lost-update race condition

import threading

counter = 0
lock = threading.Lock()

def increment_safe():
    global counter
    for _ in range(100_000):
        with lock:            # only ONE thread can hold the lock at a time
            counter += 1

threads = [threading.Thread(target=increment_safe) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)   # always exactly 400,000

import threading

# RLock — reentrant: the SAME thread can acquire it multiple times without deadlocking itself
rlock = threading.RLock()

def outer():
    with rlock:
        inner()             # inner() also acquires rlock — fine, same thread, RLock tracks a counter

def inner():
    with rlock:
        print("inner")

# A plain Lock here would deadlock: the second `with lock` in the same thread would block forever
# waiting for a lock the SAME thread already holds and will never release until this call returns.

import threading
import time

ready = threading.Event()

def waiter():
    print("waiting for signal...")
    ready.wait()            # blocks until .set() is called from any thread
    print("signal received, proceeding")

def setter():
    time.sleep(1)
    ready.set()

threading.Thread(target=waiter).start()
threading.Thread(target=setter).start()

import threading
import queue

task_queue = queue.Queue()

def producer():
    for i in range(5):
        task_queue.put(i)
    task_queue.put(None)      # sentinel value signaling "no more work"

def consumer():
    while (item := task_queue.get()) is not None:
        print(f"processing {item}")
        task_queue.task_done()
    task_queue.task_done()

threading.Thread(target=producer).start()
threading.Thread(target=consumer).start()

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

urls = ["https://example.com"] * 5

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(requests.get, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            response = future.result()
            print(f"{url}: {response.status_code}")
        except Exception as exc:
            print(f"{url} raised {exc!r}")

from multiprocessing import Process, Pool
import os

def cpu_bound_work(n):
    total = 0
    for i in range(n):
        total += i
    return total

if __name__ == "__main__":     # REQUIRED on Windows/macOS spawn start method — see gotcha below
    with Pool(processes=4) as pool:
        results = pool.map(cpu_bound_work, [50_000_000] * 4)
    print(results)   # genuinely runs on 4 cores in parallel — no GIL contention, separate interpreters

from multiprocessing import Process, Value, Array, Manager

def worker(shared_counter, lock):
    with lock:
        shared_counter.value += 1

if __name__ == "__main__":
    from multiprocessing import Lock
    counter = Value("i", 0)          # 'i' = C int, allocated in SHARED memory across processes
    lock = Lock()
    processes = [Process(target=worker, args=(counter, lock)) for _ in range(10)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(counter.value)   # 10 — Value/Array are the explicit mechanisms for sharing primitive state

# WRONG on Windows and macOS (default 'spawn' start method) — causes infinite recursive process spawning
from multiprocessing import Process

def worker():
    print("working")

p = Process(target=worker)   # module-level code, NOT guarded
p.start()
# When spawn re-imports this module in the child process, THIS LINE RUNS AGAIN,
# spawning another child, which re-imports and spawns another... (RuntimeError, usually caught by a guard)

from multiprocessing import Process

def worker():
    print("working")

if __name__ == "__main__":    # guards module-level process-spawning code — REQUIRED, not just tidy style
    p = Process(target=worker)
    p.start()
    p.join()

from concurrent.futures import ProcessPoolExecutor

def is_prime(n):
    if n < 2:
        return False
    return all(n % i for i in range(2, int(n ** 0.5) + 1))

if __name__ == "__main__":
    numbers = list(range(100_000, 100_020))
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(is_prime, numbers))
    print(list(zip(numbers, results)))

import threading

results = []

def fetch_and_store(item_id):
    data = f"result-{item_id}"
    if len(results) < 10:
        results.append(data)

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

print(len(results))   # sometimes 10, sometimes less