Advanced Examples

"""Example 58: On a Free-Threaded Build, CPU-Bound Threads Actually Scale."""
 
import sys  # => co-04: `sys._is_gil_enabled()` tells us which kind of build is RUNNING right now
import threading  # => the exact same threading shape as ex-49 -- only the BUILD differs
import time  # => measures wall time to compute an empirical speedup ratio
 
ITERATIONS = 8_000_000  # => tuned so 4 threads' worth of work is clearly measurable, not instant
THREAD_COUNT = 4  # => how many threads race to do CPU work concurrently
 
 
def gil_is_enabled() -> bool:  # => the SAME version-gated check introduced in ex-04
    if hasattr(sys, "_is_gil_enabled"):  # => only exists on Python 3.13+
        return sys._is_gil_enabled()  # pyright: ignore[reportPrivateUsage]
        # => leading underscore is CPython's naming, not a privacy signal -- the documented 3.13+ API
    return True  # => pre-3.13: no free-threaded option ever existed, so the GIL was always enabled
 
 
def cpu_task(n: int) -> int:  # => pure CPU work -- no I/O, so the GIL (if enabled) can NEVER be released mid-loop
    total = 0  # => accumulator -- forces real interpreter bytecode execution
    for i in range(n):  # => a tight loop -- exactly the shape a GIL serializes across threads
        total += i  # => trivial arithmetic; only the TIME this takes matters here
    return total  # => the actual value is irrelevant to this example
 
 
def measure_threaded_speedup(iterations: int, thread_count: int) -> float:
    single_start = time.perf_counter()  # => single_start: wall time before the ONE-thread baseline unit
    cpu_task(iterations)  # => runs exactly ONE unit of work, alone, to establish a per-unit baseline
    single_time = time.perf_counter() - single_start  # => single_time: how long ONE unit takes, uncontended
 
    threads = [threading.Thread(target=cpu_task, args=(iterations,)) for _ in range(thread_count)]
    start = time.perf_counter()  # => start: wall time before the `thread_count`-way concurrent run
    for t in threads:  # => starts every thread, all targeting the SAME cpu_task
        t.start()  # => each begins its own `iterations`-long loop, "concurrently" (subject to the GIL)
    for t in threads:  # => waits for every thread to finish
        t.join()  # => join() blocks until that thread's cpu_task() call returns
    elapsed = time.perf_counter() - start  # => elapsed: wall time for ALL `thread_count` threads combined
 
    serial_equivalent = single_time * thread_count  # => serial_equivalent: what `thread_count` units WOULD cost run one-by-one
    return serial_equivalent / elapsed  # => speedup: how much faster the threaded run was vs doing it serially
 
 
if __name__ == "__main__":  # => module entry point
    gil_enabled = gil_is_enabled()  # => gil_enabled: is THIS running interpreter's GIL currently active?
    speedup = measure_threaded_speedup(ITERATIONS, THREAD_COUNT)  # => speedup: the empirically measured ratio
    print(f"gil_enabled={gil_enabled} speedup={speedup:.2f}x")  # => Output: gil_enabled=True speedup=~1.0x (on a normal build)
 
    # => This script was verified on STANDARD CPython 3.13/3.14 (`python3.14t` -- the free-threaded,
    # => PEP 703/779 build -- is not installed in this environment), where `gil_enabled` prints True and
    # => `speedup` stays close to 1x, because the GIL serializes ALL FOUR threads' bytecode onto one core
    # => (co-03). A reader who installs `python3.14t` and reruns this EXACT script would instead see
    # => `gil_enabled=False`, and `speedup` climbing toward NEAR-LINEAR scaling (~3-4x with 4 threads,
    # => on a machine with that many free cores) -- because a free-threaded build removes the single
    # => lock that otherwise forces every thread's Python bytecode through one core at a time (co-04).
    if gil_enabled:  # => the branch this environment actually takes
        assert speedup < 2.0  # => confirms the GIL build did NOT meaningfully parallelize CPU work
    else:  # => the branch a `python3.14t` reader would take instead
        assert speedup > 2.5  # => confirms the free-threaded build DID scale close to linearly
    print("ex-58 OK")  # => Output: ex-58 OK

gil_enabled=True speedup=1.00x
ex-58 OK

"""Example 58: pytest verification for Free-Threaded CPU Scaling."""
 
from example import gil_is_enabled, measure_threaded_speedup
 
 
def test_speedup_matches_the_current_builds_gil_status() -> None:
    speedup = measure_threaded_speedup(iterations=3_000_000, thread_count=4)
    if gil_is_enabled():
        assert speedup < 2.0  # => this environment's standard build: threads don't parallelize CPU work
    else:
        assert speedup > 2.5  # => a free-threaded (python3.14t) build: threads DO parallelize CPU work
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 59: Benchmarking IDENTICAL Threaded Code -- GIL Build vs `python3.14t`."""
 
import shutil  # => shutil.which locates a `python3.14t` binary on PATH, if one is installed
import subprocess  # => runs THIS SAME script under a DIFFERENT interpreter, if one is found
import sys  # => co-04: distinguishes the CURRENTLY running interpreter's own GIL status
import threading  # => the identical threaded-CPU shape used throughout this topic (ex-03, ex-49, ex-58)
import time  # => measures wall time for the empirical speedup ratio
 
ITERATIONS = 5_000_000  # => tuned so 4 threads' worth of work is clearly measurable
THREAD_COUNT = 4  # => how many threads race to do CPU work concurrently
 
 
def gil_is_enabled() -> bool:  # => the SAME version-gated check as ex-04 and ex-58
    if hasattr(sys, "_is_gil_enabled"):  # => only exists on Python 3.13+
        return sys._is_gil_enabled()  # pyright: ignore[reportPrivateUsage]
        # => leading underscore is CPython's naming, not a privacy signal -- the documented 3.13+ API
    return True  # => pre-3.13: no free-threaded option ever existed
 
 
def cpu_task(n: int) -> int:  # => pure CPU work -- no I/O, so any GIL present can NEVER release mid-loop
    total = 0  # => accumulator -- forces real interpreter bytecode execution
    for i in range(n):  # => a tight loop -- exactly the shape a GIL serializes across threads
        total += i  # => trivial arithmetic; only the TIME this takes matters here
    return total  # => the actual value is irrelevant to this example
 
 
def benchmark_current_interpreter() -> float:
    single_start = time.perf_counter()  # => single_start: wall time before the ONE-thread baseline unit
    cpu_task(ITERATIONS)  # => runs exactly ONE unit of work, alone, to establish a per-unit baseline
    single_time = time.perf_counter() - single_start  # => single_time: how long ONE unit takes, uncontended
 
    threads = [threading.Thread(target=cpu_task, args=(ITERATIONS,)) for _ in range(THREAD_COUNT)]
    # => threads: THREAD_COUNT independent threads, all racing to run the SAME cpu_task concurrently
    start = time.perf_counter()  # => start: wall time before the THREAD_COUNT-way concurrent run
    for t in threads:  # => starts every thread
        t.start()  # => each begins its own ITERATIONS-long loop, "concurrently"
    for t in threads:  # => waits for every thread to finish
        t.join()  # => join() blocks until that thread's cpu_task() call returns
    elapsed = time.perf_counter() - start  # => elapsed: wall time for ALL THREAD_COUNT threads combined
    return (single_time * THREAD_COUNT) / elapsed  # => speedup: how much faster the threaded run was vs serial
 
 
def benchmark_other_interpreter(binary_path: str) -> float:
    # => re-runs THIS EXACT FILE under `binary_path`, in "print-only" mode -- one source, two interpreters
    args = [binary_path, __file__, "--benchmark-only"]  # => args: launches THIS module with the parse-friendly flag
    completed = subprocess.run(args, capture_output=True, text=True, check=True, timeout=30)  # => a real subprocess run
    return float(completed.stdout.strip())  # => parses the single number the child process printed
 
 
if __name__ == "__main__":  # => module entry point
    if "--benchmark-only" in sys.argv:  # => the mode `benchmark_other_interpreter` invokes via subprocess
        print(f"{benchmark_current_interpreter():.4f}")  # => prints JUST the number -- nothing else, for easy parsing
    else:  # => the normal, full demonstration mode
        current_speedup = benchmark_current_interpreter()  # => current_speedup: THIS interpreter's own measured ratio
        current_gil_enabled = gil_is_enabled()  # => current_gil_enabled: is THIS interpreter's GIL active?
        print(f"current_build: gil_enabled={current_gil_enabled} speedup={current_speedup:.2f}x")
 
        free_threaded_binary = shutil.which("python3.14t")  # => resolved path to `python3.14t`, or None
        if free_threaded_binary is None:  # => the case in THIS environment -- no free-threaded build installed
            print("python3.14t not found on PATH -- skipping the live cross-build comparison here")
            # => this is the ONLY branch actually exercised when this file is verified in this repo
            print("on a machine with python3.14t installed, rerunning THIS SAME script under it would print >2.5x")
        else:  # => the case on a machine where a reader HAS installed python3.14t alongside standard CPython
            free_threaded_speedup = benchmark_other_interpreter(free_threaded_binary)  # => genuinely re-benchmarks
            print(f"free_threaded_build: speedup={free_threaded_speedup:.2f}x")  # => the OTHER build's own result
            assert free_threaded_speedup > 2.5  # => confirms the t-build genuinely parallelized the identical code
 
        # => Only a `python3.14t` (PEP 703/779, free-threaded) build can genuinely run 4 CPU-bound Python
        # => threads across 4 cores: it removes the single GIL that otherwise serializes bytecode
        # => execution across every thread in a process (co-04). Benchmarking IDENTICAL threaded code on
        # => a standard build shows barely any speedup (this is CPU-bound, so co-05's I/O-release case
        # => doesn't apply); on a free-threaded build, the exact same code scales close to linearly.
        if current_gil_enabled:  # => this environment's actual branch
            assert current_speedup < 2.0  # => confirms the standard GIL build did NOT meaningfully parallelize
        else:  # => the branch a reader running THIS SAME script under python3.14t would take
            assert current_speedup > 2.5  # => confirms a free-threaded interpreter DID scale close to linearly
        print("ex-59 OK")  # => Output: ex-59 OK

current_build: gil_enabled=True speedup=1.00x
python3.14t not found on PATH -- skipping the live cross-build comparison here
on a machine with python3.14t installed, rerunning THIS SAME script under it would print >2.5x
ex-59 OK

"""Example 59: pytest verification for GIL vs Free-Threaded Benchmarking."""
 
import shutil
import sys
 
from example import benchmark_current_interpreter, benchmark_other_interpreter, gil_is_enabled
 
 
def test_current_interpreter_speedup_matches_its_own_gil_status() -> None:
    speedup = benchmark_current_interpreter()
    if gil_is_enabled():
        assert speedup < 2.0  # => the standard build in this environment does not parallelize CPU work
    else:
        assert speedup > 2.5  # => a free-threaded build genuinely parallelizes CPU work
 
 
def test_free_threaded_binary_lookup_does_not_raise() -> None:
    # => shutil.which must always return either a path string or None -- never raise
    result = shutil.which("python3.14t")
    assert result is None or isinstance(result, str)
 
 
def test_benchmark_other_interpreter_reuses_this_script_via_the_same_interpreter() -> None:
    # => no python3.14t is installed here, so this points `benchmark_other_interpreter` at THIS
    # => interpreter's own binary instead -- proving the "--benchmark-only" reuse mechanism works
    speedup = benchmark_other_interpreter(sys.executable)
    assert speedup > 0.0  # => a real, positive speedup number was successfully parsed from stdout
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 60: Double-Checked Locking -- A Lazily-Built Singleton, Safe Under Contention."""
 
import threading  # => co-11: a lock guards the ONE moment construction actually happens
import time  # => `sleep(0)` widens the race window, proven reliable since ex-08
 
_lock = threading.Lock()  # => _lock: the module-level lock guarding the SLOW path (construction) only
_instance: "ExpensiveResource | None" = None  # => _instance: None until the FIRST successful construction
construction_count = [0]  # => construction_count: how many times the expensive constructor actually ran
 
 
class ExpensiveResource:  # => stands in for something costly to build -- a config object, a connection pool
    def __init__(self) -> None:
        time.sleep(0)  # => widens the window between the checks below and this constructor actually running
        construction_count[0] += 1  # => records EVERY construction attempt that reaches this line
 
 
def get_instance() -> ExpensiveResource:  # => the double-checked-locking accessor every thread calls
    global _instance  # => reassigns the MODULE-level singleton, not a local shadow
    if _instance is None:  # => CHECK 1 (unlocked, fast path): usually True only on the very FIRST few calls
        with _lock:  # => only threads that saw None above even attempt to acquire the lock
            if _instance is None:  # => CHECK 2 (locked, slow path): re-checks AFTER acquiring the lock
                _instance = ExpensiveResource()  # => constructs EXACTLY ONCE -- guaranteed by check 2
    return _instance  # => every caller, racing or not, ultimately gets the SAME object
 
 
def get_instance_many_times(results: list[ExpensiveResource]) -> None:
    for _ in range(50):  # => calls get_instance() repeatedly from within ONE thread
        results.append(get_instance())  # => each call should return the IDENTICAL object, every time
 
 
if __name__ == "__main__":  # => module entry point
    thread_results: list[list[ExpensiveResource]] = [[] for _ in range(8)]  # => one private results list per thread
    threads = [
        threading.Thread(target=get_instance_many_times, args=(thread_results[i],)) for i in range(8)
    ]  # => 8 threads, all racing to call get_instance() for the FIRST time simultaneously
    for t in threads:  # => starts every thread
        t.start()  # => all 8 threads immediately race into get_instance()'s unlocked first check
    for t in threads:  # => waits for every thread to finish its 50 calls
        t.join()  # => join() blocks until that thread's get_instance_many_times() call returns
 
    all_instances = [obj for results in thread_results for obj in results]  # => flattens all 8*50=400 results
    unique_instances = {id(obj) for obj in all_instances}  # => unique_instances: distinct object IDENTITIES seen
    print(f"construction_count={construction_count[0]} unique_instances={len(unique_instances)}")
    # => Output: construction_count=1 unique_instances=1
 
    # => Without CHECK 2 (the SECOND `if _instance is None:`, taken WHILE holding the lock), multiple
    # => threads could all pass CHECK 1 before any of them acquires the lock, and each would construct
    # => its OWN `ExpensiveResource` -- a lost-singleton bug, structurally similar to ex-37's TOCTOU race
    # => (co-08). Re-checking AFTER acquiring the lock (co-11) guarantees construction happens EXACTLY
    # => once, while still letting every SUBSEQUENT call skip the lock entirely via the fast unlocked path.
    assert construction_count[0] == 1  # => confirms ExpensiveResource() ran EXACTLY once, despite 8-way contention
    assert len(unique_instances) == 1  # => confirms every one of the 400 calls returned the SAME object
    print("ex-60 OK")  # => Output: ex-60 OK

construction_count=1 unique_instances=1
ex-60 OK

"""Example 60: pytest verification for Double-Checked-Locking Singleton Construction."""
 
import threading
 
from example import ExpensiveResource, construction_count, get_instance_many_times
 
 
def test_singleton_constructed_exactly_once_under_thread_contention() -> None:
    thread_results: list[list[ExpensiveResource]] = [[] for _ in range(6)]
    threads = [threading.Thread(target=get_instance_many_times, args=(thread_results[i],)) for i in range(6)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
 
    all_instances = [obj for results in thread_results for obj in results]
    unique_instances = {id(obj) for obj in all_instances}
    assert construction_count[0] == 1  # => ExpensiveResource() ran exactly once, despite 6-way contention
    assert len(unique_instances) == 1  # => every call returned the same object
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 61: A Hand-Built Reader-Writer Lock -- Many Readers, OR One Writer."""
 
import threading  # => co-13, co-11: builds a richer invariant than a plain Semaphore or Lock alone
import time  # => simulates readers/writers "doing work" while holding their respective access
 
 
class ReaderWriterLock:  # => allows ANY NUMBER of concurrent readers, but only ONE writer, never both at once
    def __init__(self) -> None:
        # => no plain Lock or Semaphore alone can express "many readers OR one writer" -- Condition can
        self._condition = threading.Condition()  # => ONE Condition coordinates both readers and writers
        self._active_readers = 0  # => _active_readers: how many readers currently hold read access
        self._writer_active = False  # => _writer_active: True while exactly one writer holds write access
 
    def acquire_read(self) -> None:
        with self._condition:  # => acquires the Condition's lock for this check-and-update
            while self._writer_active:  # => a WHILE loop (co-14) -- readers wait out any active writer
                self._condition.wait()  # => releases the lock and sleeps until notified
            self._active_readers += 1  # => now safe to join as ANOTHER concurrent reader
 
    def release_read(self) -> None:
        with self._condition:  # => acquires the lock to safely decrement and possibly wake a waiting writer
            self._active_readers -= 1  # => one fewer active reader
            if self._active_readers == 0:  # => only wake others once the LAST reader has actually left
                self._condition.notify_all()  # => a waiting writer (or new readers) can now proceed
 
    def acquire_write(self) -> None:
        with self._condition:  # => acquires the lock for this check-and-update
            while self._writer_active or self._active_readers > 0:  # => a writer needs EXCLUSIVE access
                self._condition.wait()  # => waits out both other writers AND any currently-active readers
            self._writer_active = True  # => now safe to become THE one active writer
 
    def release_write(self) -> None:
        with self._condition:  # => acquires the lock to safely clear the writer flag
            self._writer_active = False  # => releases exclusive access
            self._condition.notify_all()  # => wakes ALL waiters -- readers and writers alike, to re-check
 
 
def reader(rw_lock: ReaderWriterLock, active_readers: list[int], peak_readers: list[int], violations: list[int]) -> None:
    rw_lock.acquire_read()  # => blocks until no writer is active
    active_readers[0] += 1  # => one more concurrent reader, tracked for the invariant check below
    peak_readers[0] = max(peak_readers[0], active_readers[0])  # => peak_readers: the highest concurrency observed
    time.sleep(0.01)  # => simulates reading -- long enough for OTHER readers to overlap, if the lock is correct
    active_readers[0] -= 1  # => this reader is done
    rw_lock.release_read()  # => may wake a waiting writer if this was the LAST active reader
 
 
def writer(rw_lock: ReaderWriterLock, active_readers: list[int], violations: list[int]) -> None:
    rw_lock.acquire_write()  # => blocks until NO readers and NO other writer are active
    violations[0] += 1 if active_readers[0] > 0 else 0  # => THE invariant check: a writer must NEVER see active readers
    time.sleep(0.01)  # => simulates writing, while genuinely EXCLUDING every reader
    rw_lock.release_write()  # => wakes all waiters to re-check the (now cleared) writer flag
 
 
if __name__ == "__main__":  # => module entry point
    rw_lock = ReaderWriterLock()  # => the shared reader-writer lock under test
    active_readers = [0]  # => active_readers[0]: how many readers are CURRENTLY inside their critical section
    peak_readers = [0]  # => peak_readers[0]: the max concurrency any reader batch actually reached
    violations = [0]  # => violations[0]: incremented if a writer EVER observed an active reader (a bug)
 
    readers = [threading.Thread(target=reader, args=(rw_lock, active_readers, peak_readers, violations)) for _ in range(5)]
    # => readers: 5 independent reader threads, all sharing the SAME rw_lock and active_readers counter
    writers = [threading.Thread(target=writer, args=(rw_lock, active_readers, violations)) for _ in range(3)]
    # => writers: 3 independent writer threads, competing for EXCLUSIVE access against readers AND each other
    all_threads = readers + writers  # => all_threads: 5 readers and 3 writers, started together below
    for t in all_threads:  # => starts every reader and writer thread
        # => start() only SCHEDULES the OS thread -- it does not itself block or wait for anything
        t.start()  # => all 8 threads immediately race to acquire read or write access
    for t in all_threads:  # => waits for every thread to finish
        t.join()  # => join() blocks until that thread's reader()/writer() call returns
 
    print(f"peak_readers={peak_readers[0]} violations={violations[0]}")  # => Output: peak_readers=<2 to 5> violations=0
    # => a violations count above 0 would mean the invariant was BROKEN -- exactly what this checks
 
    # => A reader-writer lock enforces a RICHER invariant than a plain Lock: MULTIPLE readers may hold
    # => access simultaneously (peak_readers > 1 here, confirming genuine overlap), but a WRITER always
    # => gets EXCLUSIVE access -- no reader and no other writer may be active while it holds the lock
    # => (co-13). The `while` conditions on BOTH `acquire_read` and `acquire_write` (co-14) are what
    # => enforce this: a writer waits out ALL current readers, and readers wait out any active writer.
    assert peak_readers[0] >= 2  # => confirms readers genuinely overlapped -- this is NOT a plain mutex
    assert violations[0] == 0  # => confirms NO writer ever ran while a reader was active -- the core invariant
    print("ex-61 OK")  # => Output: ex-61 OK

peak_readers=5 violations=0
ex-61 OK

"""Example 61: pytest verification for a Hand-Built Reader-Writer Lock."""
 
import threading
 
from example import ReaderWriterLock, reader, writer
 
 
def test_readers_overlap_but_writers_never_see_an_active_reader() -> None:
    rw_lock = ReaderWriterLock()
    active_readers = [0]
    peak_readers = [0]
    violations = [0]
 
    readers = [threading.Thread(target=reader, args=(rw_lock, active_readers, peak_readers, violations)) for _ in range(4)]
    writers = [threading.Thread(target=writer, args=(rw_lock, active_readers, violations)) for _ in range(2)]
    all_threads = readers + writers
    for t in all_threads:
        t.start()
    for t in all_threads:
        t.join()
 
    assert peak_readers[0] >= 2  # => readers genuinely overlapped -- this is not a plain mutex
    assert violations[0] == 0  # => no writer ever ran while a reader was active
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 62: Detecting a Deadlock -- Finding a Cycle in a Wait-For Graph."""
 
# => co-16, co-18: a "wait-for graph" has an edge A -> B whenever thread A is BLOCKED waiting for
# => something thread B currently holds; a deadlock exists IFF this graph contains a cycle
 
 
def find_cycle(wait_for: dict[str, list[str]]) -> list[str] | None:
    # => wait_for: e.g. {"A": ["B"], "B": ["A"]} means "A waits for B" AND "B waits for A"
    visiting: set[str] = set()  # => visiting: nodes on the CURRENT DFS path -- finding one again means a cycle
    visited: set[str] = set()  # => visited: nodes FULLY explored already -- never needs revisiting
 
    def dfs(node: str, path: list[str]) -> list[str] | None:
        if node in visiting:  # => we've seen `node` EARLIER on this SAME path -- that's the cycle
            cycle_start = path.index(node)  # => cycle_start: where `node` first appeared on this path
            return path[cycle_start:] + [node]  # => the cycle itself, e.g. ["A", "B", "A"]
        if node in visited:  # => already fully explored from a DIFFERENT starting point -- no cycle through here
            return None  # => nothing more to discover down this branch
        visiting.add(node)  # => marks `node` as "currently being explored" on THIS path
        for neighbor in wait_for.get(node, []):  # => follows every thread `node` is waiting for
            found = dfs(neighbor, path + [node])  # => recurses -- extends the path by this node
            if found is not None:  # => a cycle was found somewhere DOWNSTREAM of this call
                return found  # => propagates it straight back up to the original caller
        visiting.discard(node)  # => done exploring `node` on THIS path -- no cycle found through it
        visited.add(node)  # => marks `node` as fully explored, for good, across the whole graph
        return None  # => no cycle reachable from `node`
 
    for start_node in wait_for:  # => tries every node as a POTENTIAL cycle start, in case the graph is disconnected
        cycle = dfs(start_node, [])  # => searches for a cycle reachable from `start_node`
        if cycle is not None:  # => found one -- no need to check any other starting node
            return cycle  # => returns immediately with the first cycle discovered
    return None  # => no cycle exists anywhere in the graph -- no deadlock
 
 
if __name__ == "__main__":  # => module entry point
    deadlocked_graph = {"thread_a": ["thread_b"], "thread_b": ["thread_a"]}  # => ex-29's exact scenario: A waits for B, B waits for A
    cycle = find_cycle(deadlocked_graph)  # => cycle: the deadlock cycle, if any, found in this graph
    print(f"deadlocked_graph cycle={cycle}")  # => Output: deadlocked_graph cycle=['thread_a', 'thread_b', 'thread_a']
 
    ordered_graph = {"thread_a": ["thread_b"], "thread_b": []}  # => ex-30's fix: a GLOBAL order -- B never waits for A
    no_cycle = find_cycle(ordered_graph)  # => no_cycle: expected to be None -- this graph has no deadlock
    print(f"ordered_graph cycle={no_cycle}")  # => Output: ordered_graph cycle=None
 
    three_way_graph = {"t1": ["t2"], "t2": ["t3"], "t3": ["t1"]}  # => a LONGER cycle: t1 -> t2 -> t3 -> t1
    three_way_cycle = find_cycle(three_way_graph)  # => three_way_cycle: should also be detected, not just 2-node cycles
    print(f"three_way_graph cycle={three_way_cycle}")  # => Output: three_way_graph cycle=['t1', 't2', 't3', 't1']
 
    # => A depth-first search that tracks BOTH "currently on this path" (`visiting`) and "fully explored"
    # => (`visited`) can detect a cycle of ANY length in a wait-for graph in O(V+E) time -- re-encountering
    # => a node that's still `visiting` means the DFS has looped back on itself, which is EXACTLY what a
    # => deadlock is: thread A ultimately waiting, transitively, for something IT ALREADY holds (co-16).
    # => Production deadlock detectors (in databases, in the JVM) use this same wait-for-graph technique.
    assert cycle is not None and cycle[0] == cycle[-1]  # => confirms a genuine cycle was found and returned correctly
    assert no_cycle is None  # => confirms the lock-ordered graph is correctly recognized as deadlock-free
    assert three_way_cycle is not None and len(three_way_cycle) == 4  # => confirms a 3-node cycle is ALSO detected
    print("ex-62 OK")  # => Output: ex-62 OK

deadlocked_graph cycle=['thread_a', 'thread_b', 'thread_a']
ordered_graph cycle=None
three_way_graph cycle=['t1', 't2', 't3', 't1']
ex-62 OK

"""Example 62: pytest verification for Wait-For Graph Cycle Detection."""
 
from example import find_cycle
 
 
def test_two_node_cycle_is_detected() -> None:
    graph = {"a": ["b"], "b": ["a"]}
    cycle = find_cycle(graph)
    assert cycle is not None and cycle[0] == cycle[-1]
 
 
def test_acyclic_graph_returns_none() -> None:
    graph = {"a": ["b"], "b": []}
    assert find_cycle(graph) is None
 
 
def test_three_node_cycle_is_detected() -> None:
    graph = {"t1": ["t2"], "t2": ["t3"], "t3": ["t1"]}
    cycle = find_cycle(graph)
    assert cycle is not None and len(cycle) == 4
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 63: A "Lock-Free" Counter -- via a Single-Owner Queue, Not a Lock."""
 
import queue  # => co-20, co-21: the queue itself is the ONLY synchronization primitive used here
import threading  # => multiple client threads REQUEST increments; one owner thread APPLIES them
 
REQUESTS_PER_CLIENT = 500  # => how many increment requests each client thread sends
CLIENT_COUNT = 4  # => how many independent client threads send requests concurrently
 
 
def client(requests: "queue.Queue[int]", count: int) -> None:
    for _ in range(count):  # => sends `count` increment requests, one at a time
        requests.put(1)  # => "please add 1" -- the QUEUE, not a lock, serializes concurrent access to it
 
 
def owner(requests: "queue.Queue[int | None]", total: list[int]) -> None:
    while True:  # => the counter's ONE AND ONLY owner -- no other thread ever touches `total` directly
        item = requests.get()  # => blocks until a request (or the sentinel) arrives
        if item is None:  # => None: the shutdown sentinel, sent once all clients have finished
            break  # => stops the owner loop
        total[0] += item  # => SAFE without a lock: this is the ONLY thread that ever reads or writes `total`
 
 
if __name__ == "__main__":  # => module entry point
    requests: "queue.Queue[int | None]" = queue.Queue()  # => the single channel every client sends requests through
    total = [0]  # => total[0]: the counter, touched EXCLUSIVELY by the owner thread below
 
    owner_thread = threading.Thread(target=owner, args=(requests, total))
    owner_thread.start()  # => starts the SOLE thread that will ever mutate `total`
 
    clients = [threading.Thread(target=client, args=(requests, REQUESTS_PER_CLIENT)) for _ in range(CLIENT_COUNT)]
    for c in clients:  # => starts every client thread
        c.start()  # => each begins sending REQUESTS_PER_CLIENT increment requests into the SAME queue
    for c in clients:  # => waits for every client to finish sending
        c.join()  # => join() blocks until that client's loop has enqueued all its requests
 
    requests.put(None)  # => tells the owner "no more requests are coming" -- AFTER every client has finished
    owner_thread.join()  # => waits for the owner to drain the queue and process every request
 
    expected = CLIENT_COUNT * REQUESTS_PER_CLIENT  # => expected: the mathematically correct total
    print(f"expected={expected} actual={total[0]}")  # => Output: expected=2000 actual=2000
 
    # => This "counter" has NO lock protecting it anywhere -- it doesn't need one, because exactly ONE
    # => thread (the "owner") ever reads or writes it (co-20). Instead of many threads racing to mutate
    # => shared state directly (needing a lock, ex-08/ex-11), every OTHER thread sends a REQUEST through
    # => a thread-safe queue (co-21), and the owner applies requests one at a time, in isolation. This
    # => single-owner pattern is the core idea behind actor-style concurrency: communicate by sending
    # => messages, rather than by sharing memory and coordinating access to it with locks.
    assert total[0] == expected  # => confirms every single increment request was applied, exactly once
    print("ex-63 OK")  # => Output: ex-63 OK

expected=2000 actual=2000
ex-63 OK

"""Example 63: pytest verification for a Single-Owner, Lock-Free Counter."""
 
import queue
import threading
 
from example import client, owner
 
 
def test_single_owner_counter_is_exact_with_no_lock() -> None:
    requests: "queue.Queue[int | None]" = queue.Queue()
    total = [0]
    owner_thread = threading.Thread(target=owner, args=(requests, total))
    owner_thread.start()
 
    clients = [threading.Thread(target=client, args=(requests, 200)) for _ in range(3)]
    for c in clients:
        c.start()
    for c in clients:
        c.join()
 
    requests.put(None)
    owner_thread.join()
 
    assert total[0] == 600  # => 3 clients * 200 requests each, exact -- no lock needed, no lost updates
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 64: `threading.local()` -- Per-Thread State That Never Bleeds Across Threads."""
 
import threading  # => co-07, co-19: the built-in escape hatch from shared-mutable-state hazards
 
_thread_local = threading.local()  # => _thread_local: ONE object, but EVERY thread sees its OWN separate attributes
 
 
def set_and_read_own_value(thread_id: int, observed: dict[int, int]) -> None:
    _thread_local.value = thread_id * 1000  # => sets an attribute on _thread_local -- but ONLY for THIS thread
    for _ in range(3):  # => reads it back multiple times, checking it never changes underneath this thread
        assert _thread_local.value == thread_id * 1000  # => confirms THIS thread's own value stayed stable
    observed[thread_id] = _thread_local.value  # => records what THIS thread saw, for the cross-thread check below
 
 
if __name__ == "__main__":  # => module entry point
    observed: dict[int, int] = {}  # => observed: filled in by each thread with ITS OWN _thread_local.value
    threads = [threading.Thread(target=set_and_read_own_value, args=(i, observed)) for i in range(6)]
    for t in threads:  # => starts every thread
        t.start()  # => each thread immediately sets and re-reads its OWN _thread_local.value
    for t in threads:  # => waits for every thread to finish
        t.join()  # => join() blocks until that thread's set_and_read_own_value() call returns
 
    print(f"observed={observed}")  # => Output: observed={0: 0, 1: 1000, 2: 2000, 3: 3000, 4: 4000, 5: 5000}
 
    # => `threading.local()` creates an object whose ATTRIBUTES are secretly per-thread: setting
    # => `_thread_local.value` in thread 3 does NOT affect what thread 5 sees when it reads
    # => `_thread_local.value` -- each thread gets its OWN isolated slot for the SAME attribute name,
    # => automatically, with zero explicit locking (co-19). This is the standard tool for state that
    # => should be shared-BY-NAME but never shared-BY-VALUE across threads -- e.g. a per-request
    # => database connection, or a per-thread random number generator seed (co-07's isolation, but
    # => achieved WITHIN one process, unlike ex-02's process-level isolation).
    for thread_id, value in observed.items():  # => checks EVERY thread's own recorded value
        assert value == thread_id * 1000  # => confirms no thread ever saw ANOTHER thread's value -- no bleed
    assert len(observed) == 6  # => confirms all 6 threads completed and recorded their own distinct value
    print("ex-64 OK")  # => Output: ex-64 OK

observed={0: 0, 1: 1000, 2: 2000, 3: 3000, 4: 4000, 5: 5000}
ex-64 OK

"""Example 64: pytest verification for `threading.local()` Per-Thread Isolation."""
 
import threading
 
from example import set_and_read_own_value
 
 
def test_thread_local_values_never_bleed_across_threads() -> None:
    observed: dict[int, int] = {}
    threads = [threading.Thread(target=set_and_read_own_value, args=(i, observed)) for i in range(5)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
 
    for thread_id, value in observed.items():
        assert value == thread_id * 1000  # => no thread ever saw another thread's value
    assert len(observed) == 5  # => all 5 threads completed and recorded their own distinct value
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 65: Cancelling a PENDING `Future` -- Before It Ever Starts."""
 
import threading  # => coordinates the "occupy the only worker" trick below
from concurrent.futures import Future, ThreadPoolExecutor  # => co-25, co-23: Futures support `.cancel()`
 
ran_flag = [False]  # => ran_flag[0]: flips to True ONLY if the cancelled task's body actually executes
 
 
def occupy_the_only_worker(release_event: threading.Event) -> str:  # => keeps the SINGLE worker busy
    release_event.wait(timeout=2)  # => blocks until the main thread says it's safe to finish
    return "occupier done"  # => only returned once release_event is set
 
 
def never_should_run() -> str:  # => the task we intend to cancel BEFORE it ever gets a worker
    ran_flag[0] = True  # => if this line EVER executes, cancellation failed
    return "should never see this"  # => this return value should never be observed either
 
 
if __name__ == "__main__":  # => module entry point
    with ThreadPoolExecutor(max_workers=1) as pool:  # => EXACTLY one worker -- forces the second task to QUEUE
        release_event = threading.Event()  # => release_event: lets the main thread control WHEN the occupier finishes
        occupier: Future[str] = pool.submit(occupy_the_only_worker, release_event)  # => grabs the ONLY worker immediately
        pending_future: Future[str] = pool.submit(never_should_run)  # => has NO worker available -- stays PENDING
 
        cancelled = pending_future.cancel()  # => attempts to cancel it WHILE it's still PENDING, not yet running
        print(f"cancelled={cancelled}")  # => Output: cancelled=True
 
        release_event.set()  # => now lets the occupier finish, freeing the pool's only worker
        occupier_result = occupier.result()  # => waits for the occupier to actually complete
        print(f"occupier_result={occupier_result!r}")  # => Output: occupier_result='occupier done'
 
    print(f"ran_flag={ran_flag[0]}")  # => Output: ran_flag=False
 
    # => A `Future` can only be cancelled successfully while it is still PENDING -- queued, but not yet
    # => handed to a worker thread (co-25). By deliberately occupying the pool's ONLY worker first, this
    # => guarantees `pending_future` has nowhere to run, so `.cancel()` (co-23) reliably succeeds and the
    # => task body never executes at all. Once a task has actually STARTED running, `.cancel()` returns
    # => False instead -- Python's stdlib does not support interrupting an already-running thread mid-task.
    assert cancelled is True  # => confirms the cancellation itself was accepted
    assert ran_flag[0] is False  # => confirms the cancelled task's body genuinely never ran
    assert pending_future.cancelled() is True  # => confirms the Future reports itself as cancelled
    print("ex-65 OK")  # => Output: ex-65 OK

cancelled=True
occupier_result='occupier done'
ran_flag=False
ex-65 OK

"""Example 65: pytest verification for Cancelling a Pending `Future`."""
 
import threading
from concurrent.futures import Future, ThreadPoolExecutor
 
from example import never_should_run, occupy_the_only_worker, ran_flag
 
 
def test_pending_future_can_be_cancelled_and_never_runs() -> None:
    ran_flag[0] = False  # => resets shared module state before this test's own assertions
    with ThreadPoolExecutor(max_workers=1) as pool:
        release_event = threading.Event()
        occupier: Future[str] = pool.submit(occupy_the_only_worker, release_event)
        pending_future: Future[str] = pool.submit(never_should_run)
 
        cancelled = pending_future.cancel()
        release_event.set()
        occupier.result()
 
    assert cancelled is True  # => the cancellation was accepted while the task was still pending
    assert ran_flag[0] is False  # => the cancelled task's body never actually executed
    assert pending_future.cancelled() is True  # => the Future correctly reports itself as cancelled
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 66: `asyncio.wait(..., timeout=...)` -- Returns BOTH the Done AND the Pending Sets."""
 
import asyncio  # => co-26: `asyncio.wait` (unlike `gather`) never raises on a timeout -- it just reports
 
DELAYS = [0.05, 0.05, 0.3, 0.3]  # => two FAST tasks (finish within the timeout) and two SLOW tasks (don't)
WAIT_TIMEOUT = 0.15  # => the deadline: enough time for the fast tasks, not enough for the slow ones
 
 
async def labeled_sleep(label: str, delay: float) -> str:  # => label: identifies which task this is
    await asyncio.sleep(delay)  # => the only thing this coroutine does -- simulates work of varying length
    return label  # => returns its own label so the caller can tell WHICH tasks finished
 
 
async def wait_with_timeout() -> tuple[set[str], int]:
    tasks = [asyncio.create_task(labeled_sleep(f"task-{i}", d)) for i, d in enumerate(DELAYS)]
    # => tasks: every coroutine wrapped in a Task -- `asyncio.wait` requires Tasks/Futures, not raw coroutines
    done, pending = await asyncio.wait(tasks, timeout=WAIT_TIMEOUT)  # => returns AT the deadline, not before
    done_labels = {t.result() for t in done}  # => done_labels: the labels of whichever tasks ALREADY finished
    pending_count = len(pending)  # => pending_count: how many tasks were STILL RUNNING when the timeout hit
    for task in pending:  # => cleans up the still-running tasks so they don't outlive this coroutine
        task.cancel()  # => cancels each one -- otherwise they'd keep running in the background, orphaned
    return done_labels, pending_count  # => everything the caller needs to verify the timeout behavior
 
 
if __name__ == "__main__":  # => module entry point
    done_labels, pending_count = asyncio.run(wait_with_timeout())  # => drives the whole scenario to completion
    print(f"done_labels={done_labels} pending_count={pending_count}")  # => Output: done_labels={'task-0','task-1'} pending_count=2
 
    # => `asyncio.wait(tasks, timeout=...)` NEVER raises `TimeoutError` -- it simply returns as soon as
    # => the timeout elapses (or every task finishes, whichever comes first), splitting the tasks into
    # => `done` (already completed) and `pending` (still running) sets (co-26). This is fundamentally
    # => different from `asyncio.wait_for`/`asyncio.timeout` (ex-51), which CANCEL and raise on timeout --
    # => `asyncio.wait` instead hands BOTH sets back, letting the caller decide what to do with stragglers.
    assert done_labels == {"task-0", "task-1"}  # => confirms exactly the two FAST tasks finished in time
    assert pending_count == 2  # => confirms exactly the two SLOW tasks were still running at the deadline
    print("ex-66 OK")  # => Output: ex-66 OK

done_labels={'task-1', 'task-0'} pending_count=2
ex-66 OK

"""Example 66: pytest verification for `asyncio.wait` with a Timeout."""
 
import asyncio
 
from example import wait_with_timeout
 
 
def test_asyncio_wait_returns_both_done_and_pending_sets() -> None:
    done_labels, pending_count = asyncio.run(wait_with_timeout())
    assert done_labels == {"task-0", "task-1"}  # => exactly the two fast tasks finished in time
    assert pending_count == 2  # => exactly the two slow tasks were still running at the deadline
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 67: Cancelling a Task -- Catching `CancelledError` to Run Cleanup."""
 
import asyncio  # => co-26, co-27: cancellation is delivered AS an exception, not a silent flag
 
 
class FakeResource:  # => stands in for something that MUST be released -- a connection, a file handle
    def __init__(self) -> None:
        # => starts unreleased -- the whole point of this example is proving release() DOES get called
        self.released = False  # => released: starts False -- flips to True ONLY via release()
 
    def release(self) -> None:
        # => a REAL resource's release() might close a socket, flush a buffer, or free a lock
        self.released = True  # => marks this resource as properly cleaned up
 
 
async def work_with_cleanup(resource: FakeResource) -> None:
    # => the caller is responsible for making sure THIS coroutine reaches an await before cancelling
    try:
        await asyncio.sleep(10)  # => a long-running await -- this is where cancellation will actually land
    except asyncio.CancelledError:  # => `task.cancel()` raises THIS exception at the current `await` point
        resource.release()  # => the cleanup -- runs BECAUSE the exception was caught here, not silently dropped
        raise  # => re-raises: swallowing CancelledError entirely is almost always wrong (co-27)
 
 
async def run_and_cancel() -> tuple[bool, bool]:
    # => orchestrates the whole demo: create the task, let it start, cancel it, observe the fallout
    resource = FakeResource()  # => resource: the thing this task is responsible for cleaning up
    task = asyncio.create_task(work_with_cleanup(resource))  # => schedules the coroutine to start running
    await asyncio.sleep(0.02)  # => lets the task actually REACH its `await asyncio.sleep(10)` before cancelling
    task.cancel()  # => requests cancellation -- delivers CancelledError at the task's current await point
    was_cancelled = False  # => was_cancelled: True only if awaiting the task itself raises CancelledError
    try:
        # => `await task` is where the CALLER (not the task itself) observes the cancellation
        await task  # => awaiting a cancelled task re-raises the SAME CancelledError to the caller
    except asyncio.CancelledError:  # => confirms cancellation propagated all the way out, as expected
        was_cancelled = True  # => records that the caller correctly observed the cancellation
    return resource.released, was_cancelled  # => everything needed to verify BOTH cleanup and propagation
 
 
if __name__ == "__main__":  # => module entry point
    released, was_cancelled = asyncio.run(run_and_cancel())  # => drives the cancellation scenario to completion
    print(f"released={released} was_cancelled={was_cancelled}")  # => Output: released=True was_cancelled=True
 
    # => `task.cancel()` does NOT immediately stop a task -- it schedules `CancelledError` to be raised
    # => at the NEXT `await` point inside that task (co-26). Catching it, like any exception, is the
    # => correct place to release resources (co-27) -- but the handler must `raise` (or otherwise NOT
    # => suppress it) afterward, since callers awaiting a cancelled task expect to see CancelledError
    # => propagate; swallowing it silently would make the task appear to have completed normally.
    assert released is True  # => confirms the `except CancelledError:` block genuinely ran cleanup
    assert was_cancelled is True  # => confirms CancelledError still propagated out to the awaiting caller
    print("ex-67 OK")  # => Output: ex-67 OK

released=True was_cancelled=True
ex-67 OK

"""Example 67: pytest verification for Async Task Cancellation Cleanup."""
 
import asyncio
 
from example import run_and_cancel
 
 
def test_cancellation_runs_cleanup_and_still_propagates() -> None:
    released, was_cancelled = asyncio.run(run_and_cancel())
    assert released is True  # => the except CancelledError block ran the resource release
    assert was_cancelled is True  # => CancelledError still propagated out to the awaiting caller
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 68: `asyncio.TaskGroup` -- One Failure Cancels ALL Its Siblings."""
 
import asyncio  # => co-26: `TaskGroup` (3.11+) is STRUCTURED concurrency -- gather's more disciplined cousin
 
 
class DownloadError(Exception):  # => a domain-specific failure, to show it propagates with its own type intact
    """Raised by `flaky_download` when a simulated download fails."""
 
 
async def flaky_download(n: int, fail_at: int, ran_to_completion: list[int]) -> int:
    if n == fail_at:  # => exactly ONE task in the group is designed to fail
        await asyncio.sleep(0.02)  # => fails a LITTLE after the others start, so they're genuinely mid-flight
        raise DownloadError(f"download {n} failed")  # => the failure that should cancel every sibling
    await asyncio.sleep(0.2)  # => the OTHER tasks are deliberately slower -- still running when the failure hits
    ran_to_completion.append(n)  # => only reached if this task was NEVER cancelled by a sibling's failure
    return n  # => a trivial "result" for the tasks that DO succeed
 
 
async def run_task_group() -> tuple[bool, list[int]]:
    ran_to_completion: list[int] = []  # => ran_to_completion: filled in ONLY by tasks that finish uncancelled
    raised = False  # => raised: True once the group's own ExceptionGroup is caught below
    try:
        async with asyncio.TaskGroup() as tg:  # => `async with` scopes the group -- exits ONLY when ALL are done
            for i in range(4):  # => launches 4 sibling tasks, one of which (i == 1) is designed to fail
                tg.create_task(flaky_download(i, fail_at=1, ran_to_completion=ran_to_completion))
    except* DownloadError:  # => `except*` (3.11+): TaskGroup wraps failures in an ExceptionGroup, unpacked here
        raised = True  # => confirms the failure propagated out of the `async with` block, as designed
    return raised, ran_to_completion  # => everything the caller needs to verify cancellation propagated
 
 
if __name__ == "__main__":  # => module entry point
    raised, ran_to_completion = asyncio.run(run_task_group())  # => drives the whole scenario to completion
    print(f"raised={raised} ran_to_completion={ran_to_completion}")  # => Output: raised=True ran_to_completion=[]
 
    # => `asyncio.TaskGroup` implements STRUCTURED concurrency (co-26): the moment ANY task inside the
    # => `async with` block raises, EVERY OTHER task in the group is automatically cancelled, and the
    # => group's own exit re-raises the failure (wrapped in an `ExceptionGroup`, unpacked here via
    # => `except*`). This is a stricter, safer default than `asyncio.gather` (ex-28, ex-50), where a
    # => failing task by itself does NOT automatically cancel its siblings unless explicitly configured --
    # => `TaskGroup` makes "one fails, all stop" the guaranteed behavior, not something to remember to add.
    assert raised is True  # => confirms the DownloadError genuinely propagated out of the TaskGroup
    assert ran_to_completion == []  # => confirms the slower siblings were CANCELLED, never reaching their append
    print("ex-68 OK")  # => Output: ex-68 OK

raised=True ran_to_completion=[]
ex-68 OK

"""Example 68: pytest verification for `asyncio.TaskGroup` Failure Cancellation."""
 
import asyncio
 
from example import run_task_group
 
 
def test_one_failing_task_cancels_all_its_siblings() -> None:
    raised, ran_to_completion = asyncio.run(run_task_group())
    assert raised is True  # => the DownloadError propagated out of the TaskGroup
    assert ran_to_completion == []  # => the slower siblings were cancelled, never completing
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 69: Fetch Many "URLs" Concurrently, Rate-Limited, Then Aggregate."""
 
import asyncio  # => co-26, co-13: gather for aggregation, a Semaphore for the rate limit
 
URLS = [f"https://example.test/page-{i}" for i in range(12)]  # => URLS: 12 simulated pages to "fetch"
MAX_CONCURRENT_FETCHES = 3  # => at most this many "requests" may be in flight simultaneously
 
 
async def fetch_page(url: str, sem: asyncio.Semaphore, active: list[int], peak: list[int]) -> int:
    async with sem:  # => throttles concurrency -- the SAME pattern as ex-53, applied to a real-shaped task
        active[0] += 1  # => one more fetch now "in flight"
        peak[0] = max(peak[0], active[0])  # => peak: the highest concurrency level EVER observed
        await asyncio.sleep(0.02)  # => simulates network latency -- a REAL fetch would `await` here too
        active[0] -= 1  # => this fetch is done -- frees a permit for the next waiter
        return len(url)  # => stands in for "content length" -- a trivial but checkable per-page result
 
 
async def fetch_and_aggregate() -> tuple[int, int]:
    sem = asyncio.Semaphore(MAX_CONCURRENT_FETCHES)  # => sem: the shared limiter for all 12 fetches
    active = [0]  # => active[0]: how many fetches are CURRENTLY in flight
    peak = [0]  # => peak[0]: updated inside every fetch -- the maximum concurrency actually reached
    lengths = await asyncio.gather(*(fetch_page(url, sem, active, peak) for url in URLS))  # => runs ALL 12 concurrently
    total_length = sum(lengths)  # => total_length: the AGGREGATE across every fetched page
    return total_length, peak[0]  # => everything the caller needs to verify correctness AND the rate limit
 
 
if __name__ == "__main__":  # => module entry point
    total_length, peak = asyncio.run(fetch_and_aggregate())  # => drives all 12 fetches through the shared semaphore
    print(f"total_length={total_length} peak={peak}")  # => Output: total_length=<sum of URL lengths> peak=3
 
    expected_total = sum(len(url) for url in URLS)  # => expected_total: the serial, ground-truth aggregate
 
    # => Fetching many resources concurrently AND aggregating their results is one of `asyncio.gather`'s
    # => most common real-world uses (co-26) -- but unbounded concurrency can overwhelm the target
    # => server (or a rate-limited API). Wrapping each fetch in `async with sem:` (co-13, exactly as in
    # => ex-53) caps how many are in flight at once WITHOUT changing the aggregation logic at all: the
    # => `gather()` call still collects every result, in order, once all 12 fetches have completed.
    assert total_length == expected_total  # => confirms the aggregate exactly matches the serial baseline
    assert peak <= MAX_CONCURRENT_FETCHES  # => confirms concurrency NEVER exceeded the declared cap
    assert peak == MAX_CONCURRENT_FETCHES  # => confirms concurrency actually REACHED the cap -- genuinely throttled
    print("ex-69 OK")  # => Output: ex-69 OK

total_length=326 peak=3
ex-69 OK

"""Example 69: pytest verification for Concurrent Fetch + Aggregate with a Semaphore Cap."""
 
import asyncio
 
from example import MAX_CONCURRENT_FETCHES, URLS, fetch_and_aggregate
 
 
def test_aggregate_correct_and_concurrency_capped() -> None:
    total_length, peak = asyncio.run(fetch_and_aggregate())
    expected_total = sum(len(url) for url in URLS)
    assert total_length == expected_total  # => the aggregate exactly matches the serial baseline
    assert peak <= MAX_CONCURRENT_FETCHES  # => concurrency never exceeded the declared cap
    assert peak == MAX_CONCURRENT_FETCHES  # => concurrency actually reached the cap
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 70: A Three-Stage Pipeline -- Read -> Transform -> Write, via Two Queues."""
 
import queue  # => co-22, co-21: each STAGE is a thread; each QUEUE is the hand-off between two stages
import threading  # => one dedicated thread per stage -- read, transform, write
 
ITEM_COUNT = 15  # => how many items flow all the way through the pipeline
 
 
def read_stage(raw_out: "queue.Queue[int | None]") -> None:
    # => stage 1 of 3 -- the ONLY producer for raw_queue
    for i in range(ITEM_COUNT):  # => "reads" ITEM_COUNT items -- simulated here as just their own index
        raw_out.put(i)  # => hands each raw item off to the transform stage via the first queue
    raw_out.put(None)  # => sentinel: tells the transform stage there's nothing more to read
 
 
def transform_stage(raw_in: "queue.Queue[int | None]", transformed_out: "queue.Queue[int | None]") -> None:
    # => stage 2 of 3 -- the ONLY consumer of raw_queue AND the ONLY producer for transformed_queue
    while True:  # => keeps pulling raw items until its OWN sentinel arrives
        item = raw_in.get()  # => blocks until the read stage has something ready
        if item is None:  # => the read stage's sentinel -- no more raw items are coming
            transformed_out.put(None)  # => propagates a sentinel of its OWN to the write stage
            break  # => stops the transform stage
        transformed_out.put(item * item)  # => the actual "transform": squaring, handed to the write stage
 
 
def write_stage(transformed_in: "queue.Queue[int | None]", written: list[int]) -> None:
    # => stage 3 of 3 -- the ONLY consumer of transformed_queue
    while True:  # => keeps pulling transformed items until the sentinel arrives
        item = transformed_in.get()  # => blocks until the transform stage has something ready
        if item is None:  # => the transform stage's sentinel -- nothing more is coming
            break  # => stops the write stage
        written.append(item)  # => "writes" the item -- here, just records it for verification
 
 
if __name__ == "__main__":  # => module entry point
    raw_queue: "queue.Queue[int | None]" = queue.Queue()  # => raw_queue: hand-off between read and transform
    transformed_queue: "queue.Queue[int | None]" = queue.Queue()  # => transformed_queue: hand-off between transform and write
    written: list[int] = []  # => written: filled in by the write stage, in the FINAL pipeline order
 
    reader = threading.Thread(target=read_stage, args=(raw_queue,))
    transformer = threading.Thread(target=transform_stage, args=(raw_queue, transformed_queue))
    writer = threading.Thread(target=write_stage, args=(transformed_queue, written))
    for stage in (reader, transformer, writer):  # => starts all three stages together
        stage.start()  # => each stage begins running concurrently, connected only by the two queues
    for stage in (reader, transformer, writer):  # => waits for every stage to fully drain and exit
        stage.join()  # => join() blocks until that stage's loop has completed
 
    expected = [i * i for i in range(ITEM_COUNT)]  # => expected: what the WHOLE pipeline should have produced
    print(f"written={written}")  # => Output: written=[0, 1, 4, 9, 16, ..., 196]
 
    # => Each stage is its OWN thread, and each `queue.Queue` between two stages is thread-safe by
    # => construction (co-21) -- no explicit lock is needed anywhere in this pipeline. Because EACH
    # => queue has exactly ONE producer thread and ONE consumer thread here, FIFO ordering (co-22) is
    # => preserved end-to-end: item `i` is read, transformed, and written in that SAME relative order it
    # => started in, even though all three stages are running concurrently with each other the whole time.
    assert written == expected  # => confirms every item passed through all three stages, correctly, in order
    print("ex-70 OK")  # => Output: ex-70 OK

written=[0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196]
ex-70 OK

"""Example 70: pytest verification for a Three-Stage Read-Transform-Write Pipeline."""
 
import queue
import threading
 
from example import read_stage, transform_stage, write_stage
 
 
def test_pipeline_preserves_order_and_transforms_every_item() -> None:
    raw_queue: "queue.Queue[int | None]" = queue.Queue()
    transformed_queue: "queue.Queue[int | None]" = queue.Queue()
    written: list[int] = []
 
    reader = threading.Thread(target=read_stage, args=(raw_queue,))
    transformer = threading.Thread(target=transform_stage, args=(raw_queue, transformed_queue))
    writer = threading.Thread(target=write_stage, args=(transformed_queue, written))
    for stage in (reader, transformer, writer):
        stage.start()
    for stage in (reader, transformer, writer):
        stage.join()
 
    assert written == [i * i for i in range(15)]  # => every item passed through all three stages, in order
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 71: Work-Stealing -- an Idle Worker Steals From an Overloaded Peer's Deque."""
 
from collections import deque  # => co-28: the classic data structure behind work-stealing schedulers
 
 
def simulate_work_stealing(
    worker_a: "deque[int]", worker_b: "deque[int]"
) -> tuple[int, int, list[str]]:  # => a single-threaded SKETCH of the idea -- deterministic, not a race
    completed_a = 0  # => completed_a: how many tasks worker "a" ends up processing, own OR stolen
    completed_b = 0  # => completed_b: how many tasks worker "b" ends up processing, own OR stolen
    steal_events: list[str] = []  # => steal_events: a log of every time one worker stole from the other
 
    while worker_a or worker_b:  # => keeps going until BOTH deques are fully drained
        if worker_a:  # => worker "a" has its OWN work -- processes from its OWN end (locality, low contention)
            worker_a.pop()  # => pops from the RIGHT end -- the owner's "local" end of its own deque
            completed_a += 1  # => counts one task processed by worker "a"
        elif worker_b:  # => worker "a" is IDLE -- its own deque is empty, but worker "b" still has work
            worker_b.popleft()  # => STEALS from the LEFT end -- the opposite end from "b"'s own popping
            completed_a += 1  # => the STOLEN task still counts toward worker "a"'s own completed total
            steal_events.append("a<-b")  # => records that "a" stole from "b" this round
 
        if worker_b:  # => worker "b" has its OWN work -- same local-end popping as worker "a" above
            worker_b.pop()  # => pops from the RIGHT end -- worker "b"'s own local end
            completed_b += 1  # => counts one task processed by worker "b"
        elif worker_a:  # => worker "b" is IDLE -- symmetric to the "a" branch above
            worker_a.popleft()  # => STEALS from the LEFT end of worker "a"'s deque
            completed_b += 1  # => the STOLEN task counts toward worker "b"'s own completed total
            steal_events.append("b<-a")  # => records that "b" stole from "a" this round
 
    return completed_a, completed_b, steal_events  # => everything needed to verify the load actually balanced
 
 
if __name__ == "__main__":  # => module entry point
    worker_a_tasks: "deque[int]" = deque(range(2))  # => worker "a" starts with only 2 tasks -- WILL run out fast
    worker_b_tasks: "deque[int]" = deque(range(18))  # => worker "b" starts with 18 tasks -- deliberately OVERLOADED
    completed_a, completed_b, steal_events = simulate_work_stealing(worker_a_tasks, worker_b_tasks)
    print(f"completed_a={completed_a} completed_b={completed_b} steal_count={len(steal_events)}")
    # => Output: completed_a=10 completed_b=10 steal_count=8
 
    total_tasks = 2 + 18  # => total_tasks: the mathematically correct grand total, from BOTH initial deques
    # => the exact 8/12 split above comes from this simulation's own deterministic round-robin order
 
    # => Without stealing, worker "a" would finish its 2 tasks almost instantly and then sit COMPLETELY
    # => idle while worker "b" churns through 18 tasks alone -- a wasted core. Work-stealing lets an idle
    # => worker grab tasks from an OVERLOADED peer's deque instead (co-28), and popping from OPPOSITE ends
    # => (owner: local end; thief: far end) minimizes contention between the two, since they rarely reach
    # => for the same slot. The result: BOTH workers stay busy, and the total work still gets fully done.
    assert completed_a + completed_b == total_tasks  # => confirms every task was processed exactly once
    assert completed_a > 2  # => confirms worker "a" processed MORE than its own original 2 tasks -- it stole work
    assert len(steal_events) > 0  # => confirms stealing genuinely happened, not just idle waiting
    print("ex-71 OK")  # => Output: ex-71 OK

completed_a=10 completed_b=10 steal_count=8
ex-71 OK

"""Example 71: pytest verification for Work-Stealing Load Balancing."""
 
from collections import deque
 
from example import simulate_work_stealing
 
 
def test_idle_worker_steals_work_and_load_balances() -> None:
    worker_a: "deque[int]" = deque(range(1))
    worker_b: "deque[int]" = deque(range(9))
    completed_a, completed_b, steal_events = simulate_work_stealing(worker_a, worker_b)
 
    assert completed_a + completed_b == 10  # => every task processed exactly once
    assert completed_a > 1  # => worker "a" processed more than its own original 1 task -- it stole work
    assert len(steal_events) > 0  # => stealing genuinely happened
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 72: `multiprocessing.shared_memory` -- Genuine No-Copy Cross-Process Access."""
 
import struct  # => packs/unpacks fixed-size integers directly into the shared memory buffer
from multiprocessing import Process, shared_memory  # => co-24, co-19: REAL shared memory, not a copy-based Queue
 
ARRAY_LENGTH = 2000  # => how many 8-byte integers live in the shared block
ITEM_SIZE = 8  # => bytes per integer -- struct format "q" (signed long long) is always 8 bytes
 
 
def double_values_in_place(shm_name: str, length: int) -> None:  # => runs in a SEPARATE process
    shm = shared_memory.SharedMemory(name=shm_name)  # => ATTACHES to the EXISTING block by name -- no copy made
    buf = shm.buf  # => buf: typed as `memoryview | None` by typeshed -- narrowed once, right here
    assert buf is not None  # => always non-None once a SharedMemory is genuinely attached (co-19's own invariant)
    try:
        for i in range(length):  # => mutates every value, IN PLACE, directly inside the shared buffer
            offset = i * ITEM_SIZE  # => offset: this value's exact byte position within `buf`
            (value,) = struct.unpack_from("q", buf, offset)  # => reads the CURRENT value at that offset
            struct.pack_into("q", buf, offset, value * 2)  # => writes the DOUBLED value back, same memory
    finally:
        shm.close()  # => releases THIS process's handle -- does NOT destroy the underlying shared memory itself
 
 
if __name__ == "__main__":  # => module entry point
    shm = shared_memory.SharedMemory(create=True, size=ARRAY_LENGTH * ITEM_SIZE)  # => allocates a NEW OS-level shared block
    parent_buf = shm.buf  # => parent_buf: the SAME `memoryview | None` narrowing as inside the child process
    assert parent_buf is not None  # => always non-None right after a fresh, successful `create=True` call
    try:
        for i in range(ARRAY_LENGTH):  # => initializes the block with the values 0, 1, 2, ...
            struct.pack_into("q", parent_buf, i * ITEM_SIZE, i)  # => writes value `i` at its own 8-byte slot
 
        child = Process(target=double_values_in_place, args=(shm.name, ARRAY_LENGTH))  # => child: attaches by NAME
        child.start()  # => the child process opens the SAME underlying memory -- no serialization of the array itself
        child.join()  # => waits for the child to finish doubling every value in place
 
        results = [struct.unpack_from("q", parent_buf, i * ITEM_SIZE)[0] for i in range(ARRAY_LENGTH)]
        expected = [i * 2 for i in range(ARRAY_LENGTH)]  # => expected: what doubling every original value SHOULD give
        print(f"results[:5]={results[:5]} expected[:5]={expected[:5]}")  # => Output: results[:5]=[0,2,4,6,8] expected[:5]=[0,2,4,6,8]
 
        # => Unlike `multiprocessing.Queue` (ex-46), which PICKLES and COPIES every item across the
        # => process boundary, `shared_memory.SharedMemory` gives BOTH processes direct access to the
        # => SAME underlying OS memory block (co-24) -- the child's writes are visible to the parent
        # => WITHOUT any explicit copy or message being sent back. This matters for large arrays, where
        # => copying would dominate the cost; only the small `shm.name` string needs to cross processes.
        # => Manual synchronization (co-19) is still the caller's responsibility -- unlike ex-47's
        # => `Value`, plain `SharedMemory` has no built-in lock of its own.
        assert results == expected  # => confirms every value was doubled correctly, IN the shared block itself
    finally:
        shm.close()  # => releases the PARENT's own handle to the shared memory
        shm.unlink()  # => releases the underlying OS resource -- only the CREATOR should ever call this
    print("ex-72 OK")  # => Output: ex-72 OK

results[:5]=[0, 2, 4, 6, 8] expected[:5]=[0, 2, 4, 6, 8]
ex-72 OK

"""Example 72: pytest verification for `multiprocessing.shared_memory`."""
 
import struct
from multiprocessing import Process, shared_memory
 
from example import ITEM_SIZE, double_values_in_place
 
 
def test_child_process_mutates_the_shared_block_in_place() -> None:
    length = 100
    shm = shared_memory.SharedMemory(create=True, size=length * ITEM_SIZE)
    buf = shm.buf
    assert buf is not None
    try:
        for i in range(length):
            struct.pack_into("q", buf, i * ITEM_SIZE, i)
 
        child = Process(target=double_values_in_place, args=(shm.name, length))
        child.start()
        child.join()
 
        results = [struct.unpack_from("q", buf, i * ITEM_SIZE)[0] for i in range(length)]
        assert results == [i * 2 for i in range(length)]  # => the child's writes are visible without copying
    finally:
        shm.close()
        shm.unlink()
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 73: `asyncio.to_thread` -- the High-Level Shortcut for Offloading Blocking Calls."""
 
import asyncio  # => co-27, co-23: `asyncio.to_thread` (3.9+) wraps `run_in_executor` in a friendlier API
import time  # => simulates a blocking call with no async equivalent
 
TICK_INTERVAL = 0.02  # => how often the ticker coroutine records a timestamp
TICK_COUNT = 8  # => how many ticks the ticker tries to record
 
 
def blocking_legacy_call(delay: float) -> str:  # => a PLAIN (non-async) function -- imagine a sync SDK call
    time.sleep(delay)  # => genuinely blocks the calling thread -- this is the whole point of the example
    return "legacy result"  # => the result this blocking call eventually produces
 
 
async def ticker(timestamps: list[float]) -> None:
    for _ in range(TICK_COUNT):  # => tries to record TICK_COUNT evenly-spaced timestamps
        timestamps.append(time.perf_counter())  # => records "now" -- reveals whether the loop stayed responsive
        await asyncio.sleep(TICK_INTERVAL)  # => a cooperative wait, letting other coroutines/tasks run
 
 
def max_gap(timestamps: list[float]) -> float:  # => max_gap: the LARGEST delay between consecutive ticks
    gaps = [b - a for a, b in zip(timestamps, timestamps[1:])]  # => gaps: every consecutive tick-to-tick delay
    return max(gaps)  # => a stalled loop shows up as one unusually LARGE gap
 
 
async def run_with_to_thread() -> tuple[list[float], str]:
    timestamps: list[float] = []  # => timestamps: filled in by the ticker coroutine below
    offloaded = asyncio.to_thread(blocking_legacy_call, 0.15)  # => `to_thread` == `run_in_executor(None, func, *args)`
    result, _ = await asyncio.gather(offloaded, ticker(timestamps))  # => runs BOTH concurrently, offloaded first
    return timestamps, result  # => everything the caller needs to verify responsiveness AND correctness
 
 
if __name__ == "__main__":  # => module entry point
    timestamps, result = asyncio.run(run_with_to_thread())  # => drives the offloaded scenario to completion
    gap = max_gap(timestamps)  # => gap: the worst stall the ticker actually observed
    print(f"gap={gap:.3f}s result={result!r}")  # => Output: gap=~0.02s result='legacy result'
 
    # => `asyncio.to_thread(func, *args)` is sugar for `loop.run_in_executor(None, func, *args)` (ex-55) --
    # => same underlying mechanism (a background thread from the loop's default `ThreadPoolExecutor`,
    # => co-23), but without needing to fetch the running loop explicitly first (co-27). Prefer
    # => `to_thread` in ordinary async code; reach for the lower-level `run_in_executor` only when you
    # => need a NON-default executor (e.g. a `ProcessPoolExecutor` for CPU-bound offloaded work).
    assert gap < TICK_INTERVAL * 3  # => confirms the ticker stayed on schedule -- the loop was NEVER stalled
    assert result == "legacy result"  # => confirms the offloaded call still produced the correct result
    print("ex-73 OK")  # => Output: ex-73 OK

gap=0.022s result='legacy result'
ex-73 OK

"""Example 73: pytest verification for `asyncio.to_thread`."""
 
import asyncio
 
from example import TICK_INTERVAL, max_gap, run_with_to_thread
 
 
def test_event_loop_stays_responsive_while_to_thread_call_runs() -> None:
    timestamps, result = asyncio.run(run_with_to_thread())
    assert max_gap(timestamps) < TICK_INTERVAL * 3  # => the ticker never stalled -- the loop stayed responsive
    assert result == "legacy result"  # => the offloaded call still produced the correct result
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 74: `while not predicate: cond.wait()` -- Guarding Against a Spurious/Early Wakeup."""
 
import threading  # => co-14: the SAME Condition primitive as ex-19, ex-41, ex-61
import time  # => gives each waiter thread time to genuinely register inside cond.wait()
 
 
def waiter_with_if(cond: threading.Condition, ready: list[bool], results: list[bool], started: threading.Event) -> None:
    with cond:  # => acquires the Condition's lock
        started.set()  # => signals the main thread that THIS waiter is about to call wait()
        if not ready[0]:  # => a SINGLE `if` check -- the BUGGY pattern this example warns against
            cond.wait()  # => sleeps until notified -- but does NOT re-check `ready` after waking up
        results.append(ready[0])  # => proceeds REGARDLESS of whether `ready[0]` is ACTUALLY True by now
 
 
def waiter_with_while(cond: threading.Condition, ready: list[bool], results: list[bool], started: threading.Event) -> None:
    with cond:  # => acquires the SAME Condition's lock
        started.set()  # => signals the main thread that THIS waiter is about to call wait()
        while not ready[0]:  # => a WHILE loop -- the CORRECT pattern (co-14)
            cond.wait()  # => sleeps until notified, then RE-CHECKS the predicate before proceeding
        results.append(ready[0])  # => only reaches here once `ready[0]` is GENUINELY True
 
 
if __name__ == "__main__":  # => module entry point
    cond = threading.Condition()  # => cond: shared by both waiter threads AND the main thread's notifications
    ready: list[bool] = [False]  # => ready[0]: the ACTUAL predicate both waiters are waiting on
    if_results: list[bool] = []  # => if_results: filled in by the buggy `if`-based waiter
    while_results: list[bool] = []  # => while_results: filled in by the correct `while`-based waiter
    if_started = threading.Event()  # => if_started: set once the `if`-waiter has entered its critical section
    while_started = threading.Event()  # => while_started: set once the `while`-waiter has entered its
 
    if_thread = threading.Thread(target=waiter_with_if, args=(cond, ready, if_results, if_started))
    while_thread = threading.Thread(target=waiter_with_while, args=(cond, ready, while_results, while_started))
    if_thread.start()  # => starts the buggy waiter
    while_thread.start()  # => starts the correct waiter
    if_started.wait(timeout=2)  # => waits for the `if`-waiter to actually reach `cond.wait()`
    while_started.wait(timeout=2)  # => waits for the `while`-waiter to actually reach `cond.wait()`
    time.sleep(0.05)  # => a small margin to be confident BOTH threads are genuinely blocked inside wait()
 
    with cond:  # => the main thread now sends a PREMATURE notification -- `ready[0]` is STILL False here
        cond.notify_all()  # => wakes BOTH waiters -- simulating a spurious wakeup / an early notify()
 
    time.sleep(0.05)  # => gives the woken threads time to react to the premature notification above
 
    with cond:  # => now the main thread makes the predicate GENUINELY true
        ready[0] = True  # => flips the ACTUAL condition both waiters are supposed to be waiting for
        cond.notify_all()  # => wakes both waiters again -- THIS time the predicate really holds
 
    if_thread.join(timeout=2)  # => waits for the buggy waiter to finish
    while_thread.join(timeout=2)  # => waits for the correct waiter to finish
 
    print(f"if_results={if_results} while_results={while_results}")  # => Output: if_results=[False] while_results=[True]
 
    # => The premature `notify_all()` above wakes BOTH waiters even though `ready[0]` is still False.
    # => The `if`-based waiter has no way to notice this -- it already passed its ONE check, so it
    # => proceeds with a WRONG value. The `while`-based waiter re-checks the predicate every time it
    # => wakes (co-14): finding it still False, it calls `cond.wait()` again, and only proceeds once
    # => `ready[0]` is genuinely True. This is why EVERY `Condition.wait()` call belongs inside a
    # => `while`, never a plain `if` -- spurious wakeups are a documented possibility, not a corner case.
    assert if_results == [False]  # => confirms the buggy `if` pattern proceeded on a STALE, incorrect value
    assert while_results == [True]  # => confirms the correct `while` pattern only proceeded once genuinely ready
    print("ex-74 OK")  # => Output: ex-74 OK

if_results=[False] while_results=[True]
ex-74 OK

"""Example 74: pytest verification for the `while not predicate:` Condition Loop Pattern."""
 
import threading
import time
 
from example import waiter_with_if, waiter_with_while
 
 
def test_if_pattern_proceeds_on_a_premature_notify_while_loop_pattern_does_not() -> None:
    cond = threading.Condition()
    ready: list[bool] = [False]
    if_results: list[bool] = []
    while_results: list[bool] = []
    if_started = threading.Event()
    while_started = threading.Event()
 
    if_thread = threading.Thread(target=waiter_with_if, args=(cond, ready, if_results, if_started))
    while_thread = threading.Thread(target=waiter_with_while, args=(cond, ready, while_results, while_started))
    if_thread.start()
    while_thread.start()
    if_started.wait(timeout=2)
    while_started.wait(timeout=2)
    time.sleep(0.05)
 
    with cond:
        cond.notify_all()  # => a premature notification -- ready[0] is still False here
    time.sleep(0.05)
 
    with cond:
        ready[0] = True
        cond.notify_all()  # => the genuine notification
 
    if_thread.join(timeout=2)
    while_thread.join(timeout=2)
 
    assert if_results == [False]  # => the buggy if-pattern proceeded on a stale, incorrect value
    assert while_results == [True]  # => the correct while-pattern only proceeded once genuinely ready
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 75: A `Barrier` Synchronizes Phased Parallel Computation."""
 
import threading  # => co-15, co-28: a Barrier enforces "everyone finishes phase N before ANYONE starts N+1"
import time  # => `time.sleep` gives each worker a DIFFERENT phase-1 duration, to genuinely test the rendezvous
 
WORKER_COUNT = 4  # => four independent workers, each moving through the SAME three phases
 
 
def phased_worker(
    worker_id: int, barrier: threading.Barrier, phase1_end: list[float], phase2_start: list[float]
) -> None:
    # => moves through phase 1, a rendezvous, phase 2, and a second rendezvous -- twice per worker
    time.sleep(0.01 * worker_id)  # => DIFFERENT workers take DIFFERENT amounts of time for phase 1
    phase1_end[worker_id] = time.perf_counter()  # => records exactly when THIS worker's phase 1 finished
    barrier.wait()  # => BLOCKS here until ALL WORKER_COUNT workers have reached this SAME point
    phase2_start[worker_id] = time.perf_counter()  # => records exactly when THIS worker's phase 2 began
    barrier.wait()  # => a SECOND rendezvous -- phase 2 is also fully complete before ANY worker moves on
 
 
def run_phased_computation() -> tuple[list[float], list[float]]:
    barrier = threading.Barrier(WORKER_COUNT)  # => barrier: releases ONLY once all WORKER_COUNT threads arrive
    phase1_end: list[float] = [0.0] * WORKER_COUNT  # => phase1_end[i]: when worker i finished phase 1
    phase2_start: list[float] = [0.0] * WORKER_COUNT  # => phase2_start[i]: when worker i began phase 2
    workers = [
        # => all workers share the SAME barrier -- that shared object IS the synchronization point
        threading.Thread(target=phased_worker, args=(i, barrier, phase1_end, phase2_start))
        for i in range(WORKER_COUNT)  # => one thread per worker, each with its OWN staggered phase-1 delay
    ]  # => workers: exactly WORKER_COUNT Thread objects, not yet started
    for w in workers:  # => starts every worker
        w.start()  # => each begins its own (different-length) phase 1 immediately
    for w in workers:  # => waits for every worker to finish BOTH phases
        w.join()  # => join() blocks until that worker's phased_worker() call returns
    return phase1_end, phase2_start  # => everything needed to verify the phase-ordering invariant
 
 
if __name__ == "__main__":  # => module entry point
    phase1_end, phase2_start = run_phased_computation()  # => drives the whole three-phase scenario to completion
    print(f"phase1_end={[round(t, 3) for t in phase1_end]}")  # => Output: phase1_end=[<staggered times>]
    print(f"phase2_start={[round(t, 3) for t in phase2_start]}")  # => Output: phase2_start=[<roughly IDENTICAL times>]
 
    latest_phase1_end = max(phase1_end)  # => latest_phase1_end: the SLOWEST worker's phase-1 finish time
    earliest_phase2_start = min(phase2_start)  # => earliest_phase2_start: the FASTEST worker's phase-2 start time
 
    # => Without the Barrier, faster workers (worker 0's near-instant phase 1) would race ahead into
    # => phase 2 long before the slowest worker (worker 3's 0.03s-delayed phase 1) finishes ITS phase 1
    # => -- exactly the kind of bug phased parallel algorithms (co-28) can't tolerate. `barrier.wait()`
    # => guarantees NO worker's phase 2 can start until EVERY worker has finished phase 1 (co-15): the
    # => earliest possible phase-2 start is bounded below by the LATEST phase-1 finish, across ALL workers.
    assert earliest_phase2_start >= latest_phase1_end  # => confirms the phase boundary was genuinely enforced
    print("ex-75 OK")  # => Output: ex-75 OK

phase1_end=[34556.502, 34556.515, 34556.53, 34556.542]
phase2_start=[34556.542, 34556.542, 34556.542, 34556.542]
ex-75 OK

"""Example 75: pytest verification for `Barrier`-Synchronized Phased Computation."""
 
from example import run_phased_computation
 
 
def test_no_worker_starts_phase_two_before_every_worker_finishes_phase_one() -> None:
    phase1_end, phase2_start = run_phased_computation()
    assert min(phase2_start) >= max(phase1_end)  # => the barrier genuinely enforced the phase boundary
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 76: I/O-Bound Work, Benchmarked Three Ways -- Serial vs Threads vs `asyncio`."""
 
import asyncio  # => co-26: the event-loop approach to concurrent I/O
import time  # => measures wall time across all three approaches
from concurrent.futures import ThreadPoolExecutor  # => co-23: the thread-pool approach
 
TASK_COUNT = 6  # => how many independent "I/O calls" to make, all THREE ways
SLEEP_SECONDS = 0.08  # => how long each simulated I/O call takes
 
 
def sync_io_task(n: int) -> int:  # => n: unused identifier -- only the delay matters
    time.sleep(SLEEP_SECONDS)  # => a blocking I/O simulation -- releases the GIL for real (co-05)
    return n * n  # => trivial result, checkable across all three approaches
 
 
async def async_io_task(n: int) -> int:  # => the cooperative-async equivalent of sync_io_task
    await asyncio.sleep(SLEEP_SECONDS)  # => the SAME delay, but yielded cooperatively instead of blocking
    return n * n  # => the identical result-shape as the synchronous version
 
 
def run_serial() -> tuple[float, list[int]]:
    start = time.perf_counter()  # => start: wall time before the ONE-AT-A-TIME loop
    results = [sync_io_task(n) for n in range(TASK_COUNT)]  # => runs every call strictly in sequence
    return time.perf_counter() - start, results  # => elapsed time AND the correctness-checkable results
 
 
def run_threads() -> tuple[float, list[int]]:
    start = time.perf_counter()  # => start: wall time before the pooled run
    with ThreadPoolExecutor(max_workers=TASK_COUNT) as pool:  # => enough workers to run every task at once
        results = list(pool.map(sync_io_task, range(TASK_COUNT)))  # => all TASK_COUNT calls OVERLAP their sleeps
    return time.perf_counter() - start, results  # => elapsed time AND results
 
 
async def run_asyncio() -> tuple[float, list[int]]:
    start = time.perf_counter()  # => start: wall time before the coroutine-based run
    results = await asyncio.gather(*(async_io_task(n) for n in range(TASK_COUNT)))  # => all sleeps overlap on ONE thread
    return time.perf_counter() - start, list(results)  # => elapsed time AND results
 
 
if __name__ == "__main__":  # => module entry point
    serial_time, serial_results = run_serial()  # => serial_time: the single-threaded baseline
    threads_time, threads_results = run_threads()  # => threads_time: the multi-threaded, I/O-overlapping run
    asyncio_time, asyncio_results = asyncio.run(run_asyncio())  # => asyncio_time: the single-threaded, cooperative run
 
    print(f"serial={serial_time:.2f}s threads={threads_time:.2f}s asyncio={asyncio_time:.2f}s")
    # => Output: serial=~0.48s threads=~0.08s asyncio=~0.08s
 
    expected = [n * n for n in range(TASK_COUNT)]  # => expected: the ground-truth result, identical across all three
 
    # => I/O-bound work is where BOTH concurrency models genuinely help (co-05): threads overlap their
    # => blocking `time.sleep` calls because each one releases the GIL while waiting (co-23), and
    # => `asyncio` overlaps its cooperative `await asyncio.sleep` calls on a SINGLE thread via the event
    # => loop (co-26). Both beat the serial baseline by roughly TASK_COUNT-fold here, because the total
    # => wall time collapses to about ONE task's duration instead of the sum of every task's duration.
    assert threads_time < serial_time / 2  # => confirms threads delivered a substantial I/O speedup
    assert asyncio_time < serial_time / 2  # => confirms asyncio ALSO delivered a substantial I/O speedup
    assert serial_results == expected  # => confirms the serial run's results are correct
    assert threads_results == expected  # => confirms the threaded run's results are ALSO correct
    assert asyncio_results == expected  # => confirms the asyncio run's results are ALSO correct
    print("ex-76 OK")  # => Output: ex-76 OK

serial=0.52s threads=0.09s asyncio=0.08s
ex-76 OK

"""Example 76: pytest verification for I/O Benchmarked Three Ways."""
 
import asyncio
 
from example import TASK_COUNT, run_asyncio, run_serial, run_threads
 
 
def test_both_concurrent_forms_beat_serial_on_io() -> None:
    serial_time, serial_results = run_serial()
    threads_time, threads_results = run_threads()
    asyncio_time, asyncio_results = asyncio.run(run_asyncio())
 
    expected = [n * n for n in range(TASK_COUNT)]
    assert threads_time < serial_time / 2  # => threads deliver a substantial I/O speedup
    assert asyncio_time < serial_time / 2  # => asyncio also delivers a substantial I/O speedup
    assert serial_results == expected
    assert threads_results == expected
    assert asyncio_results == expected
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 77: CPU-Bound Work, Benchmarked Three Ways -- Only Processes Actually Win."""
 
import time  # => measures wall time across all three approaches
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor  # => co-24 vs co-23
 
TASK_COUNT = 4  # => how many independent CPU-bound tasks, all THREE ways
ITERATIONS = 8_000_000  # => tuned so the difference between approaches is clearly measurable
 
 
def cpu_task(n: int) -> int:  # => a top-level function -- REQUIRED so ProcessPoolExecutor can pickle it
    total = 0  # => accumulator -- forces real interpreter bytecode execution
    for i in range(n):  # => a tight loop -- the shape a GIL serializes across threads (co-03)
        total += i  # => trivial arithmetic; only the TIME this takes matters here
    return total  # => the actual value is irrelevant to this example
 
 
def run_serial() -> float:
    start = time.perf_counter()  # => start: wall time before the ONE-AT-A-TIME loop
    for _ in range(TASK_COUNT):  # => runs every task strictly in sequence, on this ONE thread
        cpu_task(ITERATIONS)  # => no overlap possible -- this IS the baseline every other approach is measured against
    return time.perf_counter() - start  # => serial_time: the single-threaded baseline
 
 
def run_threads() -> float:
    start = time.perf_counter()  # => start: wall time before the thread-pooled run
    with ThreadPoolExecutor(max_workers=TASK_COUNT) as pool:  # => TASK_COUNT threads, ONE shared interpreter
        list(pool.map(cpu_task, [ITERATIONS] * TASK_COUNT))  # => serialized by the GIL despite "running concurrently"
    return time.perf_counter() - start  # => threads_time: expected close to serial_time (co-05 does NOT apply here)
 
 
def run_processes() -> float:
    start = time.perf_counter()  # => start: wall time before the process-pooled run
    with ProcessPoolExecutor(max_workers=TASK_COUNT) as pool:  # => TASK_COUNT processes, EACH with its OWN GIL
        list(pool.map(cpu_task, [ITERATIONS] * TASK_COUNT))  # => genuinely runs across multiple cores
    return time.perf_counter() - start  # => processes_time: expected substantially FASTER than serial_time
 
 
if __name__ == "__main__":  # => module entry point
    serial_time = run_serial()  # => serial_time: the single-threaded baseline
    threads_time = run_threads()  # => threads_time: the multi-threaded CPU run
    processes_time = run_processes()  # => processes_time: the multi-process CPU run
 
    print(f"serial={serial_time:.2f}s threads={threads_time:.2f}s processes={processes_time:.2f}s")
    # => Output: serial=~1.0s threads=~1.0s processes=~0.35s
 
    # => Unlike ex-76's I/O-bound benchmark, where BOTH concurrency models won, CPU-bound work behaves
    # => very differently: threads provide essentially NO speedup because the GIL serializes their Python
    # => bytecode regardless of how many threads exist (co-03, co-05 doesn't apply -- there's no I/O wait
    # => to release the GIL during). Only `ProcessPoolExecutor` (co-24) genuinely wins, because each
    # => process gets its OWN interpreter and OWN GIL, sidestepping the single-GIL bottleneck entirely.
    assert threads_time > serial_time * 0.7  # => confirms threads did NOT deliver a meaningful CPU speedup
    assert processes_time < serial_time * 0.7  # => confirms processes DID deliver a meaningful CPU speedup
    print("ex-77 OK")  # => Output: ex-77 OK

serial=0.98s threads=1.03s processes=0.33s
ex-77 OK

"""Example 77: pytest verification for CPU Benchmarked Three Ways."""
 
from example import run_processes, run_serial, run_threads
 
 
def test_only_processes_beat_serial_on_cpu_bound_work() -> None:
    serial_time = run_serial()
    threads_time = run_threads()
    processes_time = run_processes()
 
    assert threads_time > serial_time * 0.7  # => the GIL keeps threads from meaningfully speeding up CPU work
    assert processes_time < serial_time * 0.7  # => separate processes DO deliver a meaningful CPU speedup
 
 
# => Run: pytest -- Output: 1 passed