#!/usr/bin/env python3
"""Experiment 02 — the event loop, cancellation, and structured concurrency.

Predict every output before you run it.

    python3 exp02_async.py

Requires Python 3.11+ for TaskGroup / except*. The script degrades gracefully
and says so if you are on an older interpreter.
"""

from __future__ import annotations

import asyncio
import contextlib
import sys
import time


def section(title: str) -> None:
    print(f"\n{'=' * 70}\n{title}\n{'=' * 70}")


def claim(text: str) -> None:
    print(f"\nCLAIM: {text}")


HAS_TASKGROUP = sys.version_info >= (3, 11)


# ---------------------------------------------------------------------------
async def demo_coroutine_is_not_a_task() -> None:
    section("1. Calling a coroutine function does NOT run it")
    claim("A coroutine object is inert until something schedules it.")

    ran = []

    async def work():
        ran.append("ran")
        return 42

    coro = work()
    print(f"  after calling work():  ran={ran}   <-- nothing happened")
    result = await coro
    print(f"  after awaiting it:     ran={ran}, result={result}")
    print("  A Task is a coroutine WRAPPED so the loop will step it. That is the")
    print("  whole difference, and it is why create_task starts work and await does not.")


# ---------------------------------------------------------------------------
async def demo_gather_orphans() -> None:
    section("2. gather() orphans its siblings when one fails")
    claim("gather with return_exceptions=False propagates the first exception "
          "immediately but does NOT cancel the other tasks. They keep running.")

    finished = []

    async def boom():
        await asyncio.sleep(0.01)
        raise ValueError("boom")

    async def slow():
        try:
            await asyncio.sleep(0.2)
            finished.append("slow COMPLETED")
        except asyncio.CancelledError:
            finished.append("slow CANCELLED")
            raise

    slow_task = asyncio.create_task(slow())
    try:
        await asyncio.gather(boom(), slow_task)
    except ValueError as exc:
        print(f"  gather raised   -> {exc}")
        print(f"  slow_task.done()-> {slow_task.done()}   <-- still running")

    await asyncio.sleep(0.3)
    print(f"  afterwards      -> {finished}")
    print("  That orphan is a RESOURCE LEAK, not a style preference: it holds")
    print("  connections, writes to a store you thought you rolled back, and")
    print("  outlives the scope that created it.")


# ---------------------------------------------------------------------------
async def demo_taskgroup() -> None:
    section("3. TaskGroup cancels siblings and aggregates errors")
    if not HAS_TASKGROUP:
        print("  (needs Python 3.11+; skipping)")
        return

    claim("A failing child cancels the remaining children and the group raises "
          "an ExceptionGroup.")

    finished = []

    async def boom(tag: str):
        await asyncio.sleep(0.01)
        raise ValueError(f"boom-{tag}")

    async def slow():
        try:
            await asyncio.sleep(0.5)
            finished.append("slow COMPLETED")
        except asyncio.CancelledError:
            finished.append("slow CANCELLED")
            raise

    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(boom("a"))
            group.create_task(slow())
    except* ValueError as eg:
        print(f"  caught ExceptionGroup with {len(eg.exceptions)} error(s): "
              f"{[str(e) for e in eg.exceptions]}")

    print(f"  sibling outcome -> {finished}   <-- cancelled, not orphaned")
    print("  This is structured concurrency: no task outlives its scope.")


# ---------------------------------------------------------------------------
async def demo_cancellation() -> None:
    section("4. CancelledError is a BaseException")
    claim("Since 3.8 asyncio.CancelledError inherits BaseException, so a broad "
          "`except Exception` does NOT swallow cancellation.")

    print(f"  issubclass(CancelledError, Exception)     -> "
          f"{issubclass(asyncio.CancelledError, Exception)}")
    print(f"  issubclass(CancelledError, BaseException) -> "
          f"{issubclass(asyncio.CancelledError, BaseException)}")

    state = []

    async def well_behaved():
        try:
            await asyncio.sleep(10)
        except Exception:
            state.append("swallowed by except Exception")
        except asyncio.CancelledError:
            state.append("cancellation observed, re-raising")
            raise
        finally:
            state.append("finally ran")

    task = asyncio.create_task(well_behaved())
    await asyncio.sleep(0.01)
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task
    print(f"  -> {state}")

    claim("Catching CancelledError WITHOUT re-raising makes a task uncancellable.")

    async def uncancellable():
        # Bounded so this demo terminates; the bug is the bare `pass`.
        for _ in range(15):
            try:
                await asyncio.sleep(0.02)
            except asyncio.CancelledError:
                pass  # the bug: cancellation observed and discarded
        return "finished on its own terms"

    task = asyncio.create_task(uncancellable())
    await asyncio.sleep(0.05)
    task.cancel()
    await asyncio.sleep(0.1)
    print(f"  cancel(), then 100ms later task.done() -> {task.done()}   "
          f"<-- refused to die")
    result = await task
    print(f"  it eventually returned normally: {result!r}")
    print("  A task that swallows CancelledError cannot be shut down, cannot be")
    print("  timed out, and will keep a TaskGroup's __aexit__ blocked forever.")


# ---------------------------------------------------------------------------
async def demo_fire_and_forget() -> None:
    section("5. Fire-and-forget tasks can be garbage collected")
    claim("The event loop keeps only a WEAK reference to a task. Without a strong "
          "reference of your own, a task can vanish mid-execution.")

    print("  asyncio.create_task(f())          # <-- no reference kept: a real bug")
    print("  The documented fix is to keep a strong reference:")
    print("      tasks = set()")
    print("      t = asyncio.create_task(f()); tasks.add(t)")
    print("      t.add_done_callback(tasks.discard)")
    print("  Or better: use a TaskGroup, which holds them for you and awaits them.")

    done = []
    tasks: set[asyncio.Task] = set()
    for i in range(3):
        t = asyncio.create_task(asyncio.sleep(0.01))
        tasks.add(t)
        t.add_done_callback(lambda fut, i=i: done.append(i))
        t.add_done_callback(tasks.discard)
    await asyncio.gather(*tasks)
    print(f"  with strong refs, all completed -> {sorted(done)}")


# ---------------------------------------------------------------------------
async def demo_blocking_call() -> None:
    section("6. One blocking call stalls the entire loop")
    claim("time.sleep in a coroutine blocks the loop thread: no other task runs, "
          "no I/O is polled, no timer fires.")

    ticks = []

    async def ticker():
        for _ in range(20):
            await asyncio.sleep(0.01)
            ticks.append(time.perf_counter())

    async def blocking_offender():
        await asyncio.sleep(0.02)
        time.sleep(0.15)  # the bug

    ticks.clear()
    start = time.perf_counter()
    tick_task = asyncio.create_task(ticker())
    await blocking_offender()
    await tick_task
    gaps = [b - a for a, b in zip(ticks, ticks[1:])]
    print(f"  ticker ran {len(ticks)} times in {time.perf_counter() - start:.3f}s")
    print(f"  largest gap between ticks: {max(gaps) * 1000:.0f} ms "
          f"(expected ~10 ms)  <-- the stall")

    claim("asyncio.to_thread moves the blocking work off the loop thread.")
    ticks.clear()
    start = time.perf_counter()
    tick_task = asyncio.create_task(ticker())
    await asyncio.sleep(0.02)
    await asyncio.to_thread(time.sleep, 0.15)
    await tick_task
    gaps = [b - a for a, b in zip(ticks, ticks[1:])]
    print(f"  with to_thread, largest gap: {max(gaps) * 1000:.0f} ms  <-- loop kept turning")


# ---------------------------------------------------------------------------
async def demo_async_generators() -> None:
    section("7. Async generators need explicit closing")
    claim("An async generator's finally block runs at aclose(). Relying on garbage "
          "collection means cleanup happens at an unpredictable time, or during "
          "interpreter shutdown.")

    log = []

    async def streaming_rows():
        try:
            for i in range(100):
                yield i
                await asyncio.sleep(0)
        finally:
            log.append("connection released")

    async with contextlib.aclosing(streaming_rows()) as rows:
        async for row in rows:
            if row >= 2:
                break
    print(f"  after `async with aclosing(...)` and an early break -> {log}")
    print("  Without aclosing, that finally runs whenever the GC gets to it —")
    print("  which for a database cursor is a connection leak under load.")


# ---------------------------------------------------------------------------
async def main() -> None:
    await demo_coroutine_is_not_a_task()
    await demo_gather_orphans()
    await demo_taskgroup()
    await demo_cancellation()
    await demo_fire_and_forget()
    await demo_blocking_call()
    await demo_async_generators()
    print("\n" + "=" * 70)
    print("Interview form of every claim above lives in ../README.md § B2.")
    print("=" * 70)


if __name__ == "__main__":
    asyncio.run(main())