MicroPython for Microcontrollers Module 6 – Concurrency on devices: asyncio and threads
Streams, timeouts and cancellation for I/O
Call and serve TCP with MicroPython asyncio streams, give every wait a deadline with wait_for_ms, back off between retries, and clean up on cancel.
What you will learn
- Call and serve a TCP service with asyncio.open_connection() and asyncio.start_server()
- Limit every network wait with wait_for_ms(), and explain why a deadline cannot cut a blocking lookup short
- Schedule reconnects with a capped exponential back-off
- Release sockets and switch outputs off in finally when a task is cancelled
Before you start
What you’ll learn · Before you start
What you will learn
- Call and serve a TCP service with asyncio.open_connection() and asyncio.start_server()
- Limit every network wait with wait_for_ms(), and explain why a deadline cannot cut a blocking lookup short
- Schedule reconnects with a capped exponential back-off
- Release sockets and switch outputs off in finally when a task is cancelled
Before you start
On this page
A Pico W that asks a server for its settings with a plain await reader.readline() waits for ever the day the server
accepts the connection and never answers. A board that retries at once after each failure makes 1,200 attempts while
a server is down for 12 seconds, and one that ends each attempt with writer.close() alone leaves every socket open.
This lesson talks TCP with asyncio’s streams, gives every network wait a deadline: The latest moment a wait may end. asyncio.wait_for_ms() gives a wait one: if the awaited call has not finished by then, it is cancelled and the caller receives asyncio.TimeoutError., spaces
reconnects with a back-off that has a cap, and cleans up in finally when a task is cancelled. The programs run on the
same model of MicroPython 1.29’s asyncio as the earlier lessons, with a simulated network in which a packet takes
5 ms each way; on a Pico W or an ESP32 the same code runs once Wi-Fi is up.
A stream is a socket that you read and write by awaiting
A client asks a small server on the local network for a temperature: it connects, sends one request line and its
headers, and reads a status line, headers and a JSON body. Every step that waits for the network is an await, so
other tasks run meanwhile. asyncio.open_connection(host, port) connects and returns a reader and a writer, which on
MicroPython are one and the same stream: In asyncio, the object for one TCP connection that tasks read and write with awaits: readline(), read() and readexactly(), and write() followed by drain(). MicroPython uses one Stream as both reader and writer. object.
asyncio.start_server(handler, host, port) listens, and runs handler(reader, writer) as a task of its own for
every client that connects. Step through one request:
The drawing shows the whole scene; its 6 steps are listed below.
Every step as text
- Connect. open_connection("192.168.4.10", 8080) connects in one round trip, 10 ms, while other tasks run. It returns one Stream object twice, as the reader and the writer.
- A task for the client. The server accepted the connection, so start_server() runs handle(reader, writer) as a new task, which awaits readline() for the request.
- Write, then drain. At 10 ms the client writes the request and awaits drain(), which waits until the bytes are sent and gives the loop a turn either way.
- The server answers. At 15 ms the request line arrives. handle() skips the headers, writes the status line, the headers and the JSON body, drains, and closes its end with wait_closed().
- Read to the end. At 20 ms the client reads the status line and the headers with readline(), and the body with read(), which returns once the server has closed. It parses 200 and {'temp_c': 23.5}.
- After the end. One more readline() returns b'' at once, which is how a stream says that the other end has closed. readexactly(n) would raise EOFError instead.
The program runs the server and the client side by side, the server standing for a computer on the network:
streams/tcp_temp.py
"""A TCP server and a client on asyncio streams: the client asks for /temp and gets a JSON reply."""
import json
import mpy_asyncio as asyncio # on a board: import asyncio
import mpy_time as time # on a board: import time
def log(text):
print("%5d ms %s" % (time.ticks_ms(), text))
async def handle(reader, writer): # one task for each client
request = await reader.readline() # b"GET /temp HTTP/1.0\r\n"
while await reader.readline() not in (b"\r\n", b""):
pass # skip the headers, up to the empty line
path = request.split()[1].decode()
log("server %s asked for %s" % (writer.get_extra_info("peername")[0], path))
writer.write(b"HTTP/1.0 200 OK\r\nContent-Type: application/json\r\n\r\n")
writer.write(json.dumps({"temp_c": 23.5}).encode())
await writer.drain()
await writer.wait_closed()
async def get_temp(host, port):
reader, writer = await asyncio.open_connection(host, port)
log("client connected, sends GET /temp")
writer.write(("GET /temp HTTP/1.0\r\nHost: %s\r\n\r\n" % host).encode())
await writer.drain()
status = int((await reader.readline()).split()[1].decode())
while await reader.readline() not in (b"\r\n", b""):
pass
body = json.loads((await reader.read()).decode())
log("client got %d %s" % (status, body))
log("client readline() after the end gives %r" % await reader.readline())
await writer.wait_closed()
async def main():
server = await asyncio.start_server(handle, "0.0.0.0", 8080)
log("server listening on port 8080")
await get_temp("192.168.4.10", 8080)
server.close()
await server.wait_closed()
asyncio.run(main())Output
0 ms server listening on port 8080
10 ms client connected, sends GET /temp
15 ms server 192.168.4.20 asked for /temp
20 ms client got 200 {'temp_c': 23.5}
20 ms client readline() after the end gives b''
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 tcp_temp.py
mpy_stream.py (the model) · streams/mpy_stream.py
"""asyncio's TCP streams in MicroPython 1.29 (extmod/asyncio/stream.py), on the simulated clock of mpy_time.py and a
simulated network. mpy_asyncio loads this file the first time a program asks for open_connection or start_server,
as MicroPython's asyncio loads its stream module.
What it keeps from MicroPython:
- open_connection() looks the host name up first, and the lookup blocks every task (the source marks
socket.getaddrinfo() as blocking); it returns one Stream twice, as (reader, writer), and raises OSError when the
connection is refused;
- write() sends at once, and drain() always gives the loop a turn;
- readline() returns b"" at the end of the stream, read() reads to the end, readexactly(n) raises EOFError when
the stream ends first;
- close() does nothing; wait_closed() is what closes the socket;
- start_server() runs the callback in a task of its own for each client and returns a Server.
The network: a packet arrives 5 ms after it is sent, and connecting takes one round trip, 10 ms. bench_dns() says how
long a host name takes to look up and what it gives; bench_open() counts the sockets that open_connection() made and
nothing has closed. Not MicroPython: a model for computers.
"""
import mpy_asyncio as _aio
import mpy_time as _time
LATENCY_MS = 5
_listeners = {} # port: (address it is bound to, Server)
_dns = {} # host name: (milliseconds the lookup takes, address)
_ports = [50000]
def bench_dns(name, ms, address):
_dns[name] = (ms, address)
def bench_open():
return sum(1 for s in _Socket.made if s.client and not s.closed)
class _Socket:
made = []
def __init__(self, client):
self.client = client
self.inbox = bytearray()
self.eof = False # the other end has closed
self.closed = False
self.peer = None
self._waiter = None # a task waiting to read, which the loop polls
_Socket.made.append(self)
def _ready(self):
return bool(self.inbox) or self.eof or self.closed
def send(self, data):
peer = self.peer
_time.at(_time._now + LATENCY_MS, lambda: peer.inbox.extend(data))
def close(self):
if not self.closed:
self.closed = True
peer = self.peer
_time.at(_time._now + LATENCY_MS, lambda: setattr(peer, "eof", True))
def _lookup(host):
if host.replace(".", "").isdigit():
return host
if host not in _dns:
raise OSError(-2) # the name is not known
ms, address = _dns[host]
_time.sleep_ms(ms) # blocking: no task runs while the lookup waits
return address
class Stream:
def __init__(self, sock, peername):
self.s = sock
self._peername = peername
def get_extra_info(self, v):
if v != "peername":
raise ValueError("only peername")
return self._peername
async def _wait(self):
await _aio._Request("poll", self.s)
async def read(self, n=-1):
while True:
if n >= 0 and self.s.inbox or self.s.eof or self.s.closed:
if n < 0 and not (self.s.eof or self.s.closed):
await self._wait()
continue
size = len(self.s.inbox) if n < 0 else min(n, len(self.s.inbox))
data = bytes(self.s.inbox[:size])
del self.s.inbox[:size]
return data
await self._wait()
async def readexactly(self, n):
while len(self.s.inbox) < n:
if self.s.eof or self.s.closed:
raise EOFError
await self._wait()
data = bytes(self.s.inbox[:n])
del self.s.inbox[:n]
return data
async def readline(self):
while True:
end = self.s.inbox.find(b"\n")
if end >= 0 or self.s.eof or self.s.closed:
size = end + 1 if end >= 0 else len(self.s.inbox)
line = bytes(self.s.inbox[:size])
del self.s.inbox[:size]
return line
await self._wait()
def write(self, buf):
self.s.send(bytes(buf))
async def drain(self):
await _aio.sleep_ms(0)
def close(self):
pass # as in MicroPython: the socket stays open until wait_closed()
async def wait_closed(self):
self.s.close()
aclose = wait_closed
StreamReader = Stream
StreamWriter = Stream
async def open_connection(host, port, ssl=None):
address = _lookup(host)
await _aio.sleep_ms(2 * LATENCY_MS) # one round trip
bound = _listeners.get(port)
if bound is None or bound[0] not in ("0.0.0.0", address):
raise OSError(111) # ECONNREFUSED
client, server = _Socket(True), _Socket(False)
client.peer, server.peer = server, client
_ports[0] += 1
here = ("192.168.4.20", _ports[0])
_aio.create_task(bound[1].callback(Stream(server, here), Stream(server, here)))
stream = Stream(client, (address, port))
return stream, stream
class Server:
def __init__(self, callback, port):
self.callback = callback
self.port = port
def close(self):
_listeners.pop(self.port, None)
async def wait_closed(self):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
self.close()
await self.wait_closed()
async def start_server(callback, host, port, backlog=5, ssl=None):
server = Server(callback, port)
_listeners[port] = (host, server)
await _aio.sleep_ms(0)
return servermpy_asyncio.py (the model) · streams/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) · streams/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)Runs on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
The whole exchange took 20 ms: the connect at 10 ms, the request at 15 ms, the reply at 20 ms. write() puts the
bytes on their way, and drain() waits until they are sent and gives the loop a turn either way, so call it after each
write(). readline() reads up to and including b"\n", and read() with no size reads until the other end closes.
At the end, readline() returns b"" rather than raising, while readexactly(n) raises EOFError if the stream ends
before n bytes. Since MicroPython 1.27, start_server() also accepts IPv6 addresses.
The host-name lookup blocks every task
Before it connects, open_connection() looks the host name up with socket.getaddrinfo(), which the source marks as
blocking: while a slow DNS server answers, no task runs. Look the name up once at start-up and keep the address, or
give an IP address, as these programs do.
Checkpoint
Check: Reading and writing streams
Every network wait needs a deadline
A server that accepts the connection and never answers leaves await reader.readline() waiting for ever, and the task
with it. asyncio.wait_for_ms(aw, ms) puts a limit on any wait: it runs aw as a task and waits at most ms
milliseconds. When time is up it cancels aw, which receives CancelledError at the await where it waits and can
clean up, waits for it to finish, and raises asyncio.TimeoutError in the task that called it. Step through a
2,000 ms deadline on a server that never answers, then a deadline on a lookup:
The drawing shows the whole scene; its 6 steps are listed below.
Every step as text
- The call becomes a task. At 10 ms wait_for_ms(read_reply(reader), 2000) makes a task of read_reply(), which awaits readline() on a server that will never answer.
- Two ways to wake. The waiting task can wake for two reasons, the end of read_reply() or the deadline at 2,010 ms. Had a reply come first, wait_for_ms() would return its result.
- The deadline comes. At 2,010 ms the deadline is first. wait_for_ms() cancels read_reply(), which receives CancelledError at its readline(), logs that it can clean up, and ends.
- TimeoutError. wait_for_ms() waits for that clean-up to finish, then raises asyncio.TimeoutError in the task that called it, which catches it and closes the connection.
- A lookup that blocks. A deadline acts only when the loop runs. open_connection("sensors.example.com", 8080) looks the name up with a call that blocks for 3,000 ms, so the 1,000 ms deadline cannot fire until that call returns.
- Late by 2 seconds. TimeoutError comes after 3,000 ms instead of 1,000, and a 100 ms heartbeat was 2,910 ms late. Only an await can be cut short.
The program makes both waits, with a 100 ms heartbeat task that records how late it ever woke:
"""Deadlines on network waits: a server that never answers, and a host-name lookup that blocks every task."""
import mpy_asyncio as asyncio # on a board: import asyncio
import mpy_stream # the bench: the model's network (not on a board)
import mpy_time as time # on a board: import time
mpy_stream.bench_dns("sensors.example.com", 3000, "192.168.4.10") # the bench: a lookup that takes 3 s
def log(text):
print("%5d ms %s" % (time.ticks_ms(), text))
async def silent(reader, writer): # a server that accepts, reads the request and never answers
await reader.readline()
await asyncio.sleep_ms(60000)
async def read_reply(reader):
try:
return await reader.readline()
except asyncio.CancelledError:
log("read_reply() was cancelled, and can clean up here")
raise
async def heartbeat(stats):
while True:
due = time.ticks_add(time.ticks_ms(), 100)
await asyncio.sleep_ms(100)
stats["late"] = max(stats["late"], time.ticks_diff(time.ticks_ms(), due))
async def main():
stats = {"late": 0}
asyncio.create_task(heartbeat(stats))
await asyncio.start_server(silent, "0.0.0.0", 8080)
reader, writer = await asyncio.open_connection("192.168.4.10", 8080)
writer.write(b"GET /temp HTTP/1.0\r\n\r\n")
await writer.drain()
try:
await asyncio.wait_for_ms(read_reply(reader), 2000)
except asyncio.TimeoutError:
log("no reply within 2,000 ms: TimeoutError")
await writer.wait_closed()
log("heartbeat: worst lateness so far %d ms" % stats["late"])
start = time.ticks_ms()
try:
await asyncio.wait_for_ms(asyncio.open_connection("sensors.example.com", 8080), 1000)
except asyncio.TimeoutError:
log("lookup with a 1,000 ms deadline: TimeoutError after %d ms" % time.ticks_diff(time.ticks_ms(), start))
await asyncio.sleep_ms(150)
log("heartbeat: worst lateness so far %d ms" % stats["late"])
asyncio.run(main())Output
2010 ms read_reply() was cancelled, and can clean up here 2010 ms no reply within 2,000 ms: TimeoutError 2010 ms heartbeat: worst lateness so far 0 ms 5010 ms lookup with a 1,000 ms deadline: TimeoutError after 3000 ms 5160 ms heartbeat: worst lateness so far 2910 ms
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 timeouts.py
Runs on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
The silent server cost exactly its deadline: read_reply() was cancelled at 2,010 ms, logged that it could clean up,
and the caller caught TimeoutError at the same moment, while the heartbeat had never been late. The lookup is the
other story. Its deadline was 1,000 ms, but TimeoutError came after 3,000 ms, when the blocking call returned, and
the heartbeat woke 2,910 ms late. A deadline acts only when the loop runs, at an await; it cannot cut short a call
that holds the loop.
Key idea
Give every network wait a deadline with wait_for_ms(), and keep blocking calls out of the paths it guards: a
deadline can only end an await.
Checkpoint
Check: Deadlines with wait_for_ms()
Reconnects back off, up to a cap
When a server is down, the next try is unlikely to work a few milliseconds later, yet a loop that retries at once keeps the radio busy and adds load to a server that may be struggling to come back. An exponential back-off: Waiting longer after each failure in a row, twice as long each time, such as 1, 2, 4 and 8 seconds, so that repeated failures slow down instead of repeating at full speed. More in “Watchdogs, reset causes and safe recovery” waits longer after each failure, and a cap keeps the longest wait bounded. The program reconnects to a server that is down for 12 seconds, first at once after each refusal, then waiting 500 ms and doubling, never more than 4 s:
"""Reconnecting to a server that is down for 12 seconds: at once after each failure, or with a capped back-off."""
import mpy_asyncio as asyncio # on a board: import asyncio
import mpy_time as time # on a board: import time
async def sink(reader, writer): # the server, once it is back
await writer.wait_closed()
async def server_down_for(ms, servers):
await asyncio.sleep_ms(ms)
servers.append(await asyncio.start_server(sink, "0.0.0.0", 8080))
async def connect(back_off):
attempts = []
delay = 500
while True:
attempts.append(time.ticks_ms())
try:
reader, writer = await asyncio.open_connection("192.168.4.10", 8080)
connected = time.ticks_ms()
await writer.wait_closed()
return attempts, connected
except OSError:
if back_off:
await asyncio.sleep_ms(delay)
delay = min(delay * 2, 4000) # 500, 1000, 2000, then 4000 at most
else:
await asyncio.sleep_ms(0)
async def trial(back_off):
servers = []
asyncio.create_task(server_down_for(12000, servers))
result = await connect(back_off)
await asyncio.sleep_ms(20) # let the server's task finish
servers[0].close() # down again for the next trial
return result
for back_off, label in [(False, "at once"), (True, "back-off 500 ms doubling, at most 4 s")]:
asyncio.new_event_loop()
start = time.ticks_ms()
attempts, connected = asyncio.run(trial(back_off))
print("%-38s %4d attempts, connected at %5d ms" % (label, len(attempts), time.ticks_diff(connected, start)))
if back_off:
print(" attempts at:", ", ".join("%d" % time.ticks_diff(t, start) for t in attempts), "ms")Output
at once 1200 attempts, connected at 12000 ms back-off 500 ms doubling, at most 4 s 7 attempts, connected at 15570 ms attempts at: 0, 510, 1520, 3530, 7540, 11550, 15560 ms
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 backoff.py
Runs on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
Retrying at once made 1,200 attempts, one for each 10 ms round trip, and connected the moment the server came back. The back-off made 7 attempts, at 0, 510, 1,520, 3,530, 7,540, 11,550 and 15,560 ms, and connected at 15,570 ms: 3.57 s after the server was back, the cost of trying 1,193 times less. The cap sets that cost: with a cap of 4 s, a board is never more than about 4 s late in noticing that the server is back. Press Edit, cap the waits at 1,000 ms and run it again: 14 attempts, and the board connects 640 ms after the server is back.
Start again after a success
Reset the delay to its first value after every successful connection, so that the next failure starts with a short wait, and give each try its own deadline, so that a try that hangs counts as a failure.
Checkpoint
Check: Capped exponential back-off
Cancellation arrives at an await, so clean up in finally
A pump task switches a relay on, connects to a server and reports the flow every 100 ms. When the program shuts down,
main calls task.cancel(). cancellation: Ending a task from outside with task.cancel(). CancelledError is raised inside the task at the await where it waits, so that its except and finally blocks can clean up before it ends. does not stop a task on the spot: the loop
raises CancelledError inside it at the await where it waits, and the task’s own except and finally blocks run on
the way out. Step through it:
The drawing shows the whole scene; its 5 steps are listed below.
Every step as text
- The pump runs. The pump task switched the relay on, connected, and now writes a flow report and awaits sleep_ms(100), over and over.
- cancel(). At 300 ms main calls task.cancel(). Nothing happens inside the pump yet; the loop will raise CancelledError there when it runs the pump next.
- At the await. The pump was waiting in sleep_ms(100), so that is where CancelledError appears. It is not an Exception but a BaseException, so it goes past any except Exception and on to finally.
- finally cleans up. The finally block switches the relay off and awaits writer.wait_closed(), which on MicroPython is the call that closes the socket; close() alone does nothing.
- main sees the end. The CancelledError ends the pump task, and main's await task receives it. main catches it, then finds the relay at 0 and no socket left open.
The program cancels the pump, then makes five short reports that end with close() alone and five that also await
wait_closed(), and last cancels a task whose loop has a bare except::
streams/cleanup.py
"""Cancelling a task that holds a relay and a socket: finally does the clean-up; close() alone leaks the socket."""
import machine
import mpy_asyncio as asyncio # on a board: import asyncio
import mpy_stream # the bench: the model's network (not on a board)
import mpy_time as time # on a board: import time
relay = machine.Pin(15, machine.Pin.OUT)
def log(text):
print("%5d ms %s" % (time.ticks_ms(), text))
async def sink(reader, writer): # the server: reads lines until the client closes
while await reader.readline():
pass
await writer.wait_closed()
async def pump():
relay.on()
writer = None
try:
reader, writer = await asyncio.open_connection("192.168.4.10", 8080)
while True:
writer.write(b"flow 12\n")
await writer.drain()
await asyncio.sleep_ms(100)
finally: # runs on cancel(), on an error and on return alike
relay.off()
if writer is not None:
await writer.wait_closed()
log("pump finally: relay off, socket closed")
async def report_once(close_properly):
reader, writer = await asyncio.open_connection("192.168.4.10", 8080)
writer.write(b"flow 12\n")
await writer.drain()
writer.close() # does nothing on MicroPython
if close_properly:
await writer.wait_closed() # this closes the socket
stop = False
passes = 0
async def stubborn():
global passes
while not stop:
passes += 1
try:
await asyncio.sleep_ms(100)
except: # noqa: E722 - the mistake: this also catches CancelledError
pass
async def main():
await asyncio.start_server(sink, "0.0.0.0", 8080)
task = asyncio.create_task(pump())
await asyncio.sleep_ms(300)
log("main task.cancel()")
task.cancel()
try:
await task
except asyncio.CancelledError:
log("main the pump task ended with CancelledError")
log("main relay %d, sockets open %d" % (relay.value(), mpy_stream.bench_open()))
for close_properly, how in ((False, "close() only"), (True, "close() and wait_closed()")):
before = mpy_stream.bench_open()
for _ in range(5):
await report_once(close_properly)
log("main 5 reports with %s: %d more sockets left open" % (how, mpy_stream.bench_open() - before))
global stop
task = asyncio.create_task(stubborn())
await asyncio.sleep_ms(50)
task.cancel()
at_cancel = passes
await asyncio.sleep_ms(500)
log("main the task with a bare except made %d more passes after cancel()" % (passes - at_cancel))
stop = True # the only way left to end it
await asyncio.sleep_ms(100)
asyncio.run(main())Output
300 ms main task.cancel() 300 ms pump finally: relay off, socket closed 300 ms main the pump task ended with CancelledError 300 ms main relay 0, sockets open 0 350 ms main 5 reports with close() only: 5 more sockets left open 400 ms main 5 reports with close() and wait_closed(): 0 more sockets left open 950 ms main the task with a bare except made 5 more passes after cancel()
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 cleanup.py
machine.py (the model) · streams/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 = FalseRuns on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
The pump’s finally ran at 300 ms, so the relay ended at 0 and no socket stayed open. The reports show a trap of
MicroPython’s streams: close() does nothing, and wait_closed() is the call that closes the socket, so the five
reports with close() alone left five sockets open, while CPython’s close() would have started closing each of
them. A board that reconnects every minute this way runs out of sockets in time. Last, the bare except: caught the
CancelledError and the task went on, five more passes after cancel().
Catching everything around an await
CancelledError derives from BaseException, as in CPython since 3.8, so except Exception: lets it through and
the task ends. A bare except: or except BaseException: stops it, and the task can no longer be cancelled.
Catch the errors you mean, and if you must catch everything, raise it again.
So every task that holds something, a pin that must go off, a socket, a lock, gets a try around its work and a
finally that gives the thing back. It runs on a cancel, on an error and on a normal return alike, and on
MicroPython it awaits wait_closed() for every stream.
Key idea
cancel() raises CancelledError at the task’s await; a finally block that switches outputs off and awaits wait_closed() is what makes the end of a task safe.
Checkpoint
Check: Clean-up when a task is cancelled
Practise: fetch JSON with a deadline and a back-off
Write fetch_json(), the client a sensor node uses to fetch its settings: a deadline on every connect and read,
retries with a capped back-off, a reply that is not JSON raised at once, and every connection closed. The sample tests
play a scripted server on a virtual clock and check when each try happened:
Exercise · Hard · Python
Fetch JSON with a deadline and a back-off
A sensor node fetches its settings from a small server on the local network. Write two functions in fetch.py:
parse_reply(data)takes a whole HTTP/1.0 reply as bytes and returns(status, body): the status code as an integer and the body parsed as JSON. It raisesValueErrorwhendatahas no empty line after the headers, does not start with a status line such asHTTP/1.0 200 OK, or has a body that is not JSON.fetch_json(host, port, path, timeout_ms=2000, retries=3, base_ms=500, cap_ms=4000)connects withasyncio.open_connection(), sendsGET <path> HTTP/1.0with aHostheader and an empty line, reads the whole reply withread()and returnsparse_reply()of it.
Every connect and every read gets a deadline of timeout_ms with asyncio.wait_for_ms(). When a try fails with OSError or asyncio.TimeoutError, wait and try again, up to retries more times: first base_ms, then twice as long each time, never more than cap_ms. After the last try, raise its error. A reply that parse_reply() refuses is raised at once, without another try. Close every connection you opened with await writer.wait_closed(), whatever happens.
The sample tests replace asyncio inside fetch.py with a stand-in that runs on a virtual clock and plays a scripted server: one that answers, one that never answers, one whose connect never completes, and one that refuses.
Starter code · fetch.py
import asyncio
import json
def parse_reply(data):
"""(status, body) from a whole HTTP/1.0 reply in bytes, the body parsed as JSON.
Raises ValueError when data is not such a reply, or when its body is not JSON.
"""
async def fetch_json(host, port, path, timeout_ms=2000, retries=3, base_ms=500, cap_ms=4000):
"""GET path from host:port and return parse_reply() of the reply.
Every connect and every read has a deadline of timeout_ms. On OSError or asyncio.TimeoutError, wait and try
again, up to retries more times, waiting base_ms, then twice as long each time, never more than cap_ms; after the
last try, raise the error. A reply that parse_reply() refuses is raised at once. Close every connection.
"""The sample tests · test_fetch.py
"""Sample tests for fetch.py. A stand-in for asyncio runs fetch_json() on a virtual clock and gives it a scripted
server: each connection attempt takes the next behaviour of the script. Its TimeoutError is asyncio's own class, as
on MicroPython, so catch asyncio.TimeoutError. On a board, fetch.py runs unchanged on MicroPython's asyncio."""
import fetch
OK = b'HTTP/1.0 200 OK\r\nContent-Type: application/json\r\n\r\n{"temp_c": 23.5}'
BAD = b"HTTP/1.0 200 OK\r\n\r\n{temp_c: 23.5"
FOREVER = 10**9
class Wake:
def __init__(self, at):
self.at = at
def __await__(self):
yield self.at
class Deadline:
"""wait_for_ms(): runs the awaitable until it finishes or its time is up; then cancels it and raises."""
def __init__(self, fake, aw, deadline):
self.fake, self.aw, self.deadline = fake, aw, deadline
def __await__(self):
steps = self.aw.__await__()
try:
wake = steps.send(None)
while wake <= self.deadline:
yield wake
wake = steps.send(None)
except StopIteration as done:
return done.value
yield self.deadline
try:
steps.throw(self.fake.CancelledError())
except (self.fake.CancelledError, StopIteration):
pass
raise self.fake.TimeoutError()
class FakeStream:
def __init__(self, fake, reply_at, reply):
self.fake, self.reply_at, self.reply = fake, reply_at, reply
self.sent = b""
self.closed = False
def write(self, data):
self.sent += data
async def drain(self):
pass
async def read(self, n=-1):
await Wake(self.reply_at)
return self.reply
def close(self):
pass # as on MicroPython: only wait_closed() closes
async def wait_closed(self):
self.closed = True
class FakeAsyncio:
"""sleep_ms(), wait_for_ms() and open_connection() of asyncio, on a virtual clock in milliseconds."""
TimeoutError = type("TimeoutError", (Exception,), {})
CancelledError = type("CancelledError", (BaseException,), {})
def __init__(self, script):
self.now = 0
self.script = list(script)
self.streams = []
self.attempts = []
def sleep_ms(self, ms):
return Wake(self.now + ms)
def sleep(self, seconds):
return Wake(self.now + int(seconds * 1000))
def wait_for_ms(self, aw, timeout):
return Deadline(self, aw, self.now + timeout)
async def open_connection(self, host, port):
self.attempts.append(self.now)
what = self.script.pop(0)
if what[0] == "slow": # the connect itself never completes
await Wake(FOREVER)
await Wake(self.now + 10) # one round trip
if what[0] == "refuse":
raise OSError(111) # ECONNREFUSED
reply_at = FOREVER if what[0] == "silent" else self.now + what[1]
stream = FakeStream(self, reply_at, what[2] if what[0] == "reply" else b"")
self.streams.append(stream)
return stream, stream
def run(self, coro):
while True:
try:
wake = coro.send(None)
except StopIteration as done:
return done.value
self.now = max(self.now, wake)
def bench(script):
fake = FakeAsyncio(script)
fetch.asyncio = fake # the stand-in replaces asyncio inside fetch.py
return fake
def test_parse_reply():
"""parse_reply() gives the status and the JSON body, and refuses what is not a whole reply"""
assert fetch.parse_reply(OK) == (200, {"temp_c": 23.5})
assert fetch.parse_reply(b"HTTP/1.0 404 Not Found\r\n\r\n{}") == (404, {})
for bad in (b"hello", b"HTTP/1.0 200 OK\r\n", b"FTP/1.0 200 OK\r\n\r\n{}", b"HTTP/1.0\r\n\r\n{}", BAD):
try:
fetch.parse_reply(bad)
except ValueError:
continue
raise AssertionError("parse_reply(%r) did not raise ValueError" % (bad,))
def test_first_try():
"""answers on the first try: one GET, the reply parsed, the connection closed"""
fake = bench([("reply", 30, OK)])
assert fake.run(fetch.fetch_json("192.168.4.10", 8080, "/temp")) == (200, {"temp_c": 23.5})
assert fake.now == 40
assert fake.streams[0].sent.startswith(b"GET /temp HTTP/1.0\r\n")
assert fake.streams[0].closed
def test_silent_then_reply():
"""a server that never answers times out after 2,000 ms; the second try, 500 ms later, succeeds"""
fake = bench([("silent",), ("reply", 30, OK)])
assert fake.run(fetch.fetch_json("192.168.4.10", 8080, "/temp")) == (200, {"temp_c": 23.5})
assert fake.attempts == [0, 2510]
assert fake.now == 2550
assert all(s.closed for s in fake.streams)
def test_connect_timeout():
"""a connect that never completes also times out, and nothing is left open"""
fake = bench([("slow",), ("reply", 30, OK)])
assert fake.run(fetch.fetch_json("192.168.4.10", 8080, "/temp", timeout_ms=1000)) == (200, {"temp_c": 23.5})
assert fake.attempts == [0, 1500]
assert all(s.closed for s in fake.streams)
def test_back_off_capped():
"""refused three times: waits of 500, 1,000 and 1,000 ms with a cap of 1,000"""
fake = bench([("refuse",), ("refuse",), ("refuse",), ("reply", 30, OK)])
assert fake.run(fetch.fetch_json("192.168.4.10", 8080, "/temp", cap_ms=1000)) == (200, {"temp_c": 23.5})
assert fake.attempts == [0, 510, 1520, 2530]
def test_gives_up():
"""after retries more tries, the last error is raised"""
fake = bench([("refuse",)] * 4)
try:
fake.run(fetch.fetch_json("192.168.4.10", 8080, "/temp"))
except OSError:
pass
else:
raise AssertionError("fetch_json() did not raise OSError when every try was refused")
assert fake.attempts == [0, 510, 1520, 3530]
def test_bad_reply_not_retried():
"""a reply that is not JSON raises ValueError at once, with no second try"""
fake = bench([("reply", 30, BAD), ("reply", 30, OK)])
try:
fake.run(fetch.fetch_json("192.168.4.10", 8080, "/temp"))
except ValueError:
pass
else:
raise AssertionError("fetch_json() did not raise ValueError for a reply that is not JSON")
assert fake.attempts == [0]
assert fake.streams[0].closedA hint
Loop over the tries. Put the connect, the write and the read of one try inside a try: its except decides whether there is another try, and its finally closes what that try opened. Sleep and double the delay after the try, so a try that returns never waits.
Results of the sample tests
| Test | Result | Details |
|---|
What your code printed
The sample tests run on this device, in your browser (Pyodide): nothing is sent to mysmartcopilot.com. The first run downloads Python (about 13.5 MB), which is kept for the next runs. A check in your browser is feedback for you, not proof that the code is right for every input.
Then test yourself on streams, deadlines, back-off and cancellation. The questions start where your answers have been weakest so far:
Practice that adapts to you
Practise this lesson
13 questions about 4 ideas of this lesson. Each answer is explained at once, and what you get right is kept in this browser.
Interview questions
Warm-up (fresher to mid-level): In MicroPython’s asyncio, what is the difference between writer.close() and
await writer.wait_closed()?
On MicroPython, close() does nothing at all, and wait_closed() is the call that closes the socket. In the source of
MicroPython 1.29’s streams, close() is an empty method and wait_closed() calls the socket’s own close(); it does
not wait to send anything still buffered, so await writer.drain() comes first. On CPython it is the other way round:
close() starts closing and wait_closed() waits until that is done. Code written for CPython that calls only
close() therefore leaks one socket per connection on a board, and a device that reconnects often runs out of
sockets. The habit that is right on both: await writer.drain(), then writer.close() and
await writer.wait_closed(), in a finally block.
Two longer questions on network tasks start from this lesson:
Model answers open with any pass: ₹79 for 1 week ($2.99), one payment that never renews.
-
Mid to senior
An ESP32 posts a reading to a server every minute with asyncio streams. When the server is down for an hour, the board stops answering its button, its LED freezes now and then, and after a while it resets. Explain what is going on, and design the client so that a dead server costs no more than a log line.
-
Fresher to mid
On shutdown, an app cancels a task that has a pump relay switched on and a socket open. Walk through what happens from task.cancel() to the end of the task, and name the mistakes that leave the pump running.
Flashcards · 8 cards
Recall it, then check
Say the answer to yourself before you look: recalling it is what makes it stick.
-
What does asyncio.open_connection() return on MicroPython?
A reader and a writer that are one and the same stream object. start_server() runs handler(reader, writer) as a task of its own for every client that connects.
-
Why call await writer.drain() after write()?
write() only puts the bytes on their way; drain() waits until they are sent, and gives the loop a turn either way.
-
How do readline() and readexactly(n) behave at the end of a stream?
readline() returns b"" rather than raising. readexactly(n) raises EOFError if the stream ends before n bytes have come.
-
What does asyncio.wait_for_ms(aw, ms) do when the time is up?
It cancels aw, which gets CancelledError at its await and can clean up, waits for it to end, then raises asyncio.TimeoutError in the caller.
-
Why can no deadline cut short the host-name lookup of open_connection()?
The lookup, socket.getaddrinfo(), blocks the loop, and a deadline can act only at an await. Look the name up once at start-up, or connect to an IP address.
-
Why do reconnects back off exponentially, up to a cap?
Waiting longer after each failure spares the radio and a server that is trying to come back, and the cap bounds how late the board notices that it is back. Reset the delay after a success.
-
What does writer.close() do on MicroPython?
Nothing: wait_closed() is the call that closes the socket. Code that calls only close() leaks a socket for every connection, so a finally does drain(), close() and await wait_closed().
-
Why does a bare except: around an await make a task impossible to cancel?
CancelledError derives from BaseException, so a bare except: or except BaseException catches it and the task goes on; except Exception: lets it through.
Key takeaways
open_connection()returns one stream as both reader and writer,start_server()runs a task per client,drain()follows everywrite(), andreadline()returnsb""at the end.wait_for_ms()cancels the wait when its deadline passes and raisesasyncio.TimeoutError, but it can end only an await: a blocking call such as the host-name lookup holds every task past the deadline.- Reconnects wait longer after each failure, up to a cap; 7 tries instead of 1,200 cost 3.57 s of delay in noticing that the server was back.
cancel()raisesCancelledErrorat the task’s await;finallyswitches outputs off and awaitswait_closed(), because on MicroPythonclose()alone leaves the socket open, and a bareexcept:makes a task impossible to cancel.
Terms in this lesson
- cancellation
- Ending a task from outside with task.cancel(). CancelledError is raised inside the task at the await where it waits, so that its except and finally blocks can clean up before it ends.
- deadline
- The latest moment a wait may end. asyncio.wait_for_ms() gives a wait one: if the awaited call has not finished by then, it is cancelled and the caller receives asyncio.TimeoutError.
- stream
- In asyncio, the object for one TCP connection that tasks read and write with awaits: readline(), read() and readexactly(), and write() followed by drain(). MicroPython uses one Stream as both reader and writer.
References
- asyncio – TCP stream connections, wait_for_ms() and Task.cancel() (MicroPython project)
- open_connection(), start_server() and Stream (extmod/asyncio/stream.py, v1.29.0) (MicroPython project)
- wait_for() and wait_for_ms() (extmod/asyncio/funcs.py, v1.29.0) (MicroPython project)
- CancelledError and TimeoutError (extmod/asyncio/core.py, v1.29.0) (MicroPython project)
- Python 3.8 differences: asyncio.CancelledError derives from BaseException (MicroPython project)
- socket – getaddrinfo() (MicroPython project)
- MicroPython v1.27.0 release notes (IPv6 in asyncio.start_server) (MicroPython project)
- CPython: Streams, StreamWriter.close() and wait_closed() (Python 3.14) (Python Software Foundation)
- RFC 1945: Hypertext Transfer Protocol, HTTP/1.0 (IETF)
Related tools
Report a problem with this lesson
Kept only in this browser. Your Learn progress