asyncio on a microcontroller: tasks and sleeps

tasks/blink_and_sample.py

"""Two tasks share one loop: an LED that changes every 500 ms and a sensor read every 300 ms."""
import machine
import mpy_asyncio as asyncio  # on a board: import asyncio
import mpy_time as time  # on a board: import time

machine.bench_adc(26, [41230, 41180, 40950, 40870])  # the bench: what the sensor reads (not on a board)
led = machine.Pin("LED", machine.Pin.OUT)
sensor = machine.ADC(26)


def log(task, text):
    print("%5d ms  %-6s  %s" % (time.ticks_ms(), task, text))


async def blink(period_ms):
    while True:
        led.toggle()
        log("blink", "LED on" if led.value() else "LED off")
        await asyncio.sleep_ms(period_ms)


async def sample(period_ms):
    while True:
        log("sample", "read_u16() gave %d" % sensor.read_u16())
        await asyncio.sleep_ms(period_ms)


async def main():
    asyncio.create_task(blink(500))
    asyncio.create_task(sample(300))
    log("main", "made two tasks; now sleeps 1,100 ms")
    await asyncio.sleep_ms(1100)
    log("main", "returns, so run() returns")


asyncio.run(main())

    0 ms  main    made two tasks; now sleeps 1,100 ms
    0 ms  blink   LED on
    0 ms  sample  read_u16() gave 41230
  300 ms  sample  read_u16() gave 41180
  500 ms  blink   LED off
  600 ms  sample  read_u16() gave 40950
  900 ms  sample  read_u16() gave 40870
 1000 ms  blink   LED on
 1100 ms  main    returns, so run() returns

mpy_asyncio.py (the model) · tasks/mpy_asyncio.py

"""A model of MicroPython 1.29's asyncio, on the simulated clock of mpy_time.py.

It follows extmod/asyncio of v1.29.0 (core.py, funcs.py, event.py, lock.py) wherever a program can see it:
  - one loop runs one task at a time, and a task gives the loop back only at an await that has to wait;
  - sleep_ms() and sleep() return one shared object, not a coroutine, so that a sleep allocates nothing; create_task(),
    gather() and wait_for() refuse it with TypeError("coroutine expected");
  - run() returns as soon as its own task ends; the other tasks stay queued, nothing cancels them, and a later run()
    goes on with them; new_event_loop() empties the queue;
  - a task that ends with an exception while nothing waits for it prints "Task exception wasn't retrieved" (a board
    prints it to standard error, with the traceback) and the other tasks go on;
  - CancelledError derives from BaseException and TimeoutError from Exception; wait_for() cancels the awaitable when
    its time is up and waits for it to finish; gather() returns the results in the order of its arguments, or raises
    the first exception while the other tasks go on;
  - Event.set() wakes every waiting task; Lock hands itself to the next waiting task; ThreadSafeFlag is one bit that
    an interrupt handler may set and wait() clears, with one waiting task.
The clock moves on only while no task can run, or inside a blocking call (mpy_time.sleep_ms()), so every run prints
the same times. Not MicroPython: a model for computers.
"""
import atexit
import heapq
import inspect

import mpy_time as _time


class CancelledError(BaseException):
    pass


class TimeoutError(Exception):
    pass


_queue = []  # (due, order, task, value): tasks that run once the clock reaches due
_order = 0
_ended = 0
_current = None
_polled = []  # objects the loop polls before each task: a ThreadSafeFlag, a stream with a reader waiting
_ERRNO = {19: "ENODEV", 104: "ECONNRESET", 110: "ETIMEDOUT", 111: "ECONNREFUSED", 113: "EHOSTUNREACH"}



atexit.register(lambda: _close_unstarted([entry[2] for entry in _queue]))


class _Request:
    """What a task hands the loop when it has to wait."""

    def __init__(self, *what):
        self.what = what

    def __await__(self):
        return (yield self.what)


class Task:
    """A coroutine that the loop runs. create_task() makes one; awaiting it waits for its end."""

    def __init__(self, coro):
        self.coro = coro
        self.name = getattr(coro, "__name__", "coro")
        self._done = False
        self._result = None
        self._error = None
        self._ended = 0
        self._waiters = []  # tasks waiting for this one to end
        self._throw = None  # raised inside the task when it next runs: cancel()
        self._value = None  # what the loop sends in when it next runs
        self._joined = None  # the task it awaits, to which cancel() passes on
        self._token = 0  # its live entry in the queue
        self._undo = []  # what to undo when it stops waiting
        self._watched = 0  # gather() or wait_for() collects its result

    def done(self):
        return self._done

    def cancel(self):
        """Raises CancelledError inside the task at the await where it waits. A task that awaits another task passes
        the cancel on to that one, as in MicroPython, and receives its CancelledError when it ends."""
        if self._done:
            return False
        if self is _current:
            raise RuntimeError("can't cancel self")
        target = self
        while target._joined is not None:
            target = target._joined
        target._throw = CancelledError()
        _wake(target)
        return True

    def __await__(self):
        if not self._done:
            yield ("join", self)
        if self._error is not None:
            raise self._error
        return self._result

    def __repr__(self):
        return "<Task>"


def _push(task, due, value=None):
    global _order
    _order += 1
    task._token = _order
    heapq.heappush(_queue, (due, _order, task, value))


def _unpark(task):
    task._joined = None
    undo, task._undo = task._undo, []
    for step in undo:
        step()


def _wake(task, value=None):
    _unpark(task)
    _push(task, _time._now, value)


def _drop(waiting, task):
    if task in waiting:
        waiting.remove(task)


def _unpoll(obj, task):
    if obj._waiter is task:
        obj._waiter = None
    if obj in _polled and obj._waiter is None:
        _polled.remove(obj)


def _park(task, what):
    kind = what[0]
    if kind == "sleep":
        _push(task, what[1])
    elif kind in ("join", "join-any", "join-or-time"):
        others = what[1] if kind == "join-any" else [what[1]]
        for other in others:
            other._waiters.append(task)
            task._undo.append(lambda other=other: _drop(other._waiters, task))
        if kind == "join":
            task._joined = what[1]
        if kind == "join-or-time":
            _push(task, what[2], False)  # still waiting at that time: the time is up
    elif kind == "wait":  # an Event or a Lock: its list of waiting tasks
        what[1].append(task)
        task._undo.append(lambda: _drop(what[1], task))
    elif kind == "poll":  # one slot for a waiting task: a second waiter takes the first one's place
        obj = what[1]
        obj._waiter = task
        if obj not in _polled:
            _polled.append(obj)
        task._undo.append(lambda: _unpoll(obj, task))
    else:
        raise RuntimeError("the model cannot wait for %r" % (what,))


def _text(error):
    if isinstance(error, OSError) and error.args and isinstance(error.args[0], int):
        return "[Errno %d] %s" % (error.args[0], _ERRNO.get(error.args[0], ""))
    return str(error)


def _finish(task, result=None, error=None):
    global _ended
    _ended += 1
    task._done, task._result, task._error, task._ended = True, result, error, _ended
    waiters, task._waiters = task._waiters, []
    for waiter in list(waiters):
        _wake(waiter, True)
    if error is not None and not waiters and not task._watched and not isinstance(error, CancelledError):
        print("Task exception wasn't retrieved")
        print("future: <Task> coro= <generator object '%s'>" % task.name)
        print("%s: %s" % (type(error).__name__, _text(error)))


def _step(task):
    global _current
    _current = task
    try:
        if task._throw is not None:
            error, task._throw = task._throw, None
            what = task.coro.throw(error)
        else:
            what = task.coro.send(task._value)
    except StopIteration as stop:
        _finish(task, result=stop.value)
        return
    except BaseException as error:  # CancelledError too: the task has ended
        _finish(task, error=error)
        return
    finally:
        _current = None
    _park(task, what)


def _poll():
    for obj in list(_polled):
        if obj._waiter is not None and obj._ready():
            _wake(obj._waiter)


def _run_until_complete(main):
    while True:
        _poll()
        if _queue and _queue[0][2]._token != _queue[0][1]:
            heapq.heappop(_queue)  # an entry left behind when the task was woken another way
            continue
        if not _queue or _queue[0][0] > _time._now:
            due = _queue[0][0] if _queue else None
            nxt = _time.next_event()
            if due is None and nxt is None:
                raise RuntimeError("model: every task waits for something that never comes")
            _time._advance(min(t for t in (due, nxt) if t is not None))
            continue
        _, _, task, value = heapq.heappop(_queue)
        task._token = 0
        _unpark(task)
        task._value = value
        _step(task)
        if task is main and main._done:
            if main._error is not None:
                raise main._error
            return main._result


def _promote(aw):
    return aw if isinstance(aw, Task) else create_task(aw)


class _Sleep:
    """What sleep_ms() and sleep() return: one object for every sleep, as in MicroPython. It is not a coroutine."""

    def __init__(self):
        self.due = None

    def __await__(self):
        due, self.due = self.due, None
        yield ("sleep", due)


_sleep = _Sleep()


def sleep_ms(t):
    _sleep.due = _time._now + max(0, int(t))
    return _sleep


def sleep(t):
    return sleep_ms(int(t * 1000))


def create_task(coro):
    if not hasattr(coro, "send"):
        raise TypeError("coroutine expected")
    task = Task(coro)
    _push(task, _time._now)
    return task


def current_task():
    if _current is None:
        raise RuntimeError("no running event loop")
    return _current


def run(coro):
    return _run_until_complete(create_task(coro))


def _close_unstarted(tasks):
    # A coroutine that never started holds no code to run; closing it keeps CPython from warning about it, which a
    # board never does. Coroutines that started are left as they are.
    for task in tasks:
        if inspect.iscoroutine(task.coro) and inspect.getcoroutinestate(task.coro) == inspect.CORO_CREATED:
            task.coro.close()


def new_event_loop():
    """Empties the loop: tasks still queued from an earlier run() are forgotten."""
    global _queue, _polled
    _close_unstarted([entry[2] for entry in _queue])
    _queue, _polled = [], []


async def gather(*aws, return_exceptions=False):
    tasks = [_promote(aw) for aw in aws]
    for t in tasks:
        t._watched += 1
    try:
        while True:
            failed = [t for t in tasks if t._done and t._error is not None]
            if failed and not return_exceptions:
                raise min(failed, key=lambda t: t._ended)._error
            running = [t for t in tasks if not t._done]
            if not running:
                return [t._result if t._error is None else t._error for t in tasks]
            try:
                await _Request("join-any", running)
            except CancelledError:
                for t in running:
                    t.cancel()
                raise
    finally:
        for t in tasks:
            t._watched -= 1


async def _wait_for(aw, timeout):
    task = _promote(aw)
    if timeout is None:
        return await task
    task._watched += 1
    try:
        try:
            ended = await _Request("join-or-time", task, _time._now + max(0, timeout))
        except CancelledError:
            task.cancel()
            raise
        if not ended:
            task.cancel()
            try:
                await task
            except BaseException:  # its CancelledError, or whatever its clean-up raised
                pass
            raise TimeoutError
        return await task
    finally:
        task._watched -= 1


def wait_for(aw, timeout):
    """Returns a coroutine, as in MicroPython: its argument is promoted to a task only when it runs."""
    return _wait_for(aw, None if timeout is None else int(timeout * 1000))


def wait_for_ms(aw, timeout):
    return _wait_for(aw, timeout)


class Event:
    def __init__(self):
        self._set = False
        self._waiting = []

    def is_set(self):
        return self._set

    def set(self):
        """Wakes every waiting task. Call it from a task only, never from an interrupt or another thread."""
        waiting, self._waiting = self._waiting, []
        for task in list(waiting):  # waking a task takes it out of the list
            _wake(task)
        self._set = True

    def clear(self):
        self._set = False

    async def wait(self):
        if not self._set:
            await _Request("wait", self._waiting)
        return True


class ThreadSafeFlag:
    def __init__(self):
        self._state = 0
        self._waiter = None

    def set(self):
        """Safe from an interrupt handler: it only sets a bit, which the loop polls before each task."""
        self._state = 1

    def clear(self):
        self._state = 0

    def _ready(self):
        return self._state == 1

    async def wait(self):
        if not self._state:
            await _Request("poll", self)
        self._state = 0


class Lock:
    def __init__(self):
        self._state = 0  # 0 free, 1 held, or the task it has been handed to
        self._waiting = []

    def locked(self):
        return self._state == 1

    def release(self):
        if self._state != 1:
            raise RuntimeError("Lock not acquired")
        if self._waiting:
            task = self._waiting.pop(0)
            self._state = task
            _wake(task)
        else:
            self._state = 0

    async def acquire(self):
        if self._state != 0:
            try:
                await _Request("wait", self._waiting)
            except CancelledError:
                if self._state is _current:  # handed to this task, then cancelled: pass it on
                    self._state = 1
                    self.release()
                raise
        self._state = 1
        return True

    async def __aenter__(self):
        return await self.acquire()

    async def __aexit__(self, *exc):
        self.release()



def __getattr__(name):
    """Loads the streams the first time a program asks for them, as MicroPython's asyncio loads its stream module."""
    if name in ("open_connection", "start_server", "StreamReader", "StreamWriter"):
        import mpy_stream

        return getattr(mpy_stream, name)
    raise AttributeError(name)

mpy_time.py (the model) · tasks/mpy_time.py

"""The time functions of MicroPython 1.29 that asyncio uses, on a simulated clock (extmod/modtime.c).

asyncio, machine.Timer and these functions share one clock of whole milliseconds, as they share time.ticks_ms() on a
board. ticks_ms() wraps at 2**30 on the rp2 and esp32 ports, like these functions; ticks_diff() and ticks_add() work
across the wrap. sleep_ms() and sleep() block: the clock moves on and nothing else runs except interrupts (the hard
timer callbacks of machine.py), as a blocking call holds a board's whole program. Not MicroPython: a model for
computers. at() is the bench's own: something that happens at a set time, an interrupt or a packet arriving.
"""
import heapq

TICKS_PERIOD = 1 << 30
_now = 0  # milliseconds since the bench started
_events = []  # (time, order, function): what the bench makes happen at a set time
_order = 0


def ticks_ms():
    return _now % TICKS_PERIOD


def ticks_add(ticks, delta):
    return (ticks + delta) % TICKS_PERIOD


def ticks_diff(end, start):
    half = TICKS_PERIOD // 2
    return ((end - start + half) % TICKS_PERIOD) - half


def at(when, function):
    """The bench calls function() when the clock reaches when (a time in milliseconds since the start)."""
    global _order
    _order += 1
    heapq.heappush(_events, (when, _order, function))


def next_event():
    return _events[0][0] if _events else None


def _advance(to):
    """Let the clock run to `to`, firing every interrupt and delivery due on the way, in time order."""
    global _now
    while _events and _events[0][0] <= to:
        when, _, function = heapq.heappop(_events)
        _now = max(_now, when)
        function()
    _now = max(_now, to)


def sleep_ms(ms):
    """Blocks for ms milliseconds: no task runs meanwhile, only interrupts."""
    if ms > 0:
        _advance(_now + ms)


def sleep(seconds):
    sleep_ms(int(seconds * 1000))


def reset():
    global _now, _events, _order
    _now, _events, _order = 0, [], 0


def bench_busy(ms):
    """The bench's own: ms milliseconds of work that never gives the loop back (a computation, a blocking driver)."""
    sleep_ms(ms)

machine.py (the model) · tasks/machine.py

"""A model of the parts of machine that these programs use, on the simulated clock of mpy_time.py (MicroPython 1.29,
rp2 port): Pin outputs, an ADC that returns the bench's readings, and Timer with hard callbacks.

A hard callback runs inside the timer's interrupt at the moment the timer expires, even while a task is in a blocking
call, and an exception it does not catch prints "Uncaught exception in IRQ callback handler" and stops the timer
(ports/rp2/machine_timer.c, shared/runtime/mpirq.c). This model has hard callbacks only. Not MicroPython: a model
for computers. bench_adc() is the bench's own: the readings a sensor will give.
"""
import mpy_time as _time

_adc_values = {}


def bench_adc(channel, values):
    """The readings ADC(channel).read_u16() gives, one per call, starting again after the last."""
    _adc_values[channel] = list(values)


class Pin:
    IN = 0
    OUT = 1

    def __init__(self, id, mode=IN, value=None):
        self.id = id
        self._value = 1 if value else 0

    def value(self, v=None):
        if v is None:
            return self._value
        self._value = 1 if v else 0

    def on(self):
        self._value = 1

    def off(self):
        self._value = 0

    def toggle(self):
        self._value = 1 - self._value

    def __repr__(self):
        return "Pin(%s)" % (self.id,)


class ADC:
    def __init__(self, channel):
        self.channel = channel
        self._n = 0

    def read_u16(self):
        values = _adc_values.get(self.channel, [32768])
        value = values[self._n % len(values)]
        self._n += 1
        return value


class Timer:
    ONE_SHOT = 0
    PERIODIC = 1

    def __init__(self, id=-1, **kwargs):
        self._run = 0  # which init() the queued expiries belong to
        self._active = False
        if kwargs:
            self.init(**kwargs)

    def init(self, *, mode=PERIODIC, period=-1, freq=None, callback=None, hard=False):
        if not hard:
            raise ValueError("this model runs hard callbacks only")
        self._run += 1
        self._period = int(1000 / freq) if freq else period
        self._mode, self._callback, self._active = mode, callback, True
        self._due = _time._now + self._period
        _time.at(self._due, lambda run=self._run: self._expire(run))

    def _expire(self, run):
        if not self._active or run != self._run:
            return
        try:
            self._callback(self)
        except Exception as e:
            print("Uncaught exception in IRQ callback handler")
            print("%s: %s" % (type(e).__name__, e))
            self._active = False
            return
        if self._mode == Timer.PERIODIC:
            self._due += self._period
            _time.at(self._due, lambda: self._expire(run))
        else:
            self._active = False

    def deinit(self):
        self._active = False

MicroPython 1.29 (the model) · tasks/differences.py

"""Four probes of asyncio, on the model of MicroPython 1.29's asyncio. on_cpython.py runs the same on CPython."""
import mpy_asyncio as asyncio  # on a board: import asyncio


async def pause(seconds):
    await asyncio.sleep(seconds)


cleaned_up = False


async def forever():
    global cleaned_up
    try:
        while True:
            await asyncio.sleep(0.01)
    finally:
        cleaned_up = True  # runs only if something cancels or closes the task


async def main():
    try:
        await asyncio.wait_for(asyncio.sleep(1), 0.05)
        print("1. wait_for(sleep(1), 0.05) returned")
    except TypeError as e:
        print("1. wait_for(sleep(1), 0.05): TypeError:", e)
    except asyncio.TimeoutError:
        print("1. wait_for(sleep(1), 0.05): TimeoutError")
    try:
        await asyncio.wait_for(pause(1), 0.05)
    except asyncio.TimeoutError:
        print("2. wait_for(pause(1), 0.05): TimeoutError")
    names = ["sleep_ms", "wait_for_ms", "ThreadSafeFlag", "Queue", "Semaphore", "TaskGroup", "timeout"]
    print("3.", ", ".join("%s %s" % (n, "yes" if hasattr(asyncio, n) else "no") for n in names))
    asyncio.create_task(forever())
    await asyncio.sleep(0.05)


asyncio.run(main())
print("4. after run(), the finally block of forever() has run:", cleaned_up)

1. wait_for(sleep(1), 0.05): TypeError: coroutine expected
2. wait_for(pause(1), 0.05): TimeoutError
3. sleep_ms yes, wait_for_ms yes, ThreadSafeFlag yes, Queue no, Semaphore no, TaskGroup no, timeout no
4. after run(), the finally block of forever() has run: False

CPython 3.14.8 · tasks/on_cpython.py

"""The four probes of differences.py, on CPython's own asyncio."""
import asyncio


async def pause(seconds):
    await asyncio.sleep(seconds)


cleaned_up = False


async def forever():
    global cleaned_up
    try:
        while True:
            await asyncio.sleep(0.01)
    finally:
        cleaned_up = True  # runs only if something cancels or closes the task


async def main():
    try:
        await asyncio.wait_for(asyncio.sleep(1), 0.05)
        print("1. wait_for(sleep(1), 0.05) returned")
    except TypeError as e:
        print("1. wait_for(sleep(1), 0.05): TypeError:", e)
    except asyncio.TimeoutError:
        print("1. wait_for(sleep(1), 0.05): TimeoutError")
    try:
        await asyncio.wait_for(pause(1), 0.05)
    except asyncio.TimeoutError:
        print("2. wait_for(pause(1), 0.05): TimeoutError")
    names = ["sleep_ms", "wait_for_ms", "ThreadSafeFlag", "Queue", "Semaphore", "TaskGroup", "timeout"]
    print("3.", ", ".join("%s %s" % (n, "yes" if hasattr(asyncio, n) else "no") for n in names))
    asyncio.create_task(forever())
    await asyncio.sleep(0.05)


asyncio.run(main())
print("4. after run(), the finally block of forever() has run:", cleaned_up)

1. wait_for(sleep(1), 0.05): TimeoutError
2. wait_for(pause(1), 0.05): TimeoutError
3. sleep_ms no, wait_for_ms no, ThreadSafeFlag no, Queue yes, Semaphore yes, TaskGroup yes, timeout yes
4. after run(), the finally block of forever() has run: True