Hands-on lab — Day 96: Concurrency and async Basics

#!/usr/bin/env python3
"""Waiting work: twenty requests, three ways, timed.

Run:  python3 examples/01_waiting.py

Every request goes to a fixture server on 127.0.0.1 that sleeps exactly
0.100 seconds before answering. No network is involved and no rate limit
can distort the result: the work is *waiting*, and the amount of waiting
is a number this script chose.

The question the script answers is the only question that matters when you
are picking a concurrency model: **is this work waiting, or is it
computing?** This half is waiting. Watch what threads do to it.
"""

from __future__ import annotations

import asyncio
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import labkit  # noqa: E402

REQUESTS = labkit.WAITING_REQUESTS
WORKERS = 20
REPEATS = 3


def run_sequential(base: str) -> list[str]:
    """One at a time. The simplest thing, and the slowest by a mile."""
    return [labkit.fetch(url) for url in labkit.urls(base, REQUESTS)]


def run_threaded(base: str) -> list[str]:
    """Twenty threads, each blocked in a socket read.

    A blocked socket read releases the interpreter lock, so nineteen other
    threads are free to run while any one of them waits. That single fact is
    why threads help here and will not help the next script along.

    `executor.map` returns results in the order the inputs were given, not
    the order they finished, which is usually what you wanted anyway.
    """
    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        return list(pool.map(labkit.fetch, labkit.urls(base, REQUESTS)))


async def _gather(base: str) -> list[str]:
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(labkit.fetch_async(url)) for url in labkit.urls(base, REQUESTS)]
    return [task.result() for task in tasks]


def run_asyncio(base: str) -> list[str]:
    """One thread, one event loop, twenty coroutines suspended at `await`.

    `asyncio.run` creates the loop, runs the coroutine to completion and
    closes the loop. A TaskGroup (Python 3.11 and later) is the modern way
    to say "these run together, and if one raises, cancel the rest".
    """
    return asyncio.run(_gather(base))


def main() -> int:
    print("Day 096 — waiting work: 20 requests, three ways")
    print(f"fixture server sleeps {labkit.DEFAULT_DELAY:.3f}s per request; "
          f"{REQUESTS} requests; thread pool of {WORKERS}; {REPEATS} runs each")
    print()

    with labkit.fixture_server() as base:
        # A first request that is not measured, so that the cost of importing
        # ssl-free urllib machinery and warming the server does not land on
        # the sequential column and flatter the other two.
        labkit.fetch(base + "/warmup")

        sequential_samples, sequential_bodies = labkit.repeat(
            lambda: run_sequential(base), REPEATS
        )
        threaded_samples, threaded_bodies = labkit.repeat(lambda: run_threaded(base), REPEATS)
        asyncio_samples, asyncio_bodies = labkit.repeat(lambda: run_asyncio(base), REPEATS)

    print("timings")
    sequential = labkit.report("sequential (one at a time)", sequential_samples)
    threaded = labkit.report(f"threads (ThreadPoolExecutor {WORKERS})", threaded_samples)
    evented = labkit.report("asyncio (one thread, one loop)", asyncio_samples)
    print()

    floor = REQUESTS * labkit.DEFAULT_DELAY
    print("what the numbers mean")
    print(f"  {REQUESTS} requests x {labkit.DEFAULT_DELAY:.3f}s of waiting = {floor:.3f}s of "
          "waiting in total.")
    print("  Sequential pays all of it end to end. The other two overlap it, so the floor")
    print(f"  becomes roughly one request: {labkit.DEFAULT_DELAY:.3f}s plus overhead.")
    print()

    print("correctness first — a fast wrong answer is not an answer")
    for name, bodies in (
        ("sequential", sequential_bodies),
        ("threads", threaded_bodies),
        ("asyncio", asyncio_bodies),
    ):
        assert isinstance(bodies, list)
        ok = len(bodies) == REQUESTS and all("waited 0.100s" in b for b in bodies)
        print(f"  {name:<12} {len(bodies)} bodies, all well formed: {'yes' if ok else 'no'}")
    print(f"  and the order is preserved: first body is for {sequential_bodies[0].split()[-1]}")
    print()

    labkit.result_line("waiting_sequential_s", sequential)
    labkit.result_line("waiting_threaded_s", threaded)
    labkit.result_line("waiting_asyncio_s", evented)
    labkit.result_line("waiting_threaded_speedup", sequential / threaded)
    labkit.result_line("waiting_asyncio_speedup", sequential / evented)
    labkit.shape_line(
        "threads_help_waiting_work",
        sequential / threaded >= 4.0,
        f"threads are {sequential / threaded:.1f}x faster than sequential; the claim is >= 4x",
    )
    labkit.shape_line(
        "asyncio_helps_waiting_work",
        sequential / evented >= 4.0,
        f"asyncio is {sequential / evented:.1f}x faster than sequential; the claim is >= 4x",
    )
    print()
    print("These seconds are one machine on one day. The SHAPE is what travels:")
    print("waiting work overlaps, so both threads and an event loop collapse it.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Computing work: four prime counts, four ways, timed.

Run:  python3 examples/02_computing.py

Same code shape as 01_waiting.py. Same number of runs. The only thing that
changed is what the work *is* — and the answer flips completely.

This is the script that costs people weeks when they skip it. Threads made
the waiting work fourteen times faster in the previous script, so threads
get reached for again here, and here they do nothing at all. The reason is
the interpreter lock, and the fix is processes.
"""

from __future__ import annotations

import asyncio
import os
import sys
import sysconfig
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import labkit  # noqa: E402

TASKS = labkit.COMPUTING_TASKS
LIMIT = labkit.DEFAULT_PRIME_LIMIT
REPEATS = 3
LIMITS = [LIMIT] * TASKS


def run_sequential() -> list[int]:
    return [labkit.count_primes(limit) for limit in LIMITS]


def run_threaded() -> list[int]:
    """The same code as the fast version in 01_waiting.py. Watch it not help."""
    with ThreadPoolExecutor(max_workers=TASKS) as pool:
        return list(pool.map(labkit.count_primes, LIMITS))


def run_processes() -> list[int]:
    """One character different from the line above, and a different machine underneath.

    Each worker is a separate operating-system process with its own
    interpreter and its own lock, so four of them genuinely run at once on
    four cores. The price is paid at the edges: starting the processes, and
    pickling the arguments and results across them.
    """
    with ProcessPoolExecutor(max_workers=TASKS) as pool:
        return list(pool.map(labkit.count_primes, LIMITS))


async def _await_them() -> list[int]:
    """Deliberately wrong, and instructive.

    `async def` does not make anything concurrent. There is no `await` inside
    count_primes and there could not be: it never waits for anything. So the
    event loop runs each call to completion before it looks at the next one,
    and asyncio delivers exactly sequential timing with extra ceremony.
    """
    return [labkit.count_primes(limit) for limit in LIMITS]


def run_asyncio() -> list[int]:
    return asyncio.run(_await_them())


def main() -> int:
    free_threaded = sysconfig.get_config_var("Py_GIL_DISABLED")
    print("Day 096 — computing work: 4 prime counts, four ways")
    print(f"python {sys.version.split()[0]}   cpu_count {os.cpu_count()}   "
          f"Py_GIL_DISABLED {free_threaded!r}")
    print(f"each task counts the primes below {LIMIT:,} by trial division; "
          f"{TASKS} tasks; {REPEATS} runs each")
    if not free_threaded:
        print("This interpreter was built WITH the global interpreter lock "
              "(Py_GIL_DISABLED is 0), which is what the numbers below reflect.")
    else:
        print("This interpreter is a free-threaded build (Py_GIL_DISABLED is 1); "
              "the threaded row below is not what a lock-holding build produces.")
    print()

    sequential_samples, sequential_result = labkit.repeat(run_sequential, REPEATS)
    threaded_samples, threaded_result = labkit.repeat(run_threaded, REPEATS)
    process_samples, process_result = labkit.repeat(run_processes, REPEATS)
    asyncio_samples, asyncio_result = labkit.repeat(run_asyncio, REPEATS)

    print("timings")
    sequential = labkit.report("sequential (one at a time)", sequential_samples)
    threaded = labkit.report(f"threads (ThreadPoolExecutor {TASKS})", threaded_samples)
    processes = labkit.report(f"processes (ProcessPoolExecutor {TASKS})", process_samples)
    evented = labkit.report("asyncio (one thread, one loop)", asyncio_samples)
    print()

    print("correctness first")
    expected = [78498 if LIMIT == 1_000_000 else labkit.count_primes(LIMIT)] * TASKS
    for name, got in (
        ("sequential", sequential_result),
        ("threads", threaded_result),
        ("processes", process_result),
        ("asyncio", asyncio_result),
    ):
        print(f"  {name:<12} {got}  correct: {'yes' if got == expected else 'no'}")
    print(f"  there really are {expected[0]:,} primes below {LIMIT:,}")
    print()

    labkit.result_line("computing_sequential_s", sequential)
    labkit.result_line("computing_threaded_s", threaded)
    labkit.result_line("computing_processes_s", processes)
    labkit.result_line("computing_asyncio_s", evented)
    labkit.result_line("computing_threaded_speedup", sequential / threaded)
    labkit.result_line("computing_processes_speedup", sequential / processes)
    labkit.result_line("computing_asyncio_speedup", sequential / evented)
    labkit.shape_line(
        "threads_do_not_help_computing_work",
        sequential / threaded < 1.5,
        f"threads are {sequential / threaded:.2f}x sequential; the claim is < 1.5x",
    )
    labkit.shape_line(
        "processes_do_help_computing_work",
        sequential / processes >= 1.5,
        f"processes are {sequential / processes:.2f}x sequential; the claim is >= 1.5x",
    )
    labkit.shape_line(
        "asyncio_does_not_help_computing_work",
        sequential / evented < 1.5,
        f"asyncio is {sequential / evented:.2f}x sequential; the claim is < 1.5x",
    )
    print()
    print("The one-line rule, earned rather than asserted:")
    print("  waiting work  -> threads or an event loop; the waiting overlaps")
    print("  computing work-> processes; nothing else has more than one interpreter lock")
    print()
    print("Note what processes did NOT give you: a 4x speedup from 4 workers. Starting")
    print("them, pickling to them and pickling back are real costs, and they are paid")
    print("whether or not the work was big enough to deserve them.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""The one rule you must not break, broken on purpose and then repaired.

Run:  python3 examples/03_blocking_coroutine.py

**Never call a blocking function inside a coroutine.**

An event loop is one thread running one callback at a time. A coroutine
gives that thread back at every `await` and at no other moment. So a
coroutine that calls something which blocks — `time.sleep`, a synchronous
HTTP client, a database driver that is not async, a file read on a slow
disk, `requests.get` — does not suspend. It holds the only thread there
is, and every other task on the loop stops dead until it returns.

The failure is quiet. Nothing raises. The program still produces correct
answers. It is simply serial while looking concurrent, which is why this
bug survives code review and is found in production by a latency graph.

The script measures four things:
  1. five blocking sleeps inside coroutines, gathered
  2. the same five with `await asyncio.sleep`, gathered
  3. the same five blocking calls handed to `asyncio.to_thread`
  4. what the blocking coroutine does to an UNRELATED task on the same loop
"""

from __future__ import annotations

import asyncio
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import labkit  # noqa: E402

TASKS = 5
NAP = 0.20


# ---------------------------------------------------------------------------
# 1. Broken: `async def` around a blocking call.
# ---------------------------------------------------------------------------


async def blocking_task(index: int) -> int:
    # time.sleep does not yield to the event loop. It parks the whole thread.
    time.sleep(NAP)
    return index


async def all_blocking() -> list[int]:
    return list(await asyncio.gather(*(blocking_task(i) for i in range(TASKS))))


# ---------------------------------------------------------------------------
# 2. Correct: an awaitable sleep.
# ---------------------------------------------------------------------------


async def awaiting_task(index: int) -> int:
    # asyncio.sleep is a coroutine. `await` is the pause point: the loop takes
    # the thread back here and gives it to whichever task is ready next.
    await asyncio.sleep(NAP)
    return index


async def all_awaiting() -> list[int]:
    return list(await asyncio.gather(*(awaiting_task(i) for i in range(TASKS))))


# ---------------------------------------------------------------------------
# 3. The repair for code you cannot change: push it to a thread.
# ---------------------------------------------------------------------------


async def offloaded_task(index: int) -> int:
    # asyncio.to_thread runs the blocking call in a worker thread and gives
    # you a coroutine to await. The blocking still happens — it just happens
    # somewhere that is allowed to block. This is the fix for a synchronous
    # library you do not own and cannot rewrite.
    await asyncio.to_thread(time.sleep, NAP)
    return index


async def all_offloaded() -> list[int]:
    return list(await asyncio.gather(*(offloaded_task(i) for i in range(TASKS))))


# ---------------------------------------------------------------------------
# 4. The collateral damage: what it does to a task that is not even involved.
# ---------------------------------------------------------------------------


async def heartbeat(stop_after: float) -> list[float]:
    """Tick every 10ms and record when each tick actually happened.

    A healthy loop produces ticks about 10ms apart. A loop with a blocked
    task produces a gap the size of the blockage. This is the shape you look
    for in a real service: not an error, a gap.
    """
    started = time.perf_counter()
    ticks: list[float] = []
    while time.perf_counter() - started < stop_after:
        await asyncio.sleep(0.01)
        ticks.append(time.perf_counter() - started)
    return ticks


async def measure_starvation(block: bool) -> float:
    """Return the largest gap between heartbeats while one task runs.

    The `await asyncio.sleep(0.03)` is load-bearing and is worth a sentence,
    because leaving it out is itself a classic asyncio mistake. `create_task`
    only *schedules* the heartbeat; it does not start it. Nothing on a loop
    starts until the currently running coroutine gives the thread back at an
    `await`. Without that line the heartbeat would not have ticked once
    before the blocking call began, and the gap being measured would not
    exist yet.
    """
    beat = asyncio.create_task(heartbeat(NAP * 2))
    await asyncio.sleep(0.03)
    if block:
        await blocking_task(0)
    else:
        await awaiting_task(0)
    ticks = await beat
    gaps = [second - first for first, second in zip(ticks, ticks[1:])]
    return max(gaps) if gaps else 0.0


def main() -> int:
    print("Day 096 — a blocking call inside a coroutine")
    print(f"{TASKS} tasks, each waiting {NAP:.2f}s, gathered on one event loop")
    print()

    blocking_samples, blocking_result = labkit.repeat(lambda: asyncio.run(all_blocking()), 3)
    awaiting_samples, awaiting_result = labkit.repeat(lambda: asyncio.run(all_awaiting()), 3)
    offload_samples, offload_result = labkit.repeat(lambda: asyncio.run(all_offloaded()), 3)

    print("timings")
    blocking = labkit.report("async def + time.sleep  (broken)", blocking_samples)
    awaiting = labkit.report("async def + await asyncio.sleep", awaiting_samples)
    offloaded = labkit.report("async def + asyncio.to_thread", offload_samples)
    print()

    serial_floor = TASKS * NAP
    print("what the numbers mean")
    print(f"  {TASKS} tasks x {NAP:.2f}s = {serial_floor:.2f}s if nothing overlaps.")
    print(f"  The broken version took {blocking:.3f}s, which is the serial floor. It ran")
    print("  one task at a time while the code said gather. Nothing raised.")
    print(f"  Both repairs took about {NAP:.2f}s, which is one task's worth of waiting.")
    print()

    print("correctness — all three produce the same answers, which is the trap")
    for name, got in (
        ("blocking", blocking_result),
        ("awaiting", awaiting_result),
        ("to_thread", offload_result),
    ):
        print(f"  {name:<11} {got}  correct: {'yes' if got == list(range(TASKS)) else 'no'}")
    print()

    print("collateral damage to an unrelated task on the same loop")
    blocked_gap = asyncio.run(measure_starvation(block=True))
    healthy_gap = asyncio.run(measure_starvation(block=False))
    print(f"  largest heartbeat gap while a coroutine BLOCKED : {blocked_gap * 1000:7.1f} ms")
    print(f"  largest heartbeat gap while a coroutine AWAITED : {healthy_gap * 1000:7.1f} ms")
    print("  The heartbeat asked for a tick every 10 ms and had nothing to do with the")
    print("  sleeping task. It was starved anyway, because there is only one thread.")
    print()

    labkit.result_line("blocking_gathered_s", blocking)
    labkit.result_line("awaiting_gathered_s", awaiting)
    labkit.result_line("to_thread_gathered_s", offloaded)
    labkit.result_line("blocking_vs_await_speedup", blocking / awaiting)
    labkit.result_line("blocking_vs_to_thread_speedup", blocking / offloaded)
    labkit.result_line("blocked_heartbeat_gap_ms", blocked_gap * 1000)
    labkit.result_line("healthy_heartbeat_gap_ms", healthy_gap * 1000)
    labkit.shape_line(
        "blocking_in_a_coroutine_serialises_the_loop",
        blocking >= serial_floor * 0.9,
        f"took {blocking:.3f}s against a serial floor of {serial_floor:.2f}s",
    )
    labkit.shape_line(
        "await_asyncio_sleep_repairs_it",
        blocking / awaiting >= 2.5,
        f"awaiting is {blocking / awaiting:.1f}x faster; the claim is >= 2.5x",
    )
    labkit.shape_line(
        "to_thread_repairs_it_for_code_you_cannot_change",
        blocking / offloaded >= 2.5,
        f"to_thread is {blocking / offloaded:.1f}x faster; the claim is >= 2.5x",
    )
    labkit.shape_line(
        "blocking_starves_an_unrelated_task",
        blocked_gap >= healthy_gap * 3,
        f"gap {blocked_gap * 1000:.1f} ms against {healthy_gap * 1000:.1f} ms when healthy",
    )
    print()
    print("How to find this in code you did not write: any call inside an `async def`")
    print("that is not preceded by `await` and is not obviously pure computation is a")
    print("suspect. `time.sleep`, `requests.get`, `open(...).read()` on a network mount,")
    print("and every synchronous database driver are the usual four.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Shared state: a counter that loses increments, and three ways to keep them.

Run:  python3 examples/04_race.py

The interpreter lock protects the INTERPRETER — its reference counts, its
internal structures — so that a data race can never corrupt Python's own
memory. It does not protect YOUR data structures. `total = total + 1` is a
read, an addition and a write, and if a thread switch lands between the
read and the write, one increment is silently overwritten by another.

There is an honest wrinkle in reproducing that, and this script does not
hide it. Read the output: on the interpreter this lab was written on, the
naive version at the default settings does not lose a single increment.
That is not evidence the bug is gone. It is evidence the window is narrow.
The script therefore narrows nothing and widens nothing about the CODE —
it only shortens the interval at which the interpreter considers switching
threads, from the default 5 milliseconds to 1 microsecond, which makes the
same race land every single run instead of once in a very long while.

A bug you cannot reproduce is still a bug. It is just a worse bug.
"""

from __future__ import annotations

import queue
import sys
import threading
from contextlib import contextmanager

THREADS = 8
PER_THREAD = 50_000
EXPECTED = THREADS * PER_THREAD
TIGHT_INTERVAL = 1e-6


@contextmanager
def switch_interval(seconds: float | None):
    """Temporarily change how often the interpreter considers a thread switch.

    sys.setswitchinterval is process-wide, so it is restored in a finally
    block. The default is 0.005 seconds: a thread that neither blocks nor
    calls out gets roughly five milliseconds before the interpreter asks
    whether somebody else should have a turn.
    """
    previous = sys.getswitchinterval()
    try:
        if seconds is not None:
            sys.setswitchinterval(seconds)
        yield previous
    finally:
        sys.setswitchinterval(previous)


class SharedCounter:
    """A counter shared by every thread. Read, add, write — three steps."""

    def __init__(self) -> None:
        self.value = 0

    def read(self) -> int:
        return self.value

    def write(self, value: int) -> None:
        self.value = value


def bump_unsafely(counter: SharedCounter, times: int) -> None:
    # The read and the write are two separate operations with a gap between
    # them. Whatever another thread does in that gap is lost.
    for _ in range(times):
        counter.write(counter.read() + 1)


def bump_with_lock(counter: SharedCounter, times: int, lock: threading.Lock) -> None:
    # `with lock:` makes read-add-write one indivisible step. Only one thread
    # can be between the two lines at a time, so nothing can be overwritten.
    for _ in range(times):
        with lock:
            counter.write(counter.read() + 1)


def run_threads(target, *args) -> None:
    workers = [threading.Thread(target=target, args=args) for _ in range(THREADS)]
    for worker in workers:
        worker.start()
    for worker in workers:
        worker.join()


def unsafe_total(interval: float | None) -> int:
    counter = SharedCounter()
    with switch_interval(interval):
        run_threads(bump_unsafely, counter, PER_THREAD)
    return counter.value


def locked_total(interval: float | None) -> int:
    counter = SharedCounter()
    lock = threading.Lock()
    with switch_interval(interval):
        run_threads(bump_with_lock, counter, PER_THREAD, lock)
    return counter.value


def queued_total(interval: float | None) -> int:
    """The answer that is usually better than a lock: do not share the state.

    Every worker sends its own subtotal down a queue. One consumer — here,
    the main thread after the join — adds them up. Nothing is shared and
    mutated, so there is nothing to protect and no lock to forget. A
    queue.Queue is internally locked so you do not have to be.
    """
    outbox: queue.Queue[int] = queue.Queue()

    def worker() -> None:
        subtotal = 0
        for _ in range(PER_THREAD):
            subtotal += 1  # a local variable: no other thread can see it
        outbox.put(subtotal)

    with switch_interval(interval):
        workers = [threading.Thread(target=worker) for _ in range(THREADS)]
        for w in workers:
            w.start()
        for w in workers:
            w.join()
    total = 0
    while not outbox.empty():
        total += outbox.get()
    return total


def deadlock_demonstration() -> tuple[bool, bool]:
    """Two locks, two threads, opposite orders. Returns (deadlocked, fixed_ok).

    This is a real deadlock and it would hang forever, so each thread takes
    its second lock with a timeout. The timeout is a detector, not a fix: in
    production it turns a hang into a mysterious slow path. The fix is the
    second half — every thread takes the locks in the SAME order, so a cycle
    of waiting cannot form.
    """
    first, second = threading.Lock(), threading.Lock()
    gate = threading.Barrier(2)
    stuck: list[bool] = []

    def grab(outer: threading.Lock, inner: threading.Lock) -> None:
        with outer:
            gate.wait()  # guarantee both threads hold one lock before either asks for two
            got = inner.acquire(timeout=0.5)
            stuck.append(not got)
            if got:
                inner.release()

    a = threading.Thread(target=grab, args=(first, second))
    b = threading.Thread(target=grab, args=(second, first))  # opposite order: the bug
    a.start()
    b.start()
    a.join()
    b.join()
    deadlocked = any(stuck)

    # Same two locks, same two threads, one rule: always first then second.
    #
    # Note there is no barrier here, and that is not an oversight — it is the
    # second lesson of this function. Adding one would make the FIXED version
    # hang: a thread holding `first` would wait at the barrier for a thread
    # that cannot reach the barrier because it is waiting for `first`. Forcing
    # an interleaving is itself a way to build a deadlock.
    stuck2: list[bool] = []

    def ordered() -> None:
        with first:
            got = second.acquire(timeout=0.5)
            stuck2.append(not got)
            if got:
                second.release()

    c = threading.Thread(target=ordered)
    d = threading.Thread(target=ordered)
    c.start()
    d.start()
    c.join()
    d.join()
    return deadlocked, not any(stuck2)


def main() -> int:
    print("Day 096 — a counter that loses increments")
    print(f"{THREADS} threads x {PER_THREAD:,} increments each; expected total {EXPECTED:,}")
    print(f"default switch interval on this interpreter: {sys.getswitchinterval()} s")
    print()

    print("1. the naive counter at the interpreter's DEFAULT switch interval")
    default_runs = [unsafe_total(None) for _ in range(3)]
    for index, got in enumerate(default_runs, 1):
        print(f"   run {index}: {got:,}   lost {EXPECTED - got:,}")
    print("   Report what you see, not what the textbook says. On this machine the")
    print("   read-add-write finishes well inside one thread's 5 ms slice, so the race")
    print("   almost never lands. That is a narrow window, not a safe program.")
    print()

    print(f"2. the same code with the switch interval at {TIGHT_INTERVAL} s")
    tight_runs = [unsafe_total(TIGHT_INTERVAL) for _ in range(3)]
    for index, got in enumerate(tight_runs, 1):
        print(f"   run {index}: {got:,}   lost {EXPECTED - got:,}")
    print("   The code did not change. Only the frequency of thread switches did.")
    print("   Every increment lost here was an increment that could be lost at the")
    print("   default setting too, on a busier machine, under a longer run, one day.")
    print()

    print("3. the same counter, one lock")
    locked_runs = [locked_total(TIGHT_INTERVAL) for _ in range(3)]
    for index, got in enumerate(locked_runs, 1):
        print(f"   run {index}: {got:,}   lost {EXPECTED - got:,}")
    print("   Exact, at the switch interval that broke the version above.")
    print()

    print("4. the answer that is usually better than a lock: stop sharing")
    queued_runs = [queued_total(TIGHT_INTERVAL) for _ in range(3)]
    for index, got in enumerate(queued_runs, 1):
        print(f"   run {index}: {got:,}   lost {EXPECTED - got:,}")
    print("   Each worker counts into a local variable and posts one subtotal to a")
    print("   queue. There is no shared mutable state, so there is no lock to forget,")
    print("   no lock ordering to get wrong, and nothing to deadlock.")
    print()

    print("5. deadlock, in eight lines")
    deadlocked, ordered_ok = deadlock_demonstration()
    print(f"   two locks taken in opposite orders  -> deadlocked: {'yes' if deadlocked else 'no'}")
    print(f"   the same two taken in the same order -> completed: "
          f"{'yes' if ordered_ok else 'no'}")
    print("   Each thread held one lock and waited for the other. Nothing was busy;")
    print("   nothing errored; the program simply stopped. The timeout above is a")
    print("   detector. The fix is the ordering rule.")
    print()

    worst_default = EXPECTED - min(default_runs)
    worst_tight = EXPECTED - min(tight_runs)
    print(f"RESULT race_lost_at_default_interval {worst_default}")
    print(f"RESULT race_lost_at_tight_interval {worst_tight}")
    print(f"RESULT locked_total {min(locked_runs)}")
    print(f"RESULT queued_total {min(queued_runs)}")
    print(f"RESULT expected_total {EXPECTED}")
    print(f"SHAPE unlocked_counter_loses_increments "
          f"{'yes' if worst_tight > 0 else 'no'}   (lost {worst_tight:,} of {EXPECTED:,})")
    print(f"SHAPE locked_counter_loses_nothing "
          f"{'yes' if all(r == EXPECTED for r in locked_runs) else 'no'}   "
          f"(3 runs, all exactly {EXPECTED:,})")
    print(f"SHAPE queue_version_loses_nothing "
          f"{'yes' if all(r == EXPECTED for r in queued_runs) else 'no'}   "
          f"(3 runs, all exactly {EXPECTED:,})")
    print(f"SHAPE opposite_lock_order_deadlocks {'yes' if deadlocked else 'no'}   "
          "(detected with a timeout; it would otherwise hang forever)")
    print(f"SHAPE consistent_lock_order_does_not "
          f"{'yes' if ordered_ok else 'no'}   (same locks, same threads, one rule)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""An event loop in forty lines, so `async`/`await` stops being magic.

Run:  python3 examples/05_scheduler.py

Strip asyncio of its socket handling, its thread pools, its cancellation
machinery and its exception groups, and what is left is this: a queue of
functions that can pause, and a loop that keeps taking the next one.

Python has had functions that can pause since generators arrived in 2001.
A generator runs until `yield`, hands control back to whoever called
`next()` on it, and remembers exactly where it was. That is the entire
mechanism underneath a coroutine. `async def` and `await` are a dedicated
syntax for the same idea with a nicer set of rules; PEP 492 added them in
Python 3.5, and coroutines were built on generators before that.

So: `yield` is our `await`, a generator function is our `async def`, and
the class below is our event loop. Once you have written the loop, the
question "why did my whole server freeze?" answers itself — look at the
loop and ask what happens if `next(task)` takes two seconds to return.
"""

from __future__ import annotations

import sys
from collections import deque
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))


# ---------------------------------------------------------------------------
# The scheduler. This is the whole thing.
# ---------------------------------------------------------------------------

PAUSE = None  # a task yields this to say "give somebody else a turn"


def sleep(ticks: int) -> tuple[str, int]:
    """A task yields this to say "wake me in N ticks". Our asyncio.sleep."""
    return ("sleep", ticks)


class Scheduler:
    """A ready queue, a sleeping list, a clock, and a loop. Nothing else."""

    def __init__(self) -> None:
        self.ready: deque[tuple[str, object]] = deque()
        self.sleeping: list[tuple[int, str, object]] = []
        self.tick = 0
        self.trace: list[str] = []
        self.results: dict[str, object] = {}

    def spawn(self, name: str, task) -> None:
        """Schedule a generator. Note it does not start: it is only queued."""
        self.ready.append((name, task))

    def _wake(self) -> None:
        due = [entry for entry in self.sleeping if entry[0] <= self.tick]
        self.sleeping = [entry for entry in self.sleeping if entry[0] > self.tick]
        for _wake_at, name, task in sorted(due, key=lambda e: e[0]):
            self.ready.append((name, task))

    def run(self) -> dict[str, object]:
        while self.ready or self.sleeping:
            if not self.ready:
                # Everybody is waiting. A real loop would block in select()
                # here until a socket became readable; we jump the clock.
                self.tick = min(entry[0] for entry in self.sleeping)
                self._wake()
            name, task = self.ready.popleft()
            try:
                instruction = next(task)  # resume it; it runs until its next yield
            except StopIteration as finished:
                self.results[name] = finished.value
                continue
            if instruction is PAUSE:
                self.ready.append((name, task))  # straight to the back of the queue
            else:
                _kind, ticks = instruction
                self.sleeping.append((self.tick + ticks, name, task))
            self.tick += 1
            self._wake()
        return self.results

    def step(self, name: str) -> None:
        """Record that a task did one unit of visible work."""
        self.trace.append(f"{name}:{self.tick}")


# ---------------------------------------------------------------------------
# Tasks written against it. `yield` is the pause point.
# ---------------------------------------------------------------------------


def counting_task(loop: Scheduler, name: str, steps: int):
    for _ in range(steps):
        loop.step(name)
        yield PAUSE  # <- this is the `await`. Control goes back to the loop.
    return f"{name} did {steps} step" + ("" if steps == 1 else "s")


def sleeping_task(loop: Scheduler, name: str, nap: int):
    loop.step(f"{name}-start")
    yield sleep(nap)  # <- like `await asyncio.sleep(nap)`
    loop.step(f"{name}-woke")
    return f"{name} slept {nap}"


def greedy_task(loop: Scheduler, name: str, steps: int):
    """The bug, in a form you can see. It never yields until it is finished."""
    for _ in range(steps):
        loop.step(name)
    yield PAUSE
    return f"{name} hogged {steps} steps"


def main() -> int:
    print("Day 096 — a cooperative scheduler built from generators")
    print()

    print("1. three tasks, round-robin")
    loop = Scheduler()
    loop.spawn("alpha", counting_task(loop, "alpha", 3))
    loop.spawn("beta", counting_task(loop, "beta", 2))
    loop.spawn("gamma", counting_task(loop, "gamma", 1))
    results = loop.run()
    order = [entry.split(":")[0] for entry in loop.trace]
    print(f"   trace:  {' '.join(loop.trace)}")
    print(f"   order:  {' '.join(order)}")
    print("   Each task ran ONE step and then gave the loop back. That interleaving")
    print("   is concurrency, and it happened in a single thread with no lock in")
    print("   sight — which is why no increment can be lost between two yields.")
    for name in sorted(results):
        print(f"   returned: {results[name]}")
    print()

    print("2. a task that waits, while the others carry on")
    loop = Scheduler()
    loop.spawn("napper", sleeping_task(loop, "napper", 4))
    loop.spawn("worker", counting_task(loop, "worker", 5))
    loop.run()
    print(f"   trace:  {' '.join(loop.trace)}")
    print("   napper yielded a sleep instruction and left the ready queue entirely.")
    print("   worker had the loop to itself until napper's tick came round. That is")
    print("   exactly what an await on a socket read does in a real event loop.")
    print()

    print("3. the same loop, with one task that refuses to yield")
    loop = Scheduler()
    loop.spawn("greedy", greedy_task(loop, "greedy", 4))
    loop.spawn("polite", counting_task(loop, "polite", 3))
    loop.run()
    print(f"   trace:  {' '.join(loop.trace)}")
    print("   polite did not run once until greedy had finished all four steps.")
    print("   Nothing errored. Nothing was slow. The loop simply never got the")
    print("   thread back, because `next(task)` does not return until the task")
    print("   reaches a yield. That is the blocking call in 03_blocking_coroutine.py,")
    print("   seen from inside the loop rather than from outside it.")
    print()

    expected_round_robin = "alpha beta gamma alpha beta alpha"
    loop = Scheduler()
    loop.spawn("alpha", counting_task(loop, "alpha", 3))
    loop.spawn("beta", counting_task(loop, "beta", 2))
    loop.spawn("gamma", counting_task(loop, "gamma", 1))
    loop.run()
    actual = " ".join(entry.split(":")[0] for entry in loop.trace)
    print(f"RESULT round_robin_order {actual.replace(' ', ',')}")
    print(f"SHAPE scheduler_interleaves_tasks "
          f"{'yes' if actual == expected_round_robin else 'no'}   "
          f"(expected '{expected_round_robin}')")

    loop = Scheduler()
    loop.spawn("greedy", greedy_task(loop, "greedy", 4))
    loop.spawn("polite", counting_task(loop, "polite", 3))
    loop.run()
    hogged = " ".join(entry.split(":")[0] for entry in loop.trace)
    print(f"RESULT greedy_order {hogged.replace(' ', ',')}")
    print(f"SHAPE a_task_that_never_yields_starves_the_others "
          f"{'yes' if hogged.startswith('greedy greedy greedy greedy') else 'no'}   "
          f"(got '{hogged}')")
    print()
    print("Forty lines. A ready queue, a sleeping list and a while loop. Everything")
    print("asyncio adds on top of this — socket readiness from the operating system,")
    print("cancellation, timeouts, TaskGroups, thread offloading — is machinery around")
    print("that same idea, not a different idea.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Cancellation, timeouts, and what gather does that TaskGroup does not.

Run:  python3 examples/06_timeouts.py

Starting concurrent work is the easy half. Stopping it is where the bugs
are. Four things are measured here, all against the fixture server so the
"slow" request is slow by arrangement rather than by luck:

  1. asyncio.timeout around a request that will not finish in time
  2. what cancellation actually is — a CancelledError raised INSIDE the
     task at its next await, which means cleanup in a finally block runs
  3. asyncio.gather with return_exceptions, where one failure does not
     stop the others and you get the exception back as a value
  4. asyncio.TaskGroup, where one failure DOES cancel its siblings

The difference in 3 and 4 is a design decision, not a style preference.
"Fetch nine things and tell me which ones worked" wants gather. "Do these
four things or do none of them" wants a TaskGroup.
"""

from __future__ import annotations

import asyncio
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import labkit  # noqa: E402

SLOW = 0.40
BUDGET = 0.15


async def slow_fetch(base: str, tag: str, log: list[str]) -> str:
    """A request that takes longer than the caller's patience.

    The finally block is the part worth studying. Cancellation in asyncio is
    an exception raised inside the coroutine at its next suspension point,
    so ordinary Python cleanup works: finally runs, context managers exit,
    connections close. It is not a kill.
    """
    log.append(f"{tag} started")
    try:
        return await labkit.fetch_async(f"{base}/slow/{tag}")
    except asyncio.CancelledError:
        log.append(f"{tag} received CancelledError")
        raise  # re-raise: swallowing it is how you get a task that will not die
    finally:
        log.append(f"{tag} cleaned up")


async def failing_task(tag: str, log: list[str]) -> str:
    await asyncio.sleep(0.02)
    log.append(f"{tag} about to fail")
    raise ValueError(f"{tag} could not be fetched")


async def steady_task(tag: str, seconds: float, log: list[str]) -> str:
    try:
        await asyncio.sleep(seconds)
        log.append(f"{tag} finished")
        return f"{tag} ok"
    except asyncio.CancelledError:
        log.append(f"{tag} was cancelled")
        raise


async def demo_timeout(base: str) -> tuple[bool, list[str], float]:
    log: list[str] = []
    timed_out = False
    start = asyncio.get_running_loop().time()
    try:
        async with asyncio.timeout(BUDGET):
            await slow_fetch(base, "report", log)
    except TimeoutError:
        timed_out = True
    return timed_out, log, asyncio.get_running_loop().time() - start


async def demo_gather() -> tuple[list[object], list[str]]:
    log: list[str] = []
    results = await asyncio.gather(
        steady_task("a", 0.05, log),
        failing_task("b", log),
        steady_task("c", 0.10, log),
        return_exceptions=True,
    )
    return list(results), log


async def demo_taskgroup() -> tuple[list[str], list[str]]:
    log: list[str] = []
    messages: list[str] = []
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(steady_task("a", 0.05, log))
            group.create_task(failing_task("b", log))
            group.create_task(steady_task("c", 0.50, log))
    except* ValueError as group_error:
        messages = [str(error) for error in group_error.exceptions]
    return messages, log


async def main_async() -> int:
    with labkit.fixture_server(delay=SLOW) as base:
        print("Day 096 — cancellation and timeouts")
        print(f"the fixture server now sleeps {SLOW:.2f}s; the caller's budget is "
              f"{BUDGET:.2f}s")
        print()

        print("1. asyncio.timeout around work that will not finish in time")
        timed_out, timeout_log, elapsed = await demo_timeout(base)
        print(f"   TimeoutError raised: {'yes' if timed_out else 'no'}")
        print(f"   gave up after {elapsed:.3f}s, not after {SLOW:.2f}s")
        for line in timeout_log:
            print(f"   log: {line}")
        print("   Cancellation is an exception delivered inside the task, so the")
        print("   finally block ran and the socket was closed. Nothing leaked.")
        print()

        print("2. asyncio.gather(return_exceptions=True): one failure, others survive")
        gathered, gather_log = await demo_gather()
        for index, item in enumerate(gathered):
            kind = type(item).__name__ if isinstance(item, BaseException) else "result"
            print(f"   [{index}] {kind:<10} {item}")
        for line in gather_log:
            print(f"   log: {line}")
        survivors = sum(1 for item in gathered if not isinstance(item, BaseException))
        print(f"   {survivors} of {len(gathered)} finished; the failure came back as a value")
        print()

        print("3. asyncio.TaskGroup: one failure cancels its siblings")
        messages, group_log = await demo_taskgroup()
        for message in messages:
            print(f"   caught in the ExceptionGroup: {message}")
        for line in group_log:
            print(f"   log: {line}")
        cancelled = [line for line in group_log if "cancelled" in line]
        print(f"   {len(cancelled)} sibling(s) were cancelled rather than left running")
        print("   Task c asked for half a second and never got it. With gather it")
        print("   would have run to completion, doing work nobody was going to use.")
        print()

        labkit.result_line("timeout_elapsed_s", elapsed)
        labkit.result_line("timeout_budget_s", BUDGET)
        labkit.result_line("gather_survivors", float(survivors))
        labkit.result_line("taskgroup_cancelled_siblings", float(len(cancelled)))
        labkit.shape_line(
            "timeout_fires_at_the_budget_not_at_the_work",
            timed_out and elapsed < SLOW,
            f"gave up after {elapsed:.3f}s with a budget of {BUDGET:.2f}s and "
            f"work of {SLOW:.2f}s",
        )
        labkit.shape_line(
            "cancellation_runs_the_finally_block",
            any("cleaned up" in line for line in timeout_log),
            "the cancelled task's finally block appears in the log above",
        )
        labkit.shape_line(
            "gather_lets_siblings_finish",
            survivors == 2,
            f"{survivors} of 3 completed despite one raising",
        )
        labkit.shape_line(
            "taskgroup_cancels_siblings",
            len(cancelled) >= 1,
            f"{len(cancelled)} sibling(s) cancelled when one task raised",
        )
        print()
        print("Choosing between them: gather when partial success is a real answer,")
        print("TaskGroup when it is not. TaskGroup also refuses to let you leak a task,")
        print("because the `async with` block does not exit until every child is done.")
    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main_async()))

"""Reference answers to the eight exercises in starter/01_exercises.py.

Read this AFTER you have tried. Every function has the same name and
signature as the starter's, so the checker can be pointed at either:

    bash starter/02_check.sh examples/07_solutions.py

Each answer is the shortest one that is actually correct, with a note on
the mistake that version exists to avoid.
"""

from __future__ import annotations

import asyncio
import sys
import threading
import time
from collections import deque
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import labkit  # noqa: E402


def fetch_all_sequentially(urls: list[str]) -> list[str]:
    """The baseline. Correct, obvious, and pays every wait end to end."""
    return [labkit.fetch(url) for url in urls]


def fetch_all_with_threads(urls: list[str], workers: int) -> list[str]:
    """Threads help because a blocked socket read releases the interpreter lock.

    `pool.map` returns results in INPUT order. `as_completed` would return
    them in finish order, which is what you want for a progress bar and not
    what you want when the caller expects a list lined up with its input.
    """
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(labkit.fetch, urls))


def fetch_all_with_asyncio(urls: list[str]) -> list[str]:
    """One thread, one loop, N coroutines suspended at await.

    `asyncio.gather` preserves argument order in its result list. Note this
    is an ordinary def: `asyncio.run` is the boundary between synchronous
    code and the loop, and calling it from inside a running loop is an error.
    """

    async def gather_all() -> list[str]:
        return list(await asyncio.gather(*(labkit.fetch_async(url) for url in urls)))

    return asyncio.run(gather_all())


def count_primes_with_threads(limits: list[int], workers: int) -> list[int]:
    """Identical to the fast version above, and it will not be faster.

    Nothing here waits, so nothing releases the interpreter lock for long
    enough to matter. The answers are right; the wall clock is unmoved.
    """
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(labkit.count_primes, limits))


def count_primes_with_processes(limits: list[int], workers: int) -> list[int]:
    """One word changed, and now there are four interpreter locks.

    `labkit.count_primes` is a module-level function, which is why the child
    processes can find it. A lambda, a closure, or a locally defined function
    raises a pickling error here — the most common first failure with
    ProcessPoolExecutor, and one worth meeting deliberately.
    """
    with ProcessPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(labkit.count_primes, limits))


def wait_without_blocking_the_loop(naps: list[float]) -> list[float]:
    """The repair for a synchronous library you cannot rewrite.

    `asyncio.to_thread(fn, *args)` returns a coroutine that runs fn in a
    worker thread. The blocking still happens; it just happens somewhere
    that is allowed to block, so the loop keeps its thread.
    """

    async def nap(seconds: float) -> float:
        await asyncio.to_thread(time.sleep, seconds)
        return seconds

    async def all_naps() -> list[float]:
        return list(await asyncio.gather(*(nap(seconds) for seconds in naps)))

    return asyncio.run(all_naps())


def counter_that_loses_nothing(threads: int, per_thread: int) -> int:
    """The better answer: do not share the mutable state at all.

    Each worker counts into a local variable that no other thread can see,
    and posts one subtotal at the end. There is exactly one shared mutation
    per thread instead of `per_thread` of them, and it is under a lock.

    The direct answer — one shared counter with `with lock:` around every
    read-add-write — is also correct and also passes. It is roughly an order
    of magnitude slower here, because it takes and releases a lock several
    hundred thousand times to protect an addition.
    """
    lock = threading.Lock()
    total = 0

    def worker() -> None:
        nonlocal total
        subtotal = 0
        for _ in range(per_thread):
            subtotal += 1
        with lock:
            total += subtotal

    workers = [threading.Thread(target=worker) for _ in range(threads)]
    for thread in workers:
        thread.start()
    for thread in workers:
        thread.join()
    return total


def round_robin(tasks: list[tuple[str, object]]) -> list[str]:
    """An event loop, minus everything that is not the loop.

    A ready queue, take the front one, advance it exactly one step, put it
    at the back unless it has finished. That is scheduling. Everything
    asyncio adds is about deciding WHEN a task becomes ready again.
    """
    ready = deque(tasks)
    order: list[str] = []
    while ready:
        name, task = ready.popleft()
        try:
            next(task)  # type: ignore[arg-type]
        except StopIteration:
            continue  # finished: do not put it back
        order.append(name)
        ready.append((name, task))
    return order

"""Shared kit for the Day 096 lab: something that waits, something that
computes, and an honest way to time both.

Standard library only. Nothing here reaches the internet. The "waiting"
work is a small HTTP server bound to the loopback address 127.0.0.1 on an
ephemeral port the operating system chooses, which sleeps a fixed time
before answering. That makes waiting *real* — a socket, a kernel, a
genuine blocked thread — and *reproducible*, because the delay is a
number you set rather than whatever the internet felt like today.

The CPU-bound work is a prime count by trial division. It was chosen for
three properties: it is pure Python, so it holds the interpreter lock; it
is deterministic, so a wrong answer is detectable; and it takes long
enough that process start-up cost does not swamp the measurement.

Every function here is used by at least two of the example scripts, the
starter checker and the test suite.
"""

from __future__ import annotations

import asyncio
import statistics
import threading
import time
import urllib.parse
import urllib.request
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# One tenth of a second per request. Long enough that the waiting dominates
# the measurement, short enough that twenty sequential requests still finish
# in about two seconds.
DEFAULT_DELAY = 0.10

# Trial division below this limit. Calibrated on the authoring machine so a
# single call takes a few hundred milliseconds — long enough that starting a
# process pool is not the thing being measured, short enough that the whole
# test suite still finishes in under a minute. See requirements/README.md.
DEFAULT_PRIME_LIMIT = 500_000

# How many of each kind of unit of work the example scripts use.
WAITING_REQUESTS = 20
COMPUTING_TASKS = 4


# ---------------------------------------------------------------------------
# The waiting half: a fixture server that sleeps, then answers.
# ---------------------------------------------------------------------------


class _WaitHandler(BaseHTTPRequestHandler):
    """Answers any GET after sleeping. The sleep is the whole point."""

    # HTTP/1.0 means the server closes the connection when it has finished
    # writing, so a client can simply read to end-of-stream. That keeps the
    # raw-socket asyncio client in 01_waiting.py down to a dozen lines.
    protocol_version = "HTTP/1.0"

    def do_GET(self) -> None:  # noqa: N802 - name fixed by the base class
        delay = self.server.delay  # type: ignore[attr-defined]
        time.sleep(delay)
        body = f"waited {delay:.3f}s for {self.path}\n".encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args: object) -> None:
        """Silence the per-request log line; the measurements are the output."""


class _WaitServer(ThreadingHTTPServer):
    """A threading server, so it can actually answer requests in parallel.

    This matters more than it looks. If the fixture server handled one
    request at a time, every client would measure the SERVER's serialisation
    rather than its own concurrency, and the whole lab would prove nothing.
    """

    daemon_threads = True
    delay = DEFAULT_DELAY


@contextmanager
def fixture_server(delay: float = DEFAULT_DELAY):
    """Run the fixture server for the duration of the block.

    Yields the base URL. Binding to port 0 asks the operating system for a
    free ephemeral port, so two copies of this lab can run at once and
    neither needs a privileged port or a firewall rule.
    """
    server = _WaitServer(("127.0.0.1", 0), _WaitHandler)
    server.delay = delay
    thread = threading.Thread(target=server.serve_forever, name="fixture-server", daemon=True)
    thread.start()
    host, port = server.server_address[0], server.server_address[1]
    try:
        yield f"http://{host}:{port}"
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5.0)


def fetch(url: str, timeout: float = 30.0) -> str:
    """Fetch one URL and block until it answers. This is a BLOCKING call."""
    with urllib.request.urlopen(url, timeout=timeout) as response:  # noqa: S310
        return response.read().decode("utf-8")


async def fetch_async(url: str) -> str:
    """Fetch one URL without blocking the event loop.

    Written against raw sockets on purpose. There is no HTTP client in the
    standard library that speaks asyncio, and writing the twelve lines makes
    the point that `await` is where this coroutine gives the loop its turn:
    at the connect, at the drain, and at the read.
    """
    parts = urllib.parse.urlsplit(url)
    host = parts.hostname or "127.0.0.1"
    port = parts.port or 80
    path = parts.path or "/"
    reader, writer = await asyncio.open_connection(host, port)
    request = f"GET {path} HTTP/1.0\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n"
    writer.write(request.encode("ascii"))
    await writer.drain()
    raw = await reader.read()
    writer.close()
    await writer.wait_closed()
    _headers, _sep, body = raw.partition(b"\r\n\r\n")
    return body.decode("utf-8")


def paths(count: int) -> list[str]:
    """The request paths used everywhere in this lab, so counts line up."""
    return [f"/item/{n}" for n in range(1, count + 1)]


def urls(base: str, count: int) -> list[str]:
    return [base + p for p in paths(count)]


# ---------------------------------------------------------------------------
# The computing half: a function that burns CPU and holds the lock.
# ---------------------------------------------------------------------------


def count_primes(limit: int = DEFAULT_PRIME_LIMIT) -> int:
    """Count the primes below `limit` by trial division.

    Deliberately not a sieve. A sieve would be fast and mostly memory-bound;
    this is a tight arithmetic loop in pure Python, which is exactly the
    shape of work that cannot overlap while one interpreter lock exists.
    """
    if limit <= 2:
        return 0
    count = 1  # 2 is prime and is the only even one
    for number in range(3, limit, 2):
        factor = 3
        while factor * factor <= number:
            if number % factor == 0:
                break
            factor += 2
        else:
            count += 1
    return count


# ---------------------------------------------------------------------------
# Measuring, honestly.
# ---------------------------------------------------------------------------


def timed(function, *args, **kwargs) -> tuple[float, object]:
    """Return (elapsed seconds, result). perf_counter is monotonic."""
    start = time.perf_counter()
    result = function(*args, **kwargs)
    return time.perf_counter() - start, result


def repeat(function, times: int = 3) -> tuple[list[float], object]:
    """Run the same thing `times` over and keep every sample.

    One measurement is an anecdote. Three is still a small sample, but it
    shows you the spread, and the spread is the part most benchmarks hide.
    """
    samples: list[float] = []
    result: object = None
    for _ in range(times):
        elapsed, result = timed(function)
        samples.append(elapsed)
    return samples, result


def report(label: str, samples: list[float]) -> float:
    """Print every sample and the median, and return the median.

    The median is the headline because it is the sample least disturbed by
    one unlucky run, and because a mean over three samples with one outlier
    is a lie with a decimal point on it.
    """
    median = statistics.median(samples)
    spread = max(samples) - min(samples)
    runs = ", ".join(f"{s:6.3f}" for s in samples)
    print(f"  {label:<34} runs: {runs}   median {median:6.3f}s   spread {spread:5.3f}s")
    return median


def result_line(name: str, value: float) -> None:
    """One machine-readable line the test suite can parse."""
    print(f"RESULT {name} {value:.4f}")


def shape_line(name: str, holds: bool, detail: str) -> None:
    """The claim that has to survive on another machine: a SHAPE, not a time."""
    print(f"SHAPE {name} {'yes' if holds else 'no'}   ({detail})")

lesson_id: D096
day: 96
kind: guided-build
languages: [python, bash]
setup_commands:
  - cd labs/sections/programming-with-python/day-096-concurrency-and-async-basics
  - python3 --version
  - 'python3 -c "import sys, sysconfig; print(sys.version); print(sysconfig.get_config_var(''Py_GIL_DISABLED''))"'
  - 'python3 -c "import os; print(os.cpu_count(), ''logical CPUs'')"'
run_commands:
  - bash tests/run_tests.sh
  - bash starter/02_check.sh
  - bash starter/02_check.sh examples/07_solutions.py
  - python3 examples/01_waiting.py
  - python3 examples/02_computing.py
  - python3 examples/03_blocking_coroutine.py
  - python3 examples/04_race.py
  - python3 examples/05_scheduler.py
  - python3 examples/06_timeouts.py
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - find . -type d -name __pycache__ -prune -exec rm -rf -- {} +
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: false
requires_api_key: false
estimated_minutes: 35
last_executed: '2026-08-16'
executed_on: 'macOS 26.5.2 (Apple Silicon, arm64, 14 logical CPUs), Python 3.14.0 with Py_GIL_DISABLED=0, default switch interval 0.005 s, bash 3.2.57 — bash tests/run_tests.sh -> 58 checks, 0 failure(s), exit 0'

# Requirements — Day 096

## What you need

| Tool | Minimum | Used on the authoring machine | Why the minimum is what it is |
| --- | --- | --- | --- |
| `python3` | 3.11 | 3.14.0 | `asyncio.TaskGroup`, `asyncio.timeout` and `except*` all arrived in 3.11. Earlier versions raise a `SyntaxError` on `except*` |
| `bash` | 3.2 | 3.2.57 | `tests/run_tests.sh` and `starter/02_check.sh`. Written for bash 3.2 so the macOS system bash works unmodified |

Two or more logical CPUs. The machine used here had 14.

## What you do not need

`requirements.txt` in this directory lists no packages, and that is
deliberate rather than an oversight. Every concurrency model this lab
compares ships with Python:

| Model | Module | In the standard library since |
| --- | --- | --- |
| Threads | `threading` | Long enough that the documentation no longer records a version; it predates every other row here |
| Processes | `multiprocessing` | Python 2.6, 2008 |
| Uniform pool interface | `concurrent.futures` | Python 3.2, 2011 |
| Event loop and coroutines | `asyncio` | Python 3.4, 2014 |
| `async` / `await` syntax | language | Python 3.5, 2015 |
| Structured concurrency | `asyncio.TaskGroup` | Python 3.11, 2022 |

Also used, all standard library: `queue`, `time`, `urllib.request`,
`urllib.parse`, `http.server`, `socket` (indirectly, through `asyncio` and
`http.server`), `collections`, `statistics`, `sysconfig`, `importlib.util`,
`contextlib`, `pathlib`, `ast`.

`tests/run_tests.sh` parses every import in every `.py` file in this lab
against `sys.stdlib_module_names` and fails if anything outside it appears.
So this claim is checked on each run rather than asserted here.

## The libraries this lab talks about but does not install

The lesson's Alternatives section covers `trio`, `anyio`, `gevent` and
Celery. None of them is installed, none is imported, and **no output from any
of them is reproduced anywhere in this lab or lesson** — they are described
from their documentation and labelled as such. If you want to try them, they
belong in a virtual environment of your own, outside this directory.

## How the workload sizes were chosen

Both are calibrated, and the reasoning matters because a badly sized workload
makes a measurement lie.

**Waiting work** — 20 requests at 0.100 s each. The delay is a `time.sleep`
inside the fixture server, so it is exact and identical on every machine. It
is long enough that the waiting dominates the per-request overhead of urllib
and of opening a socket, and short enough that the sequential baseline
finishes in about two seconds. The result is a sequential run pinned near
2.0 s on any hardware, which is what makes the ratios comparable across
machines.

**Computing work** — 4 tasks, each counting the primes below 500,000 by
trial division, measured on the authoring machine at 0.364 s per call. The
full calibration, captured from the command below:

| Limit | Primes below it | Time on the authoring machine |
| --- | --- | --- |
| 120,000 | 11,301 | 0.046 s |
| 400,000 | 33,860 | 0.260 s |
| 500,000 | 41,538 | 0.364 s |
| 700,000 | 56,543 | 0.594 s |
| 1,000,000 | 78,498 | 1.014 s |

500,000 was chosen as the smallest size that is comfortably larger than the
cost of starting a process pool — on macOS, `ProcessPoolExecutor` uses the
spawn start method, so each of the four children re-imports the module before
doing any work. Too small a task and that start-up cost is the entire
measurement; too large and the test suite becomes tedious to run.

Trial division was chosen over a sieve on purpose: it is a tight arithmetic
loop in pure Python that holds the interpreter lock, which is precisely the
shape of work the comparison is about. A sieve would spend much of its time
in list operations and would muddy the result.

The starter checker uses 400,000 rather than 500,000, and 12 requests rather
than 20, so that a learner re-running it after every edit is not waiting
around.

## Reproducing the calibration

```bash
python3 - <<'PY'
import sys, time
sys.path.insert(0, "examples")
import labkit
for limit in (120_000, 400_000, 500_000, 700_000, 1_000_000):
    start = time.perf_counter()
    count = labkit.count_primes(limit)
    print(f"{limit:>9,}  {count:>7,} primes  {time.perf_counter() - start:.3f}s")
PY
```

Run it from the lab directory. Your times will differ; the prime counts will
not, and are checkable against any table of the prime-counting function.

## What is deliberately absent

- **No `pytest`.** The harness is a bash assert script, as everywhere else in
  this course, so the lab has no dependency at all.
- **No benchmarking library.** `time.perf_counter` and three repetitions with
  the spread printed is the right amount of machinery for teaching the shape
  of a result. A serious benchmark of a serious system needs more, and the
  lesson says so rather than pretending three runs is rigorous.
- **No async HTTP client.** There is none in the standard library. The lab
  writes twelve lines of raw `asyncio.open_connection` instead, which is more
  instructive than importing one would have been: you can see exactly which
  three lines are the `await` points.

# Day 096 — Concurrency and async Basics
#
# This file is intentionally empty of packages.
#
# Every concurrency model this lab compares ships with Python itself:
# threading, multiprocessing, concurrent.futures, asyncio and queue are all
# standard library. There is nothing to install and nothing to keep up to
# date, which is why this lab runs offline on a fresh machine.
#
# tests/run_tests.sh parses every import in every .py file here against
# sys.stdlib_module_names and fails if anything outside it appears, so this
# is checked on every run rather than merely claimed.
#
# The alternatives the lesson discusses — trio, anyio, gevent and Celery —
# are described from their documentation and are NOT installed. No output
# from any of them is reproduced anywhere in this lab. If you want to try
# them, use a virtual environment outside this directory.

"""The progress checker behind `bash starter/02_check.sh`.

You do not need to edit this file, but reading it is worthwhile: it is the
same measure-do-not-believe discipline the lesson argues for, applied to
your own code. Nothing here inspects how you wrote a function. Every check
either compares a real answer against the right one, or times two versions
of the same work and compares the ratio.

Speed checks are stated as ratios with generous margins — "at least three
times faster", never "under 200 milliseconds" — so that a slow laptop, a
busy machine or a different operating system does not fail work that is
correct. The ratio is the claim that travels.

Usage:  python3 starter/_progress.py [path-to-module]
"""

from __future__ import annotations

import importlib.util
import sys
import time
from pathlib import Path

LAB = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(LAB / "examples"))

import labkit  # noqa: E402

TOTAL = 8
REQUESTS = 12
REQUEST_DELAY = 0.05
PRIME_LIMIT = 400_000
PRIME_TASKS = 4
PRIME_ANSWER = 33860  # the number of primes below 400,000
NAPS = [0.15] * 5
RACE_THREADS = 8
RACE_PER_THREAD = 50_000
TIGHT_INTERVAL = 1e-6

results: list[tuple[int, str, bool, str]] = []


def record(number: int, title: str, passed: bool, detail: str) -> None:
    results.append((number, title, passed, detail))


def load(path: Path):
    spec = importlib.util.spec_from_file_location("under_test", path)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot import {path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def call(module, name: str, *args):
    """Call an exercise. Returns (elapsed, value) or (None, reason-string)."""
    function = getattr(module, name, None)
    if function is None:
        return None, f"no function named {name}"
    start = time.perf_counter()
    try:
        value = function(*args)
    except NotImplementedError:
        return None, "not started"
    except Exception as error:  # noqa: BLE001 - report anything, do not crash
        return None, f"{type(error).__name__}: {error}"
    return time.perf_counter() - start, value


def bodies_ok(value: object, count: int) -> bool:
    return (
        isinstance(value, list)
        and len(value) == count
        and all(isinstance(body, str) and "waited" in body for body in value)
        and all(f"/item/{n}" in value[n - 1] for n in range(1, count + 1))
    )


def check_waiting(module) -> float | None:
    """Exercises 1-3. Returns the sequential baseline, or None if it failed."""
    with labkit.fixture_server(delay=REQUEST_DELAY) as base:
        labkit.fetch(base + "/warmup")
        targets = labkit.urls(base, REQUESTS)

        elapsed, value = call(module, "fetch_all_sequentially", targets)
        if elapsed is None:
            record(1, "fetch_all_sequentially", False, str(value))
            baseline = None
        elif not bodies_ok(value, REQUESTS):
            record(1, "fetch_all_sequentially", False, "wrong bodies or wrong order")
            baseline = None
        else:
            baseline = elapsed
            record(1, "fetch_all_sequentially", True, f"{REQUESTS} bodies in order, {elapsed:.3f}s")

        for number, name, args in (
            (2, "fetch_all_with_threads", (targets, REQUESTS)),
            (3, "fetch_all_with_asyncio", (targets,)),
        ):
            elapsed, value = call(module, name, *args)
            if elapsed is None:
                record(number, name, False, str(value))
                continue
            if not bodies_ok(value, REQUESTS):
                record(number, name, False, "wrong bodies or wrong order")
                continue
            if baseline is None:
                record(number, name, False, "cannot judge speed: exercise 1 is not working yet")
                continue
            ratio = baseline / elapsed
            record(
                number,
                name,
                ratio >= 2.5,
                f"{ratio:.1f}x faster than sequential ({elapsed:.3f}s); needs >= 2.5x",
            )
    return baseline


def check_computing(module) -> None:
    """Exercises 4 and 5."""
    limits = [PRIME_LIMIT] * PRIME_TASKS
    expected = [PRIME_ANSWER] * PRIME_TASKS
    start = time.perf_counter()
    sequential_answer = [labkit.count_primes(limit) for limit in limits]
    baseline = time.perf_counter() - start
    assert sequential_answer == expected, "the checker's own baseline is wrong"

    elapsed, value = call(module, "count_primes_with_threads", limits, PRIME_TASKS)
    if elapsed is None:
        record(4, "count_primes_with_threads", False, str(value))
    elif value != expected:
        record(4, "count_primes_with_threads", False, f"wrong counts: {value}")
    else:
        record(
            4,
            "count_primes_with_threads",
            True,
            f"counts correct; {baseline / elapsed:.2f}x sequential — threads do not help here, "
            "and are not required to",
        )

    elapsed, value = call(module, "count_primes_with_processes", limits, PRIME_TASKS)
    if elapsed is None:
        record(5, "count_primes_with_processes", False, str(value))
    elif value != expected:
        record(5, "count_primes_with_processes", False, f"wrong counts: {value}")
    else:
        ratio = baseline / elapsed
        record(
            5,
            "count_primes_with_processes",
            ratio >= 1.4,
            f"counts correct; {ratio:.2f}x sequential ({elapsed:.3f}s); needs >= 1.4x",
        )


def check_offloading(module) -> None:
    """Exercise 6."""
    elapsed, value = call(module, "wait_without_blocking_the_loop", NAPS)
    if elapsed is None:
        record(6, "wait_without_blocking_the_loop", False, str(value))
        return
    if value != NAPS:
        record(6, "wait_without_blocking_the_loop", False, f"expected {NAPS}, got {value}")
        return
    serial = sum(NAPS)
    record(
        6,
        "wait_without_blocking_the_loop",
        elapsed < serial * 0.6,
        f"{elapsed:.3f}s against a serial floor of {serial:.2f}s; needs < {serial * 0.6:.2f}s",
    )


def check_counter(module) -> None:
    """Exercise 7, run at a switch interval that breaks unprotected code."""
    function = getattr(module, "counter_that_loses_nothing", None)
    if function is None:
        record(7, "counter_that_loses_nothing", False, "no function of that name")
        return
    expected = RACE_THREADS * RACE_PER_THREAD
    previous = sys.getswitchinterval()
    seen: list[int] = []
    try:
        sys.setswitchinterval(TIGHT_INTERVAL)
        for _ in range(3):
            try:
                seen.append(function(RACE_THREADS, RACE_PER_THREAD))
            except NotImplementedError:
                record(7, "counter_that_loses_nothing", False, "not started")
                return
            except Exception as error:  # noqa: BLE001
                record(7, "counter_that_loses_nothing", False, f"{type(error).__name__}: {error}")
                return
    finally:
        sys.setswitchinterval(previous)
    exact = all(value == expected for value in seen)
    worst = min(seen)
    record(
        7,
        "counter_that_loses_nothing",
        exact,
        f"3 runs at a {TIGHT_INTERVAL}s switch interval; "
        + (f"all exactly {expected:,}" if exact else f"lost up to {expected - worst:,} increments"),
    )


def make_task(steps: int):
    def task():
        for _ in range(steps):
            yield

    return task()


def check_scheduler(module) -> None:
    """Exercise 8."""
    tasks = [("a", make_task(3)), ("b", make_task(2)), ("c", make_task(1))]
    elapsed, value = call(module, "round_robin", tasks)
    if elapsed is None:
        record(8, "round_robin", False, str(value))
        return
    expected = ["a", "b", "c", "a", "b", "a"]
    record(
        8,
        "round_robin",
        value == expected,
        f"expected {expected}, got {value}",
    )


def main() -> int:
    target = Path(sys.argv[1]) if len(sys.argv) > 1 else LAB / "starter" / "01_exercises.py"
    if not target.is_absolute():
        target = (Path.cwd() / target).resolve()
    if not target.exists():
        print(f"No such file: {target}")
        return 2

    print(f"Checking {target.name}")
    print(f"python {sys.version.split()[0]}   "
          f"default switch interval {sys.getswitchinterval()} s")
    print()

    try:
        module = load(target)
    except Exception as error:  # noqa: BLE001
        print(f"The module would not import: {type(error).__name__}: {error}")
        print()
        print(f"0 of {TOTAL} exercises complete.")
        return 1

    check_waiting(module)
    check_computing(module)
    check_offloading(module)
    check_counter(module)
    check_scheduler(module)

    passed = 0
    for number, title, ok, detail in sorted(results):
        mark = "ok  " if ok else "open"
        passed += 1 if ok else 0
        print(f"  [{mark}] {number}. {title}")
        print(f"         {detail}")
    print()
    print(f"{passed} of {TOTAL} exercises complete.")
    return 0 if passed == TOTAL else 1


if __name__ == "__main__":
    raise SystemExit(main())

# The brief — Waiting Versus Computing

Read this before you write anything. It takes three minutes and it is the
whole point of the day.

## The situation

You have inherited a small service. It does two things.

It **fetches** twenty records from an upstream system, one HTTP request each,
and each request takes about a tenth of a second because the upstream is slow.
And it **computes** a checksum over each batch, which is a few hundred
milliseconds of arithmetic per batch with no I/O at all.

Both are slow. Somebody suggests "make it concurrent". That instruction is
not actionable, because the right answer for the first half is the wrong
answer for the second half, and the wrong answer looks like it worked
because the code compiles, the tests pass and the timings do not move.

## The one question

Before you choose a tool, answer this about the work in front of you:

> **Is this work waiting, or is it computing?**

- **Waiting** — for a socket, a disk, a database, a lock, a subprocess, a
  human. The CPU is idle. Threads help. An event loop helps more, at the
  cost of every library in the call path having to cooperate.
- **Computing** — arithmetic, parsing, encoding, compression, model
  inference on the CPU. The CPU is busy. Threads do not help, because
  CPython lets only one thread execute Python bytecode at a time. Processes
  help, because each process has its own interpreter.

That is the decision. Everything else on this day is you proving it to
yourself rather than taking it on trust.

## What you are going to build

Eight functions in `starter/01_exercises.py`. Run
`bash starter/02_check.sh` at any point to see where you stand; it starts by
saying `0 of 8 exercises complete.` and names each one that is still open.

| # | Function | The point |
| --- | --- | --- |
| 1 | `fetch_all_sequentially` | The baseline. Everything is measured against it |
| 2 | `fetch_all_with_threads` | Waiting work + threads. Must be at least 2.5x faster |
| 3 | `fetch_all_with_asyncio` | Waiting work + event loop. Must be at least 2.5x faster |
| 4 | `count_primes_with_threads` | Computing work + threads. Correct, and **not** faster |
| 5 | `count_primes_with_processes` | Computing work + processes. Must be at least 1.4x faster |
| 6 | `wait_without_blocking_the_loop` | A blocking call rescued with `asyncio.to_thread` |
| 7 | `counter_that_loses_nothing` | A race the checker forces to happen, and your fix |
| 8 | `round_robin` | An event loop of your own, in about a dozen lines |

Exercise 4 is the one people skip because it looks like a repeat of exercise
2. It is the opposite of exercise 2, and the checker will print the ratio so
you can see it.

## What the requests actually talk to

Nothing outside your machine. `examples/labkit.py` starts a small HTTP
server bound to `127.0.0.1` on a port the operating system picks, and that
server sleeps for a fixed number of milliseconds before answering. So the
waiting is genuine — a real socket, a really blocked thread — and it is
also exactly reproducible, which the internet never is. Nothing in this lab
needs a network connection, a key or an account.

## How you will be judged

By behaviour and by ratios, never by a stopwatch reading. The checker asks
"is the threaded version at least two and a half times faster than the
sequential one on this machine, right now?" — never "did it finish in under
200 milliseconds". A millisecond figure is a fact about the machine that
produced it. A ratio is a fact about the program.

Hold yourself to the same rule when you report a performance result to
anybody: say what you measured, say how many times you ran it, say what the
spread was, and say which machine on which day.

## The order to work in

1. Exercise 1, then 2, then 3, and **look at the ratios** the checker prints.
2. Exercise 4, then 5, and look at those ratios too. This is the lesson.
3. Exercise 6 — the mistake that makes an async service serve one request at
   a time while looking busy.
4. Exercise 7 — shared state. Read `examples/04_race.py` afterwards, because
   it contains a result that contradicts what most books say about this, and
   the reason it does is worth knowing.
5. Exercise 8 — write the loop. After this, `await` is not magic.

Then run the six example scripts in order and read them. They are the
narrated version of everything you just did, with the measurements printed.

"""Day 096 starter — eight exercises. This is where YOUR work happens.

Read `starter/00_brief.md` first. Then run:

    bash starter/02_check.sh

It will say `0 of 8 exercises complete.` and tell you which. Work down the
list, re-running the checker as you go. The checker never looks at how you
wrote a function — only at whether it behaves correctly and, where the
point of the exercise is speed, whether it is actually faster.

Everything you need is in the standard library and in `examples/labkit.py`,
which is imported for you below. The pieces of labkit you will want:

    labkit.fetch(url)              blocking fetch, returns the body as str
    labkit.fetch_async(url)        coroutine fetch, must be awaited
    labkit.count_primes(limit)     CPU-bound, returns an int
    labkit.urls(base, count)       build the list of request URLs

Do not change any function's name or signature: the checker calls them.
"""

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "examples"))

import labkit  # noqa: E402,F401  (imported for you; every exercise below uses it)


# ---------------------------------------------------------------------------
# Exercise 1 — the baseline. Fetch every URL, one after another.
#
# Use labkit.fetch on each URL in turn and return the bodies in the same
# order as the URLs. A list comprehension is enough. This is deliberately
# the slow one: you need it to measure the others against.
# ---------------------------------------------------------------------------
def fetch_all_sequentially(urls: list[str]) -> list[str]:
    raise NotImplementedError("Exercise 1: return [labkit.fetch(u) for u in urls]")


# ---------------------------------------------------------------------------
# Exercise 2 — the same work with threads, and it must actually be faster.
#
# Use concurrent.futures.ThreadPoolExecutor with max_workers=workers, and
# its .map method, which returns results in INPUT order rather than
# completion order. Wrap the executor in a `with` block so the pool shuts
# down cleanly.
#
# The checker requires this to be at least 2.5x faster than exercise 1 on
# the same URLs — a real speed-up, not just a different spelling.
# ---------------------------------------------------------------------------
def fetch_all_with_threads(urls: list[str], workers: int) -> list[str]:
    raise NotImplementedError("Exercise 2: use ThreadPoolExecutor(max_workers=workers).map")


# ---------------------------------------------------------------------------
# Exercise 3 — the same work on one thread with an event loop.
#
# This is an ordinary def, not an async def. Inside it, define a coroutine
# (or use asyncio.gather directly) that awaits labkit.fetch_async for every
# URL, and drive it with asyncio.run. Return the bodies in URL order —
# asyncio.gather preserves the order of its arguments.
#
# The checker requires this to be at least 2.5x faster than exercise 1.
# ---------------------------------------------------------------------------
def fetch_all_with_asyncio(urls: list[str]) -> list[str]:
    raise NotImplementedError("Exercise 3: asyncio.run over asyncio.gather(*fetch_async(u))")


# ---------------------------------------------------------------------------
# Exercise 4 — CPU-bound work with threads. It will NOT be faster.
#
# Same shape as exercise 2, but map labkit.count_primes over `limits`.
# Return the counts in input order.
#
# The checker only requires the ANSWERS to be right here. It then measures
# the speed-up and prints it, so you can see for yourself that threads did
# nothing for work that never waits.
# ---------------------------------------------------------------------------
def count_primes_with_threads(limits: list[int], workers: int) -> list[int]:
    raise NotImplementedError("Exercise 4: ThreadPoolExecutor over labkit.count_primes")


# ---------------------------------------------------------------------------
# Exercise 5 — the same CPU work with processes, and this one must be faster.
#
# Change ThreadPoolExecutor to ProcessPoolExecutor. That is the whole edit.
# It works because labkit.count_primes is a module-level function, so the
# child processes can import it; a lambda or a closure would fail to pickle.
#
# The checker requires this to be at least 1.4x faster than the sequential
# version of the same work.
# ---------------------------------------------------------------------------
def count_primes_with_processes(limits: list[int], workers: int) -> list[int]:
    raise NotImplementedError("Exercise 5: ProcessPoolExecutor over labkit.count_primes")


# ---------------------------------------------------------------------------
# Exercise 6 — repair a blocking call inside a coroutine.
#
# You are given a list of nap lengths in seconds and must wait all of them
# CONCURRENTLY, using only the blocking time.sleep — imagine it is a
# synchronous database driver you do not own.
#
# Writing `async def nap(n): time.sleep(n)` and gathering them does not
# work: it takes sum(naps), because time.sleep never gives the loop back.
# Use `await asyncio.to_thread(time.sleep, n)` instead, gather those, and
# return the nap lengths in input order.
#
# The checker requires the whole thing to finish in well under sum(naps).
# ---------------------------------------------------------------------------
def wait_without_blocking_the_loop(naps: list[float]) -> list[float]:
    raise NotImplementedError("Exercise 6: gather asyncio.to_thread(time.sleep, n) for each nap")


# ---------------------------------------------------------------------------
# Exercise 7 — a counter that loses nothing.
#
# Start `threads` threads. Each increments one shared counter `per_thread`
# times. Return the final value, which must be exactly threads * per_thread.
#
# The checker runs your function with the interpreter's thread-switch
# interval turned down to a microsecond, which makes an unprotected
# read-add-write lose increments on every single run. A threading.Lock held
# around the read and the write is the direct answer. Accumulating into a
# local variable per thread and adding the subtotals at the end is the
# better one, and also passes.
# ---------------------------------------------------------------------------
def counter_that_loses_nothing(threads: int, per_thread: int) -> int:
    raise NotImplementedError("Exercise 7: protect the read-add-write, or stop sharing it")


# ---------------------------------------------------------------------------
# Exercise 8 — write the event loop.
#
# `tasks` is a list of (name, generator) pairs. Each generator yields to
# pause and eventually stops. Run them round-robin: take the first task,
# advance it exactly one step with next(), and if it has not finished, put
# it at the BACK of the queue. Return the list of names in the order they
# were stepped.
#
# collections.deque with popleft() and append() is the natural structure.
# A generator that has finished raises StopIteration from next(); catch it
# and drop that task.
#
# With three tasks of 3, 2 and 1 steps the answer is:
#   ['a', 'b', 'c', 'a', 'b', 'a']
# ---------------------------------------------------------------------------
def round_robin(tasks: list[tuple[str, object]]) -> list[str]:
    raise NotImplementedError("Exercise 8: a deque, popleft, next(task), append unless finished")

#!/usr/bin/env bash
# How far through the eight exercises are you?
#
#   bash starter/02_check.sh                          # checks your work
#   bash starter/02_check.sh examples/07_solutions.py  # checks the reference
#
# Exits 0 only when all eight are complete, so it cannot be mistaken for
# finished. Every check is a real value or a real speed ratio; none of them
# looks at how you wrote anything.
set -u

starter_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
lab_dir="$(cd "${starter_dir}/.." && pwd)"

python_bin="${PYTHON:-$(command -v python3 || true)}"
if [ -z "${python_bin}" ] || [ ! -x "${python_bin}" ]; then
  echo "python3 was not found. Install Python 3.11 or newer, or set PYTHON=/path/to/python3."
  exit 2
fi

# Keep the lab directory clean: no __pycache__ left behind by the import.
export PYTHONDONTWRITEBYTECODE=1

if [ "$#" -ge 1 ]; then
  target="$1"
  case "${target}" in
    /*) ;;
    *) target="${lab_dir}/${target}" ;;
  esac
else
  target="${starter_dir}/01_exercises.py"
fi

"${python_bin}" "${starter_dir}/_progress.py" "${target}"

#!/usr/bin/env bash
# Tests for the Day 096 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# WHAT THIS SUITE ASSERTS, AND WHAT IT DELIBERATELY DOES NOT
#
# It never asserts a number of milliseconds. A millisecond figure is a fact
# about one machine on one day; it is flaky everywhere else and it teaches
# nothing. What this suite asserts is the SHAPE of the result, with margins
# wide enough to survive a slow laptop, a busy CI runner or a different
# operating system:
#
#   * waiting work gets much faster with threads, and with an event loop
#   * computing work does NOT get meaningfully faster with threads
#   * computing work DOES get faster with processes
#   * a blocking call inside a coroutine collapses the loop to serial, and
#     asyncio.to_thread repairs it
#   * an unprotected shared counter loses increments; a protected one does not
#   * two locks taken in opposite orders deadlock; taken in one order they do not
#   * the hand-written generator scheduler really interleaves its tasks
#
# The example scripts print machine-readable `RESULT name value` lines. This
# suite parses those and applies its own thresholds, rather than trusting the
# scripts' own `SHAPE` verdicts — a test that asks the code under test whether
# it passed is not a test.
#
# Nothing here touches the network: the only sockets are on 127.0.0.1. Nothing
# needs sudo. Everything is built in a temporary directory removed in a trap.
set -u

lab_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
work=""
checks=0
failures=0

cleanup() {
  [ -n "${work}" ] && [ -d "${work}" ] && rm -rf "${work}"
  find "${lab_dir}" -type d -name __pycache__ -prune -exec rm -rf -- {} + 2>/dev/null
  return 0
}
trap cleanup EXIT INT TERM

check() {
  local label="$1" ok="$2"
  checks=$((checks + 1))
  if [ "${ok}" = "yes" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    failures=$((failures + 1))
  fi
}

check_eq() {
  local label="$1" expected="$2" actual="$3"
  checks=$((checks + 1))
  if [ "${expected}" = "${actual}" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    echo "        expected: ${expected}"
    echo "        actual:   ${actual}"
    failures=$((failures + 1))
  fi
}

python_bin="${PYTHON:-$(command -v python3 || true)}"
if [ -z "${python_bin}" ] || [ ! -x "${python_bin}" ]; then
  echo "python3 was not found. Install Python 3.11+ or set PYTHON=/path/to/python3."
  exit 1
fi

export PYTHONDONTWRITEBYTECODE=1
work="$(mktemp -d)"

# Clear any bytecode cache left by a hand-run of one of the scripts, so that
# the "nothing left behind" check at the end is an assertion about THIS suite
# rather than about whatever happened in this directory earlier.
find "${lab_dir}" -type d -name __pycache__ -prune -exec rm -rf -- {} + 2>/dev/null

# metric FILE NAME — pull one `RESULT name value` line out of a captured run.
metric() {
  grep "^RESULT $2 " "$1" 2>/dev/null | awk '{print $3}' | head -1
}

# ratio_at_least VALUE FLOOR — arithmetic in Python, because bash has no floats.
ratio_at_least() {
  "${python_bin}" -c '
import sys
try:
    print("yes" if float(sys.argv[1]) >= float(sys.argv[2]) else "no")
except (ValueError, IndexError):
    print("no")
' "${1:-}" "${2:-}"
}

ratio_below() {
  "${python_bin}" -c '
import sys
try:
    print("yes" if float(sys.argv[1]) < float(sys.argv[2]) else "no")
except (ValueError, IndexError):
    print("no")
' "${1:-}" "${2:-}"
}

cores="$("${python_bin}" -c 'import os; print(os.cpu_count() or 1)')"
gil_disabled="$("${python_bin}" -c 'import sysconfig; print(sysconfig.get_config_var("Py_GIL_DISABLED"))')"

echo "Day 096 — Concurrency and async Basics"
echo "python3:          $("${python_bin}" -c 'import sys; print(sys.version.split()[0])')"
echo "cpu_count:        ${cores}"
echo "Py_GIL_DISABLED:  ${gil_disabled}"
echo "switch interval:  $("${python_bin}" -c 'import sys; print(sys.getswitchinterval())') s"
echo "work:             a temporary directory, removed when this script exits"
echo

# ---------------------------------------------------------------------------
echo "0. The interpreter is the one these measurements assume"
# ---------------------------------------------------------------------------
check "python is 3.11 or newer (asyncio.timeout and TaskGroup are required)" \
  "$("${python_bin}" -c 'import sys; print("yes" if sys.version_info >= (3, 11) else "no")')"
check "at least two usable CPUs, or the process comparison cannot mean anything" \
  "$([ "${cores}" -ge 2 ] && echo yes || echo no)"
# This is a fact about the build, reported rather than assumed. On a
# free-threaded build (PEP 703) the threads-do-not-help check below is
# expected NOT to hold, and the suite says so instead of failing silently.
if [ "${gil_disabled}" = "1" ]; then
  echo "  note: this is a free-threaded build. The 'threads do not help computing"
  echo "        work' check is skipped, because on this build it should not hold."
fi

# ---------------------------------------------------------------------------
echo
echo "1. Waiting work: threads and an event loop both collapse it"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/01_waiting.py" > "${work}/waiting.txt" 2>&1
waiting_status=$?
check_eq "01_waiting.py exits 0" "0" "${waiting_status}"

seq_s="$(metric "${work}/waiting.txt" waiting_sequential_s)"
thr_speedup="$(metric "${work}/waiting.txt" waiting_threaded_speedup)"
aio_speedup="$(metric "${work}/waiting.txt" waiting_asyncio_speedup)"

check "the sequential baseline really did pay all 20 waits (>= 1.5s of waiting)" \
  "$(ratio_at_least "${seq_s}" 1.5)"
check "threads are at least 4x faster than sequential on waiting work" \
  "$(ratio_at_least "${thr_speedup}" 4.0)"
check "asyncio is at least 4x faster than sequential on waiting work" \
  "$(ratio_at_least "${aio_speedup}" 4.0)"
check "all three approaches returned 20 well-formed bodies" \
  "$([ "$(grep -c '20 bodies, all well formed: yes' "${work}/waiting.txt")" = "3" ] && echo yes || echo no)"
check "order is preserved: the first body is the one for /item/1" \
  "$(grep -q 'first body is for /item/1' "${work}/waiting.txt" && echo yes || echo no)"

# ---------------------------------------------------------------------------
echo
echo "2. Computing work: the answer flips, and that is the whole day"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/02_computing.py" > "${work}/computing.txt" 2>&1
computing_status=$?
check_eq "02_computing.py exits 0" "0" "${computing_status}"

cpu_thr="$(metric "${work}/computing.txt" computing_threaded_speedup)"
cpu_proc="$(metric "${work}/computing.txt" computing_processes_speedup)"
cpu_aio="$(metric "${work}/computing.txt" computing_asyncio_speedup)"

if [ "${gil_disabled}" = "1" ]; then
  check "skipped on a free-threaded build: threads may legitimately help here" yes
else
  check "threads do NOT meaningfully speed up computing work (< 1.5x)" \
    "$(ratio_below "${cpu_thr}" 1.5)"
fi
check "processes DO speed up computing work (>= 1.5x)" \
  "$(ratio_at_least "${cpu_proc}" 1.5)"
check "asyncio does NOT speed up computing work either (< 1.5x)" \
  "$(ratio_below "${cpu_aio}" 1.5)"
check "processes beat threads on this workload by a clear margin" \
  "$("${python_bin}" -c '
import sys
print("yes" if float(sys.argv[1]) >= float(sys.argv[2]) * 1.4 else "no")
' "${cpu_proc}" "${cpu_thr}")"
check "every approach still produced the right answer: 41538 primes below 500,000" \
  "$([ "$(grep -c 'correct: yes' "${work}/computing.txt")" = "4" ] && echo yes || echo no)"
check "the script reports the GIL status of the interpreter it actually ran on" \
  "$(grep -q "Py_GIL_DISABLED ${gil_disabled}" "${work}/computing.txt" && echo yes || echo no)"

# ---------------------------------------------------------------------------
echo
echo "3. A blocking call inside a coroutine, and its two repairs"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/03_blocking_coroutine.py" > "${work}/blocking.txt" 2>&1
blocking_status=$?
check_eq "03_blocking_coroutine.py exits 0" "0" "${blocking_status}"

blocked_s="$(metric "${work}/blocking.txt" blocking_gathered_s)"
await_speedup="$(metric "${work}/blocking.txt" blocking_vs_await_speedup)"
thread_speedup="$(metric "${work}/blocking.txt" blocking_vs_to_thread_speedup)"
blocked_gap="$(metric "${work}/blocking.txt" blocked_heartbeat_gap_ms)"
healthy_gap="$(metric "${work}/blocking.txt" healthy_heartbeat_gap_ms)"

check "gathering 5 blocking coroutines takes the SERIAL time (>= 0.9s for 5 x 0.2s)" \
  "$(ratio_at_least "${blocked_s}" 0.9)"
check "await asyncio.sleep makes the same five overlap (>= 2.5x faster)" \
  "$(ratio_at_least "${await_speedup}" 2.5)"
check "asyncio.to_thread does too, for code you cannot rewrite (>= 2.5x faster)" \
  "$(ratio_at_least "${thread_speedup}" 2.5)"
check "a blocking coroutine starves an unrelated task on the same loop (3x the gap)" \
  "$("${python_bin}" -c '
import sys
print("yes" if float(sys.argv[1]) >= float(sys.argv[2]) * 3 else "no")
' "${blocked_gap}" "${healthy_gap}")"
check "and the healthy loop kept its 10ms heartbeat roughly on time (< 50ms)" \
  "$(ratio_below "${healthy_gap}" 50)"
check "all three versions returned identical correct results — the failure is silent" \
  "$([ "$(grep -c 'correct: yes' "${work}/blocking.txt")" = "3" ] && echo yes || echo no)"

# ---------------------------------------------------------------------------
echo
echo "4. Shared state: a counter that loses increments, and three fixes"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/04_race.py" > "${work}/race.txt" 2>&1
race_status=$?
check_eq "04_race.py exits 0" "0" "${race_status}"

expected_total="$(metric "${work}/race.txt" expected_total)"
lost_tight="$(metric "${work}/race.txt" race_lost_at_tight_interval)"
locked="$(metric "${work}/race.txt" locked_total)"
queued="$(metric "${work}/race.txt" queued_total)"

check_eq "the counter should reach 400000: 8 threads x 50000" "400000" "${expected_total}"
check "the UNPROTECTED counter loses increments — and loses a lot of them (>= 1000)" \
  "$(ratio_at_least "${lost_tight}" 1000)"
check_eq "the LOCKED counter loses none: exactly 400000" "400000" "${locked}"
check_eq "the queue version loses none either: exactly 400000" "400000" "${queued}"
check "two locks taken in opposite orders really deadlock" \
  "$(grep -q 'opposite_lock_order_deadlocks yes' "${work}/race.txt" && echo yes || echo no)"
check "the same two locks taken in one consistent order do not" \
  "$(grep -q 'consistent_lock_order_does_not yes' "${work}/race.txt" && echo yes || echo no)"
check "the script restores the interpreter's switch interval when it is done" \
  "$("${python_bin}" - "${lab_dir}" <<'PY'
import importlib.util
import sys
from pathlib import Path

lab = Path(sys.argv[1])
sys.path.insert(0, str(lab / "examples"))
spec = importlib.util.spec_from_file_location("race", lab / "examples" / "04_race.py")
race = importlib.util.module_from_spec(spec)
spec.loader.exec_module(race)

before = sys.getswitchinterval()
race.unsafe_total(race.TIGHT_INTERVAL)   # runs at 1e-6 internally
after = sys.getswitchinterval()
print("yes" if after == before else f"no: left it at {after}")
PY
)"
check "and it says plainly what the DEFAULT switch interval produced on this machine" \
  "$(grep -q 'RESULT race_lost_at_default_interval' "${work}/race.txt" && echo yes || echo no)"

# ---------------------------------------------------------------------------
echo
echo "5. The scheduler built from generators actually interleaves"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/05_scheduler.py" > "${work}/scheduler.txt" 2>&1
scheduler_status=$?
check_eq "05_scheduler.py exits 0" "0" "${scheduler_status}"

check_eq "three tasks of 3, 2 and 1 steps interleave as a b c a b a" \
  "alpha,beta,gamma,alpha,beta,alpha" \
  "$(metric "${work}/scheduler.txt" round_robin_order)"
check_eq "a task that never yields runs to completion before any other starts" \
  "greedy,greedy,greedy,greedy,polite,polite,polite" \
  "$(metric "${work}/scheduler.txt" greedy_order)"
check "a sleeping task leaves the ready queue and the other task runs meanwhile" \
  "$(grep -q 'napper-start:0 worker:1 worker:2 worker:3 worker:4 napper-woke:5' "${work}/scheduler.txt" && echo yes || echo no)"
check "each task's return value is collected through StopIteration.value" \
  "$(grep -q 'returned: gamma did 1 step$' "${work}/scheduler.txt" && echo yes || echo no)"

# ---------------------------------------------------------------------------
echo
echo "6. Cancellation and timeouts"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/06_timeouts.py" > "${work}/timeouts.txt" 2>&1
timeouts_status=$?
check_eq "06_timeouts.py exits 0" "0" "${timeouts_status}"

timeout_s="$(metric "${work}/timeouts.txt" timeout_elapsed_s)"
survivors="$(metric "${work}/timeouts.txt" gather_survivors)"
cancelled="$(metric "${work}/timeouts.txt" taskgroup_cancelled_siblings)"

check "the timeout fired at the caller's budget, not at the work's length (< 0.35s)" \
  "$(ratio_below "${timeout_s}" 0.35)"
check "and it did fire rather than the request quietly succeeding (>= 0.10s)" \
  "$(ratio_at_least "${timeout_s}" 0.10)"
check "cancellation ran the task's finally block: the socket was closed, not leaked" \
  "$(grep -q 'report cleaned up' "${work}/timeouts.txt" && echo yes || echo no)"
check "the cancelled task saw CancelledError inside itself and re-raised it" \
  "$(grep -q 'report received CancelledError' "${work}/timeouts.txt" && echo yes || echo no)"
check_eq "gather(return_exceptions=True): 2 of 3 finish despite one raising" "2.0000" "${survivors}"
check "and the failure came back as a ValueError VALUE, not as a raise" \
  "$(grep -q '\[1\] ValueError b could not be fetched' "${work}/timeouts.txt" && echo yes || echo no)"
check_eq "TaskGroup: the same failure cancels both siblings instead" "2.0000" "${cancelled}"
check "the TaskGroup error arrives in an ExceptionGroup, caught with except*" \
  "$(grep -q 'caught in the ExceptionGroup: b could not be fetched' "${work}/timeouts.txt" && echo yes || echo no)"

# ---------------------------------------------------------------------------
echo
echo "7. The starter reports honest progress, and the reference completes it"
# ---------------------------------------------------------------------------
before="$(bash "${lab_dir}/starter/02_check.sh" 2>&1)"
before_status=$?
check "the untouched starter reports 0 of 8 exercises complete" \
  "$(printf '%s' "${before}" | grep -q '^0 of 8 exercises complete\.$' && echo yes || echo no)"
check_eq "and exits non-zero, so it cannot be mistaken for finished" "incomplete" \
  "$([ "${before_status}" -ne 0 ] && echo incomplete || echo "exit ${before_status}")"
check "it names the process-pool exercise among the work still to do" \
  "$(printf '%s' "${before}" | grep -q 'count_primes_with_processes' && echo yes || echo no)"

solved="${work}/solved"
mkdir -p "${solved}"
cp -R "${lab_dir}/examples" "${lab_dir}/starter" "${solved}/"
cp "${solved}/examples/07_solutions.py" "${solved}/starter/01_exercises.py"
after="$(bash "${solved}/starter/02_check.sh" 2>&1)"
after_status=$?
check "with the reference answers in place it reports 8 of 8" \
  "$(printf '%s' "${after}" | grep -q '^8 of 8 exercises complete\.$' && echo yes || echo no)"
check_eq "and exits 0" "0" "${after_status}"

# The suite must be able to fail, or it proves nothing. Break exercise 2 so it
# does the work sequentially, keeping every answer correct, and confirm the
# checker refuses it on speed alone.
"${python_bin}" - "${solved}" <<'PY'
import pathlib, sys
target = pathlib.Path(sys.argv[1]) / "starter" / "01_exercises.py"
source = target.read_text(encoding="utf-8")
broken = source.replace(
    "    with ThreadPoolExecutor(max_workers=workers) as pool:\n"
    "        return list(pool.map(labkit.fetch, urls))",
    "    return [labkit.fetch(url) for url in urls]",
    1,
)
assert broken != source, "the sabotage did not apply — the reference file changed shape"
target.write_text(broken, encoding="utf-8")
PY
sabotaged="$(bash "${solved}/starter/02_check.sh" 2>&1)"
check "a 'threaded' answer that is secretly sequential is caught, not waved through" \
  "$(printf '%s' "${sabotaged}" | grep -q '^8 of 8 exercises complete\.$' && echo no || echo yes)"
check "and it is caught on SPEED, not on correctness — the bodies were all right" \
  "$(printf '%s' "${sabotaged}" | grep -q 'fetch_all_with_threads' && \
     printf '%s' "${sabotaged}" | grep -q 'needs >= 2.5x' && echo yes || echo no)"

# ---------------------------------------------------------------------------
echo
echo "8. Hygiene: offline, no sudo, nothing left behind"
# ---------------------------------------------------------------------------
"${python_bin}" - "${lab_dir}" > "${work}/hygiene.txt" <<'PY'
import re
import sys
from pathlib import Path

root = Path(sys.argv[1])
hosts, sudo_lines = set(), []
comment = re.compile(r"^\s*#")
for path in sorted(root.rglob("*")):
    if not path.is_file() or path.suffix not in {".py", ".sh"}:
        continue
    for number, line in enumerate(
        path.read_text(encoding="utf-8", errors="ignore").splitlines(), 1
    ):
        for host in re.findall(r"https?://([^/\s\"')]+)", line):
            # A literal loopback address, or an f-string placeholder that can
            # only ever be filled in from server_address, is fine. Anything
            # else is a URL this lab has no business containing.
            if host != "127.0.0.1" and not host.startswith("{"):
                hosts.add(f"{path.name}:{number} {host}")
        if re.search(r"(^|[;|&(]\s*)sudo\s", line) and not comment.match(line):
            sudo_lines.append(f"{path.name}:{number}")
print("HOSTS " + " ".join(sorted(hosts)))
print("SUDO " + " ".join(sudo_lines))
PY
check_eq "no URL in this lab names any host but the loopback address" \
  "HOSTS" \
  "$(grep '^HOSTS ' "${work}/hygiene.txt" | sed 's/ *$//')"
check_eq "no line in this lab would actually invoke sudo" "SUDO" \
  "$(grep '^SUDO ' "${work}/hygiene.txt" | sed 's/ *$//')"
check "every socket this lab opens is bound to 127.0.0.1" \
  "$(grep -q '"127.0.0.1", 0' "${lab_dir}/examples/labkit.py" && echo yes || echo no)"
check "no captured output leaks an absolute home path" \
  "$(grep -rl '/Users/\|/home/' "${lab_dir}/expected-output" >/dev/null 2>&1 && echo no || echo yes)"
check "no __pycache__ directory survives inside the lab" \
  "$(find "${lab_dir}" -type d -name __pycache__ | grep -q . && echo no || echo yes)"
check "nothing in this lab imports a third-party package" \
  "$("${python_bin}" - "${lab_dir}" <<'PY'
import ast
import sys
from pathlib import Path

STDLIB = set(sys.stdlib_module_names)
LOCAL = {"labkit"}
bad = []
for path in sorted(Path(sys.argv[1]).rglob("*.py")):
    tree = ast.parse(path.read_text(encoding="utf-8"))
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            names = [alias.name.split(".")[0] for alias in node.names]
        elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
            names = [node.module.split(".")[0]]
        else:
            continue
        bad += [n for n in names if n not in STDLIB and n not in LOCAL]
print("yes" if not bad else "no: " + ", ".join(sorted(set(bad))))
PY
)"

echo
echo "${checks} checks, ${failures} failure(s)."
[ "${failures}" -eq 0 ]