"""A single-threaded event loop: a ready queue, a timer heap, and tasks that
are generators driven by send().

The shape follows CPython's `asyncio.base_events.BaseEventLoop._run_once`
closely enough to reproduce its timing, which is the point: the toy exists to
show *where a timer's lateness comes from*, and the answer is entirely in the
queueing, not in the clock.

Time is an integer count of virtual milliseconds. `Loop.clock` moves for
exactly two reasons:

  1. a task declares that it burned CPU (`yield burn(ms)`), or
  2. nothing is runnable, so the loop idles forward to the next timer.

There is no `time.monotonic()` anywhere. A run is a pure function of its
tasks, so every transcript on the commentary page is byte-identical on every
machine.

A coroutine here is a generator that yields exactly two commands:

    yield burn(ms)    # "I just consumed ms of CPU, and here is a yield point"
    yield sleep(ms)   # "wake me no earlier than now + ms"

`burn` is the model of a chunk of ordinary Python between two awaits: it
costs time and then hands control back, i.e. `do_work(); await sleep(0)`.
"""

import heapq
from collections import deque, namedtuple

BURN, SLEEP = "burn", "sleep"

# One timer delivery, in the three stages it actually has:
#   deadline    -- when the sleeper asked to be woken
#   expired_at  -- when the loop's sweep noticed the timer had come due
#   fired_at    -- when the timer's own callback got the CPU
#   ran_at      -- when the sleeping task's code resumed
# Every millisecond of lateness in this toy accrues between the last three.
Wake = namedtuple("Wake", "task deadline expired_at fired_at ran_at")


def burn(ms):
    return (BURN, ms)


def sleep(ms):
    return (SLEEP, ms)


class Timer:
    """An armed callback plus the clock reading at which the loop expired it.

    `expired_at` exists only so the demo can show that the timer heap itself
    is never the guilty party.
    """

    __slots__ = ("when", "callback", "expired_at")

    def __init__(self, when, callback):
        self.when = when
        self.callback = callback
        self.expired_at = None


class Future:
    """A slot for a result plus the one callback waiting on it.

    Resolving a future does *not* resume the waiter. It schedules the waiter,
    so the waiter runs on a later pass of the ready queue. That indirection is
    the second hop, and it is the reason `await asyncio.sleep(1.0)` is
    delivered later than a bare `loop.call_later(1.0, cb)` asking for the
    same instant.
    """

    def __init__(self, loop):
        self.loop = loop
        self.done = False
        self.waiter = None

    def set_result(self):
        self.done = True
        if self.waiter is not None:
            waiter, self.waiter = self.waiter, None
            self.loop.call_soon(waiter)          # hop 2


class Task:
    """A generator plus the bookkeeping to keep resuming it."""

    def __init__(self, loop, name, coro):
        self.loop = loop
        self.name = name
        self.coro = coro
        self.deadline = None      # set while suspended on a timer
        self.expired_at = None
        self.fired_at = None
        loop.call_soon(self.step)

    def step(self):
        """Resume the generator once and dispose of whatever it yields.

        Entering `step` *is* the delivery: `self.loop.clock` right now is when
        this task's own code got the CPU, which is the only reading its author
        would ever be able to observe.
        """
        if self.deadline is not None:
            self.loop.trace.append(Wake(self.name, self.deadline,
                                        self.expired_at, self.fired_at,
                                        self.loop.clock))
            self.deadline = self.expired_at = self.fired_at = None

        try:
            kind, arg = self.coro.send(None)
        except StopIteration:
            return

        if kind == BURN:
            self.loop.clock += arg
            self.loop.call_soon(self.step)
        elif kind == SLEEP:
            self._suspend(arg)
        else:
            raise ValueError(f"unknown command {kind!r}")

    def _suspend(self, delay):
        """Model of `asyncio.sleep(delay)` for delay > 0.

        CPython does not arm a timer that resumes the task. It arms a timer
        whose callback resolves a *future*, and awaits that future:

            h = loop.call_later(delay, futures._set_result_unless_cancelled,
                                future, result)
            return await future

        (verbatim shape of `inspect.getsource(asyncio.sleep)`). So the wakeup
        costs two trips through the ready queue, not one.

        `loop.hops == 1` is the counterfactual: the armed callback resumes the
        task itself, which is what `loop.call_later(delay, cb)` gives you.
        """
        loop = self.loop
        self.deadline = loop.clock + delay
        fut = Future(loop)
        fut.waiter = self.step

        def fire():                   # hop 1: the callback the timer armed
            self.expired_at = timer.expired_at
            self.fired_at = loop.clock
            if loop.hops == 2:
                fut.set_result()      # -> call_soon(self.step), a second trip
            else:
                self.step()           # resumed in place, no second trip

        timer = loop.call_later(delay, fire)


class Loop:
    """The scheduler. `hops` selects how a timer reaches its sleeper: 2 is
    faithful to `asyncio.sleep`, 1 is faithful to `loop.call_later`.
    """

    def __init__(self, hops=2):
        self.clock = 0            # integer virtual milliseconds
        self.ready = deque()      # callbacks runnable right now
        self.timers = []          # heap of (when, seq, Timer)
        self.seq = 0              # tie-break, so equal deadlines stay FIFO
        self.hops = hops
        self.trace = []

    def call_soon(self, callback):
        """Queue a callback for the next pass, at the *back*. A task that
        reschedules itself never jumps ahead of a waiting timer -- and a
        timer never jumps ahead of already-queued work either.
        """
        self.ready.append(callback)

    def call_later(self, delay, callback):
        timer = Timer(self.clock + delay, callback)
        heapq.heappush(self.timers, (timer.when, self.seq, timer))
        self.seq += 1
        return timer

    def spawn(self, name, coro):
        return Task(self, name, coro)

    def run_once(self):
        """One pass, in CPython's order: expire timers, then run a snapshot
        of the ready queue.
        """
        while self.timers and self.timers[0][0] <= self.clock:
            _, _, timer = heapq.heappop(self.timers)
            timer.expired_at = self.clock       # noticed exactly on time
            self.ready.append(timer.callback)   # ...and queued behind the rest

        # A SNAPSHOT, not a drain. Anything these callbacks schedule waits for
        # the next pass. Change `for _ in range(len(self.ready))` to
        # `while self.ready:` and a task that reschedules itself every step
        # never lets the loop reach a timer again -- silently, forever.
        for _ in range(len(self.ready)):
            self.ready.popleft()()

    def run(self, until):
        """Run until the virtual clock reaches `until`, or nothing is left."""
        while self.clock < until:
            if not self.ready:
                if not self.timers:
                    return
                # Nothing runnable: idle forward to the next deadline. This is
                # the only place the clock moves without a task having burned
                # CPU, and it is why a lone sleeper is never late.
                self.clock = max(self.clock, self.timers[0][0])
            self.run_once()


def deliveries(loop, name):
    """Every timer delivery for one task, as raw Wake rows."""
    return [w for w in loop.trace if w.task == name]


def periods(loop, name):
    """Observed gaps between consecutive resumptions of one task."""
    runs = [w.ran_at for w in loop.trace if w.task == name]
    return [b - a for a, b in zip(runs, runs[1:])]