This tutorial is intended for users having varying levels of experience with asyncio and includes a section for complete beginners.
- Introduction
0.1 Installing uasyncio on bare metal - Cooperative scheduling
1.1 Modules - uasyncio
2.1 Program structure: the event loop
2.2 Coroutines (coros)
2.2.1 Queueing a coro for scheduling
2.2.2 Running a callback function
2.2.3 Notes Coros as bound methods. Returning values.
2.3 Delays - Synchronisation
3.1 Lock
3.1.1 Locks and timeouts
3.2 Event
3.2.1 The event's value
3.3 Barrier
3.4 Semaphore
3.4.1 BoundedSemaphore
3.5 Queue
3.6 Task cancellation
3.7 Other synchronisation primitives - Designing classes for asyncio
4.1 Awaitable classes
4.1.1 Use in context managers
4.1.2 Awaiting a coro
4.2 Asynchronous iterators
4.3 Asynchronous context managers
4.4 Coroutines with timeouts
4.5 Exceptions - Interfacing hardware
5.1 Timing issues
5.2 Polling hardware with a coroutine
5.3 Using the stream mechanism
5.3.1 A UART driver example
5.4 Writing streaming device drivers
5.5 A complete example: aremote.py A driver for an IR remote control receiver.
5.6 Driver for HTU21D A temperature and humidity sensor. - Hints and tips
6.1 Program hangs
6.2 uasyncio retains state
6.3 Garbage Collection
6.4 Testing
6.5 A common error This can be hard to find.
6.6 Socket programming
6.7 Event loop constructor args - Notes for beginners
7.1 Problem 1: event loops
7.2 Problem 2: blocking methods
7.3 The uasyncio approach
7.4 Scheduling in uasyncio
7.5 Why cooperative rather than pre-emptive?
7.6 Communication
7.7 Polling
Most of this document assumes some familiarity with asynchronous programming. For those new to it an introduction may be found in section 7.
The MicroPython uasyncio library comprises a subset of Python's asyncio
library. It is designed for use on microcontrollers. As such it has a small RAM
footprint and fast context switching with zero RAM allocation. This document
describes its use with a focus on interfacing hardware devices. The aim is to
design drivers in such a way that the application continues to run while the
driver is awaiting a response from the hardware. The application remains
responsive to events and to user interaction.
Another major application area for asyncio is in network programming: many guides to this may be found online.
Note that MicroPython is based on Python 3.4 with minimal Python 3.5 additions.
Except where detailed below, asyncio features of versions >3.4 are
unsupported. As stated above it is a subset; this document identifies supported
features.
This tutorial aims to present a consistent programming style compatible with CPython V3.5 and above.
MicroPython libraries are located on PyPi. Libraries to be installed are:
- micropython-uasyncio
- micropython-uasyncio.queues
- micropython-uasyncio.synchro
The queues and synchro modules are optional, but are required to run all
the examples below.
The official approach is to use the upip utility as described
here. Network enabled
hardware has this included in the firmware so it can be run locally. This is
the preferred approach.
On non-networked hardware there are two options. One is to use upip under a
Linux real or virtual machine. This involves installing and building the Unix
version of MicroPython, using upip to install to a directory on the PC, and
then copying the library to the target.
The need for Linux and the Unix build may be avoided by using
micropip.py.
This runs under Python 3.2 or above. Create a temporary directory on your PC
and install to that. Then copy the contents of the temporary directory to the
device. The following assume Linux and a temporary directory named ~/syn -
adapt to suit your OS. The first option requires that micropip.py has
executable permission.
$ ./micropip.py install -p ~/syn micropython-uasyncio
$ python3 -m micropip.py install -p ~/syn micropython-uasyncio
The uasyncio modules may be frozen as bytecode in the usual way, by placing
the uasyncio and collections directories in the port's modules directory
and rebuilding.
The technique of cooperative multi-tasking is widely used in embedded systems. It offers lower overheads than pre-emptive scheduling and avoids many of the pitfalls associated with truly asynchronous threads of execution.
The following modules are provided which may be copied to the target hardware.
Libraries
- asyn.py Provides synchronisation primitives
Lock,Event,Barrier,Semaphore,BoundedSemaphore,Conditionandgather. Provides support for task cancellation viaNamedTaskandCancellableclasses. - aswitch.py Provides classes for interfacing switches and pushbuttons and also a software retriggerable delay object. Pushbuttons are a generalisation of switches providing logical rather than physical status along with double-clicked and long pressed events.
Demo Programs
The first two are the most immediately rewarding as they produce visible results by accessing Pyboard hardware.
- aledflash.py Flashes the four Pyboard LEDs asynchronously for 10s. The simplest uasyncio demo. Import it to run.
- apoll.py A device driver for the Pyboard accelerometer. Demonstrates the use of a coroutine to poll a device. Runs for 20s. Import it to run.
- astests.py Test/demonstration programs for the aswitch module.
- asyn_demos.py Simple task cancellation demos.
- roundrobin.py Demo of round-robin scheduling. Also a benchmark of scheduling performance.
- awaitable.py Demo of an awaitable class. One way of implementing a device driver which polls an interface.
- chain.py Copied from the Python docs. Demo of chaining coroutines.
- aqtest.py Demo of uasyncio
Queueclass. - aremote.py Example device driver for NEC protocol IR remote control.
- auart.py Demo of streaming I/O via a Pyboard UART.
- auart_hd.py Use of the Pyboard UART to communicate with a device using a half-duplex protocol. Suits devices such as those using the 'AT' modem command set.
- iorw.py Demo of a read/write device driver using the stream I/O mechanism.
Test Programs
- asyntest.py Tests for the synchronisation primitives in asyn.py.
- cantest.py Task cancellation tests.
Utility
- check_async_code.py A Python3 utility to locate a particular coding error which can be hard to find. See para 6.5.
Benchmarks
The benchmarks directory contains scripts to test and characterise the
uasyncio scheduler. See this doc.
The asyncio concept is of cooperative multi-tasking based on coroutines, referred to in this document as coros or tasks.
Consider the following example:
import uasyncio as asyncio
async def bar():
count = 0
while True:
count += 1
print(count)
await asyncio.sleep(1) # Pause 1s
loop = asyncio.get_event_loop()
loop.create_task(bar()) # Schedule ASAP
loop.run_forever()Program execution proceeds normally until the call to loop.run_forever. At
this point execution is controlled by the scheduler. A line after
loop.run_forever would never be executed. The scheduler runs bar
because this has been placed on the scheduler's queue by loop.create_task.
In this trivial example there is only one coro: bar. If there were others,
the scheduler would schedule them in periods when bar was paused.
Most embedded applications have an event loop which runs continuously. The event
loop can also be started in a way which permits termination, by using the event
loop's run_until_complete method; this is mainly of use in testing. Examples
may be found in the astests.py module.
The event loop instance is a singleton, instantiated by a program's first call
to asyncio.get_event_loop(). This takes two optional integer args being the
lengths of the two coro queues. Typically both will have the same value being
at least the number of concurrent coros in the application. The default of 16
is usually sufficient. If using non-default values see
Event loop constructor args.
If a coro needs to call an event loop method (usually create_task), calling
asyncio.get_event_loop() (without args) will efficiently return it.
A coro is instantiated as follows:
async def foo(delay_secs):
await asyncio.sleep(delay_secs)
print('Hello')A coro can allow other coroutines to run by means of the await coro
statement. A coro must contain at least one await statement. This causes
coro to run to completion before execution passes to the next instruction.
Consider these lines of code:
await asyncio.sleep(delay_secs)
await asyncio.sleep(0)The first causes the code to pause for the duration of the delay, with other
coros being scheduled for the duration. A delay of 0 causes any pending coros
to be scheduled in round-robin fashion before the following line is run. See
the roundrobin.py example.
EventLoop.create_taskArg: the coro to run. The scheduler queues the coro to run ASAP. Thecreate_taskcall returns immediately. The coro arg is specified with function call syntax with any required arguments passed.EventLoop.run_until_completeArg: the coro to run. The scheduler queues the coro to run ASAP. The coro arg is specified with function call syntax with any required arguments passed. Therun_until_completecall returns when the coro terminates: this method provides a way of quitting the scheduler.awaitArg: the coro to run, specified with function call syntax. Starts the coro ASAP. The awaiting coro blocks until the awaited one has run to completion.
The above are compatible with CPython. Additional uasyncio methods are discussed in 2.2.3 below.
Callbacks should be Python functions designed to complete in a short period of time. This is because coroutines will have no opportunity to run for the duration.
The following EventLoop methods schedule callbacks:
call_soonCall as soon as possible. Args:callbackthe callback to run,*argsany positional args may follow separated by commas.call_laterCall after a delay in secs. Args:delay,callback,*argscall_later_msCall after a delay in ms. Args:delay,callback,*args.
loop = asyncio.get_event_loop()
loop.call_soon(foo, 5) # Schedule callback 'foo' ASAP with an arg of 5.
loop.call_later(2, foo, 5) # Schedule after 2 seconds.
loop.call_later_ms(50, foo, 5) # Schedule after 50ms.
loop.run_forever()A coro can contain a return statement with arbitrary return values. To
retrieve them issue:
result = await my_coro()Coros may be bound methods. A coro must contain at least one await statement.
Where a delay is required in a coro there are two options. For longer delays and those where the duration need not be precise, the following should be used:
async def foo(delay_secs, delay_ms):
await asyncio.sleep(delay_secs)
print('Hello')
await asyncio.sleep_ms(delay_ms)While these delays are in progress the scheduler will schedule other coros. This is generally highly desirable, but it does introduce uncertainty in the timing as the calling routine will only be rescheduled when the one running at the appropriate time has yielded. The amount of latency depends on the design of the application, but is likely to be on the order of tens or hundreds of ms; this is discussed further in Section 5.
Very precise delays may be issued by using the utime functions sleep_ms
and sleep_us. These are best suited for short delays as the scheduler will
be unable to schedule other coros while the delay is in progress.
There is often a need to provide synchronisation between coros. A common example is to avoid what are known as "race conditions" where multiple coros compete to access a single resource. An example is provided in the astests.py program and discussed in the docs. Another hazard is the "deadly embrace" where two coros each wait on the other's completion.
In simple applications communication may be achieved with global flags or bound
variables. A more elegant approach is to use synchronisation primitives. The
module
asyn.py
offers "micro" implementations of Event, Barrier, Semaphore and
Condition primitives. These are for use only with asyncio. They are not
thread safe and should not be used with the _thread module or from an
interrupt handler except where mentioned. A Lock primitive is provided which
is an alternative to the official implementation.
Another synchronisation issue arises with producer and consumer coros. The
producer generates data which the consumer uses. Asyncio provides the Queue
object. The producer puts data onto the queue while the consumer waits for its
arrival (with other coros getting scheduled for the duration). The Queue
guarantees that items are removed in the order in which they were received.
Alternatively a Barrier instance can be used if the producer must wait
until the consumer is ready to access the data.
The following provides a brief overview of the primitives. Full documentation may be found here.
This describes the use of the official Lock primitive.
This guarantees unique access to a shared resource. In the following code
sample a Lock instance lock has been created and is passed to all coros
wishing to access the shared resource. Each coro attempts to acquire the lock,
pausing execution until it succeeds.
import uasyncio as asyncio
from uasyncio.synchro import Lock
async def task(i, lock):
while 1:
await lock.acquire()
print("Acquired lock in task", i)
await asyncio.sleep(0.5)
lock.release()
async def killer():
await asyncio.sleep(10)
loop = asyncio.get_event_loop()
lock = Lock() # The global Lock instance
loop.create_task(task(1, lock))
loop.create_task(task(2, lock))
loop.create_task(task(3, lock))
loop.run_until_complete(killer()) # Run for 10sAt time of writing (5th Jan 2018) the official Lock class is not complete.
If a coro is subject to a timeout
and the timeout is triggered while it is waiting on a lock, the timeout will be
ineffective. It will not receive the TimeoutError until it has acquired the
lock. The same observation applies to task cancellation.
The module asyn.py offers a Lock class which works in these
situations see docs. It is significantly less
efficient than the official class but supports additional interfaces as per the
CPython version including context manager usage.
This provides a way for one or more coros to pause until another flags them to
continue. An Event object is instantiated and made accessible to all coros
using it:
import asyn
event = asyn.Event()Coros waiting on the event issue await event whereupon execution pauses until
another issues event.set(). Full details.
This presents a problem if event.set() is issued in a looping construct; the
code must wait until the event has been accessed by all waiting coros before
setting it again. In the case where a single coro is awaiting the event this
can be achieved by the receiving coro clearing the event:
async def eventwait(event):
await event
event.clear()The coro raising the event checks that it has been serviced:
async def foo(event):
while True:
# Acquire data from somewhere
while event.is_set():
await asyncio.sleep(1) # Wait for coro to respond
event.set()Where multiple coros wait on a single event synchronisation can be achieved by means of an acknowledge event. Each coro needs a separate event.
async def eventwait(event, ack_event):
await event
ack_event.set()An example of this is provided in the event_test function in asyntest.py.
This is cumbersome. In most cases - even those with a single waiting coro - the
Barrier class below offers a simpler approach.
An Event can also provide a means of communication between an interrupt handler and a coro. The handler services the hardware and sets an event which is tested in slow time by the coro.
The event.set() method can accept an optional data value of any type. A
coro waiting on the event can retrieve it by means of event.value(). Note
that event.clear() will set the value to None. A typical use for this
is for the coro setting the event to issue event.set(utime.ticks_ms()). Any
coro waiting on the event can determine the latency incurred, for example to
perform compensation for this.
This has two uses. Firstly it can cause a coro to pause until one or more other coros have terminated.
Secondly it enables multiple coros to rendezvous at a particular point. For
example producer and consumer coros can synchronise at a point where the
producer has data available and the consumer is ready to use it. At that point
in time the Barrier can run an optional callback before the barrier is
released and all waiting coros can continue. Full details.
The callback can be a function or a coro. In most applications a function is likely to be used: this can be guaranteed to run to completion before the barrier is released.
An example is the barrier_test function in asyntest.py. In the code
fragment from that program:
import asyn
def callback(text):
print(text)
barrier = asyn.Barrier(3, callback, ('Synch',))
async def report():
for i in range(5):
print('{} '.format(i), end='')
await barriermultiple instances of report print their result and pause until the other
instances are also complete and waiting on barrier. At that point the
callback runs. On its completion the coros resume.
A semaphore limits the number of coros which can access a resource. It can be used to limit the number of instances of a particular coro which can run concurrently. It performs this using an access counter which is initialised by the constructor and decremented each time a coro acquires the semaphore. Full details.
The easiest way to use it is with a context manager:
import asyn
sema = asyn.Semaphore(3)
async def foo(sema):
async with sema:
# Limited access hereAn example is the semaphore_test function in asyntest.py.
This works identically to the Semaphore class except that if the release
method causes the access counter to exceed its initial value, a ValueError
is raised. Full details.
The Queue class is officially supported and the sample program aqtest.py
demonstrates its use. A queue is instantiated as follows:
from uasyncio.queues import Queue
q = Queue()A typical producer coro might work as follows:
async def producer(q):
while True:
result = await slow_process() # somehow get some data
await q.put(result) # may pause if a size limited queue fillsand the consumer works along these lines:
async def consumer(q):
while True:
result = await(q.get()) # Will pause if q is empty
print('Result was {}'.format(result))The Queue class provides significant additional functionality in that the
size of queues may be limited and the status may be interrogated. The behaviour
on empty status and (where size is limited) the behaviour on full status may be
controlled. Documentation of this is in the code.
uasyncio now provides a cancel(coro) function. This works by throwing an
exception to the coro in a special way: cancellation is deferred until the coro
is next scheduled. This mechanism works with nested coros. However there is a
limitation. If a coro issues await uasyncio.sleep(secs) or
uasyncio.sleep_ms(ms) scheduling will not occur until the time has elapsed.
This introduces latency into cancellation which matters in some use-cases.
Other potential sources of latency take the form of slow code. uasyncio has
no mechanism for verifying when cancellation has actually occurred. The asyn
library provides verification via the following classes:
CancellableThis allows one or more tasks to be assigned to a group. A coro can cancel all tasks in the group, pausing until this has been achieved. Documentation may be found here.NamedTaskThis enables a coro to be associated with a user-defined name. The running status of named coros may be checked. For advanced usage more complex groupings of tasks can be created. Documentation may be found here.
A typical use-case is as follows:
async def comms(): # Perform some communications task
while True:
await initialise_link()
try:
await do_communications() # Launches Cancellable tasks
except CommsError:
await Cancellable.cancel_all()
# All sub-tasks are now known to be stopped. They can be re-started
# with known initial state on next pass.Examples of the usage of these classes may be found in asyn_demos.py. For an
illustration of the mechanism a cancellable task is defined as below:
@asyn.cancellable
async def print_nums(num):
while True:
print(num)
num += 1
await asyn.sleep(1)It is launched and cancelled with:
async def foo():
loop = asyncio.get_event_loop()
loop.create_task(asyn.Cancellable(print_nums, 42)())
await asyn.sleep(7.5)
await asyn.Cancellable.cancel_all()
print('Done')Note It is bad practice to issue the close or throw methods of a
de-scheduled coro. This subverts the scheduler by causing the coro to execute
code even though descheduled. This is likely to have unwanted consequences.
The asyn.py library provides 'micro' implementations of CPython capabilities, namely the Condition class and the gather method.
The Condition class enables a coro to notify other coros which are waiting on
a locked resource. Once notified they will access the resource and release the
lock in turn. The notifying coro can limit the number of coros to be notified.
The CPython gather method enables a list of coros to be launched. When the
last has completed a list of results is returned. This 'micro' implementation
uses different syntax. Timeouts may be applied to any of the coros.
In the context of device drivers the aim is to ensure nonblocking operation. The design should ensure that other coros get scheduled in periods while the driver is waiting for the hardware. For example a task awaiting data arriving on a UART or a user pressing a button should allow other coros to be scheduled until the event occurs..
A coro can pause execution by waiting on an awaitable object. Under CPython
a custom class is made awaitable by implementing an __await__ special
method. This returns a generator. An awaitable class is used as follows:
import uasyncio as asyncio
class Foo():
def __await__(self):
for n in range(5):
print('__await__ called')
yield from asyncio.sleep(1) # Other coros get scheduled here
return 42
__iter__ = __await__ # See note below
async def bar():
foo = Foo() # Foo is an awaitable class
print('waiting for foo')
res = await foo # Retrieve result
print('done', res)
loop = asyncio.get_event_loop()
loop.run_until_complete(bar())Currently MicroPython doesn't support __await__
issue #2678 and
__iter__ must be used. The line __iter__ = __await__ enables portability
between CPython and MicroPython. Example code may be found in the Event,
Barrier, Cancellable and Condition classes in asyn.py.
Awaitable objects can be used in synchronous or asynchronous CM's by providing the necessary special methods. The syntax is:
with await awaitable as a: # The 'as' clause is optional
# code omitted
async with awaitable as a: # Asynchronous CM (see below)
# do somethingTo achieve this the __await__ generator should return self. This is passed
to any variable in an as clause and also enables the special methods to work.
See asyn.Condition and asyntest.condition_test, where the Condition class
is awaitable and may be used in a synchronous CM.
The Python language requires that __await__ is a generator function. In
MicroPython generators and coroutines are identical, so the solution is to use
yield from coro(args).
This tutorial aims to offer code portable to CPython 3.5 or above. In CPython
coroutines and generators are distinct. CPython coros have an __await__
special method which retrieves a generator. This is portable:
up = False # Running under MicroPython?
try:
import uasyncio as asyncio
up = True # Or can use sys.implementation.name
except ImportError:
import asyncio
async def times_two(n): # Coro to await
await asyncio.sleep(1)
return 2 * n
class Foo():
def __await__(self):
res = 1
for n in range(5):
print('__await__ called')
if up: # MicroPython
res = yield from times_two(res)
else: # CPython
res = yield from times_two(res).__await__()
return res
__iter__ = __await__
async def bar():
foo = Foo() # foo is awaitable
print('waiting for foo')
res = await foo # Retrieve value
print('done', res)
loop = asyncio.get_event_loop()
loop.run_until_complete(bar())Note that, in __await__, yield from asyncio.sleep(1) is allowed by CPython.
I haven't yet established how this is achieved.
These provide a means of returning a finite or infinite sequence of values
and could be used as a means of retrieving successive data items as they arrive
from a read-only device. An asynchronous iterable calls asynchronous code in
its next method. The class must conform to the following requirements:
- It has an
__aiter__method defined withasync defand returning the asynchronous iterator. - It has an
__anext__method which is a coro - i.e. defined withasync defand containing at least oneawaitstatement. To stop iteration it must raise aStopAsyncIterationexception.
Successive values are retrieved with async for as below:
class AsyncIterable:
def __init__(self):
self.data = (1, 2, 3, 4, 5)
self.index = 0
async def __aiter__(self):
return self
async def __anext__(self):
data = await self.fetch_data()
if data:
return data
else:
raise StopAsyncIteration
async def fetch_data(self):
await asyncio.sleep(0.1) # Other coros get to run
if self.index >= len(self.data):
return None
x = self.data[self.index]
self.index += 1
return x
async def run():
ai = AsyncIterable()
async for x in ai:
print(x)Classes can be designed to support asynchronous context managers. These are CM's
having enter and exit procedures which are coros. An example is the Lock
class described above. This has an __aenter__ coro which is logically
required to run asynchronously. To support the asynchronous CM protocol its
__aexit__ method also must be a coro, achieved by including
await asyncio.sleep(0). Such classes are accessed from within a coro with
the following syntax:
async def bar(lock):
async with lock:
print('bar acquired lock')As with normal context managers an exit method is guaranteed to be called when
the context manager terminates, whether normally or via an exception. To
achieve this the special methods __aenter__ and __aexit__ must be
defined, both being coros waiting on a coro or awaitable object. This example
comes from the Lock class:
async def __aenter__(self):
await self.acquire() # a coro defined with async def
return self
async def __aexit__(self, *args):
self.release() # A conventional method
await asyncio.sleep_ms(0)If the async with has an as variable clause the variable receives the
value returned by __aenter__.
There was a bug in the implementation whereby if an explicit return was issued
within an async with block, the __aexit__ method was not called. This was
fixed as of 27th June 2018 PR 3890.
Timeouts are implemented by means of uasyncio.wait_for(). This takes as
arguments a coroutine and a timeout in seconds. If the timeout expires a
TimeoutError will be thrown to the coro in such a way that the next time the
coro is scheduled for execution the exception will be raised. The coro should
trap this and quit.
import uasyncio as asyncio
async def forever():
print('Starting')
try:
while True:
await asyncio.sleep_ms(300)
print('Got here')
except asyncio.TimeoutError:
print('Got timeout')
async def foo():
await asyncio.wait_for(forever(), 5)
await asyncio.sleep(2)
loop = asyncio.get_event_loop()
loop.run_until_complete(foo())Note that if the coro issues await asyncio.sleep(t) where t is a long delay
it will not be rescheduled until t has elapsed. If the timeout has elapsed
before the sleep is complete the TimeoutError will occur when the coro is
scheduled - i.e. when t has elapsed. In real time and from the point of view
of the calling coro, its response to the TimeoutError will be delayed.
If this matters to the application, create a long delay by awaiting a short one
in a loop. The coro asyn.sleep supports this.
Where an exception occurs in a coro, it should be trapped either in that coro or in a coro which is awaiting its completion. This ensures that the exception is not propagated to the scheduler. If this occurred it would stop running, passing the exception to the code which started the scheduler.
Using throw or close to throw an exception to a coro is unwise. It subverts
uasyncio by forcing the coro to run, and possibly terminate, when it is still
queued for execution.
There is a "gotcha" illustrated by this code sample. If allowed to run to completion it works as expected.
import uasyncio as asyncio
async def foo():
await asyncio.sleep(3)
print('About to throw exception.')
1/0
async def bar():
try:
await foo()
except ZeroDivisionError:
print('foo was interrupted by zero division') # Happens
raise # Force shutdown to run by propagating to loop.
except KeyboardInterrupt:
print('foo was interrupted by ctrl-c') # NEVER HAPPENS
raise
async def shutdown():
print('Shutdown is running.') # Happens in both cases
await asyncio.sleep(1)
print('done')
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(bar())
except ZeroDivisionError:
loop.run_until_complete(shutdown())
except KeyboardInterrupt:
print('Keyboard interrupt at loop level.')
loop.run_until_complete(shutdown())However issuing a keyboard interrupt causes the exception to go to the event
loop. This is because uasyncio.sleep causes execution to be transferred to
the event loop. Consequently applications requiring cleanup code in response to
a keyboard interrupt should trap the exception at the event loop level.
At heart all interfaces between uasyncio and external asynchronous events
rely on polling. Hardware requiring a fast response may use an interrupt. But
the interface between the interrupt service routine (ISR) and a user coro will
be polled. For example the ISR might trigger an Event or set a global flag,
while a coroutine awaiting the outcome polls the object each time it is
scheduled.
Polling may be effected in two ways, explicitly or implicitly. The latter is
performed by using the stream I/O mechanism which is a system designed for
stream devices such as UARTs and sockets. At its simplest explicit polling may
consist of code like this:
async def poll_my_device():
global my_flag # Set by device ISR
while True:
if my_flag:
my_flag = False
# service the device
await asyncio.sleep(0)In place of a global, an instance variable, an Event object or an instance of
an awaitable class might be used. Explicit polling is discussed
further below.
Implicit polling consists of designing the driver to behave like a stream I/O
device such as a socket or UART, using stream I/O. This polls devices using
Python's select.poll system: because the polling is done in C it is faster
and more efficient than explicit polling. The use of stream I/O is discussed
here.
Owing to its efficiency implicit polling benefits most fast I/O device drivers: streaming drivers can be written for many devices not normally considered as streaming devices section 5.4.
Both explicit and implicit polling are currently based on round-robin scheduling. Assume I/O is operating concurrently with N user coros each of which yields with a zero delay. When I/O has been serviced it will next be polled once all user coros have been scheduled. The implied latency needs to be considered in the design. I/O channels may require buffering, with an ISR servicing the hardware in real time from buffers and coroutines filling or emptying the buffers in slower time.
The possibility of overrun also needs to be considered: this is the case where something being polled by a coroutine occurs more than once before the coro is actually scheduled.
Another timing issue is the accuracy of delays. If a coro issues
await asyncio.sleep_ms(t)
# next linethe scheduler guarantees that execution will pause for at least tms. The
actual delay may be greater depending on the system state when t expires.
If, at that time, all other coros are waiting on nonzero delays, the next line
will immediately be scheduled. But if other coros are pending execution (either
because they issued a zero delay or because their time has also elapsed) they
may be scheduled first. This introduces a timing uncertainty into the sleep()
and sleep_ms() functions. The worst-case value for this overrun may be
calculated by summing, for every other coro, the worst-case execution time
between yielding to the scheduler.
The fast_io version of uasyncio in this repo provides a way
to ensure that stream I/O is polled on every iteration of the scheduler. It is
hoped that official uasyncio will adopt code to this effect in due course.
This is a simple approach, but is most appropriate to hardware which may be polled at a relatively low rate. This is primarily because polling with a short (or zero) polling interval may cause the coro to consume more processor time than is desirable.
The example apoll.py demonstrates this approach by polling the Pyboard
accelerometer at 100ms intervals. It performs some simple filtering to ignore
noisy samples and prints a message every two seconds if the board is not moved.
Further examples may be found in aswitch.py which provides drivers for
switch and pushbutton devices.
An example of a driver for a device capable of reading and writing is shown
below. For ease of testing Pyboard UART 4 emulates the notional device. The
driver implements a RecordOrientedUart class, where data is supplied in
variable length records consisting of bytes instances. The object appends a
delimiter before sending and buffers incoming data until the delimiter is
received. This is a demo and is an inefficient way to use a UART compared to
stream I/O.
For the purpose of demonstrating asynchronous transmission we assume the
device being emulated has a means of checking that transmission is complete
and that the application requires that we wait on this. Neither assumption is
true in this example but the code fakes it with await asyncio.sleep(0.1).
Link pins X1 and X2 to run.
import uasyncio as asyncio
from pyb import UART
class RecordOrientedUart():
DELIMITER = b'\0'
def __init__(self):
self.uart = UART(4, 9600)
self.data = b''
def __iter__(self): # Not __await__ issue #2678
data = b''
while not data.endswith(self.DELIMITER):
yield from asyncio.sleep(0) # Necessary because:
while not self.uart.any():
yield from asyncio.sleep(0) # timing may mean this is never called
data = b''.join((data, self.uart.read(self.uart.any())))
self.data = data
async def send_record(self, data):
data = b''.join((data, self.DELIMITER))
self.uart.write(data)
await self._send_complete()
# In a real device driver we would poll the hardware
# for completion in a loop with await asyncio.sleep(0)
async def _send_complete(self):
await asyncio.sleep(0.1)
def read_record(self): # Synchronous: await the object before calling
return self.data[0:-1] # Discard delimiter
async def run():
foo = RecordOrientedUart()
rx_data = b''
await foo.send_record(b'A line of text.')
for _ in range(20):
await foo # Other coros are scheduled while we wait
rx_data = foo.read_record()
print('Got: {}'.format(rx_data))
await foo.send_record(rx_data)
rx_data = b''
loop = asyncio.get_event_loop()
loop.run_until_complete(run())This can be illustrated using a Pyboard UART. The following code sample demonstrates concurrent I/O on one UART. To run, link Pyboard pins X1 and X2 (UART Txd and Rxd).
import uasyncio as asyncio
from pyb import UART
uart = UART(4, 9600)
async def sender():
swriter = asyncio.StreamWriter(uart, {})
while True:
await swriter.awrite('Hello uart\n')
await asyncio.sleep(2)
async def receiver():
sreader = asyncio.StreamReader(uart)
while True:
res = await sreader.readline()
print('Received', res)
loop = asyncio.get_event_loop()
loop.create_task(sender())
loop.create_task(receiver())
loop.run_forever()The supporting code may be found in __init__.py in the uasyncio library.
The mechanism works because the device driver (written in C) implements the
following methods: ioctl, read, readline and write. See
Writing streaming device drivers
for details on how such drivers may be written in Python.
A UART can receive data at any time. The stream I/O mechanism checks for pending incoming characters whenever the scheduler has control. When a coro is running an interrupt service routine buffers incoming characters; these will be removed when the coro yields to the scheduler. Consequently UART applications should be designed such that coros minimise the time between yielding to the scheduler to avoid buffer overflows and data loss. This can be ameliorated by using a larger UART read buffer or a lower baudrate. Alternatively hardware flow control will provide a solution if the data source supports it.
The program auart_hd.py illustrates a method of communicating with a half duplex device such as one responding to the modem 'AT' command set. Half duplex means that the device never sends unsolicited data: its transmissions are always in response to a command from the master.
The device is emulated, enabling the test to be run on a Pyboard with two wire links.
The (highly simplified) emulated device responds to any command by sending four lines of data with a pause between each, to simulate slow processing.
The master sends a command, but does not know in advance how many lines of data will be returned. It starts a retriggerable timer, which is retriggered each time a line is received. When the timer times out it is assumed that the device has completed transmission, and a list of received lines is returned.
The case of device failure is also demonstrated. This is done by omitting the transmission before awaiting a response. After the timeout an empty list is returned. See the code comments for more details.
The stream I/O mechanism is provided to support I/O to stream devices. Its
typical use is to support streaming I/O devices such as UARTs and sockets. The
mechanism may be employed by drivers of any device which needs to be polled:
the polling is delegated to the scheduler which uses select to schedule the
handlers for any devices which are ready. This is more efficient than running
multiple coros each polling a device, partly because select is written in C
but also because the coroutine performing the polling is descheduled until the
poll object returns a ready status.
A device driver capable of employing the stream I/O mechanism may support
StreamReader, StreamWriter instances or both. A readable device must
provide at least one of the following methods. Note that these are synchronous
methods. The ioctl method (see below) ensures that they are only called if
data is available. The methods should return as fast as possible with as much
data as is available.
readline() Return as many characters as are available up to and including any
newline character. Required if you intend to use StreamReader.readline()
read(n) Return as many characters as are available but no more than n.
Required to use StreamReader.read() or StreamReader.readexactly()
A writeable driver must provide this synchronous method:
write Args buf, off, sz. Arguments:
buf is the buffer to write.
off is the offset into the buffer of the first character to write.
sz is the requested number of characters to write.
It should return immediately. The return value is the number of characters
actually written (may well be 1 if the device is slow). The ioctl method
ensures that this is only called if the device is ready to accept data.
All devices must provide an ioctl method which polls the hardware to
determine its ready status. A typical example for a read/write driver is:
import io
MP_STREAM_POLL_RD = const(1)
MP_STREAM_POLL_WR = const(4)
MP_STREAM_POLL = const(3)
MP_STREAM_ERROR = const(-1)
class MyIO(io.IOBase):
# Methods omitted
def ioctl(self, req, arg): # see ports/stm32/uart.c
ret = MP_STREAM_ERROR
if req == MP_STREAM_POLL:
ret = 0
if arg & MP_STREAM_POLL_RD:
if hardware_has_at_least_one_char_to_read:
ret |= MP_STREAM_POLL_RD
if arg & MP_STREAM_POLL_WR:
if hardware_can_accept_at_least_one_write_character:
ret |= MP_STREAM_POLL_WR
return retThe following is a complete awaitable delay class:
import uasyncio as asyncio
import utime
import io
MP_STREAM_POLL_RD = const(1)
MP_STREAM_POLL = const(3)
MP_STREAM_ERROR = const(-1)
class MillisecTimer(io.IOBase):
def __init__(self):
self.end = 0
self.sreader = asyncio.StreamReader(self)
def __iter__(self):
await self.sreader.readline()
def __call__(self, ms):
self.end = utime.ticks_add(utime.ticks_ms(), ms)
return self
def readline(self):
return b'\n'
def ioctl(self, req, arg):
ret = MP_STREAM_ERROR
if req == MP_STREAM_POLL:
ret = 0
if arg & MP_STREAM_POLL_RD:
if utime.ticks_diff(utime.ticks_ms(), self.end) >= 0:
ret |= MP_STREAM_POLL_RD
return retwhich may be used as follows:
async def timer_test(n):
timer = ms_timer.MillisecTimer()
await timer(30) # Pause 30msWith official uasyncio this confers no benefit over await asyncio.sleep_ms().
Using fast_io it offers much more precise delays under the
common usage pattern where coros await a zero delay.
It is possible to use I/O scheduling to associate an event with a callback.
This is more efficient than a polling loop because the coro doing the polling
is descheduled until ioctl returns a ready status. The following runs a
callback when a pin changes state.
import uasyncio as asyncio
import io
MP_STREAM_POLL_RD = const(1)
MP_STREAM_POLL = const(3)
MP_STREAM_ERROR = const(-1)
class PinCall(io.IOBase):
def __init__(self, pin, *, cb_rise=None, cbr_args=(), cb_fall=None, cbf_args=()):
self.pin = pin
self.cb_rise = cb_rise
self.cbr_args = cbr_args
self.cb_fall = cb_fall
self.cbf_args = cbf_args
self.pinval = pin.value()
self.sreader = asyncio.StreamReader(self)
loop = asyncio.get_event_loop()
loop.create_task(self.run())
async def run(self):
while True:
await self.sreader.read(1)
def read(self, _):
v = self.pinval
if v and self.cb_rise is not None:
self.cb_rise(*self.cbr_args)
return b'\n'
if not v and self.cb_fall is not None:
self.cb_fall(*self.cbf_args)
return b'\n'
def ioctl(self, req, arg):
ret = MP_STREAM_ERROR
if req == MP_STREAM_POLL:
ret = 0
if arg & MP_STREAM_POLL_RD:
v = self.pin.value()
if v != self.pinval:
self.pinval = v
ret = MP_STREAM_POLL_RD
return retOnce again with official uasyncio latency can be high. Depending on
application design the fast_io version can greatly reduce
this.
The demo program iorw.py illustrates a complete example. Note that, at the
time of writing there is a bug in uasyncio which prevents this from working.
See this GitHub thread.
There are two solutions. A workround is to write two separate drivers, one
read-only and the other write-only. Alternatively the
fast_io version addresses this.
In the official uasyncio I/O is scheduled quite infrequently. See
see this GitHub RFC.
The fast_io version addresses this issue.
See aremote.py documented here.
The demo provides a complete device driver example: a receiver/decoder for an
infra red remote controller. The following notes are salient points regarding
its asyncio usage.
A pin interrupt records the time of a state change (in μs) and sets an event, passing the time when the first state change occurred. A coro waits on the event, yields for the duration of a data burst, then decodes the stored data before calling a user-specified callback.
Passing the time to the Event instance enables the coro to compensate for
any asyncio latency when setting its delay period.
This chip provides accurate measurements of temperature and humidity. The
driver is documented here. It has a continuously running
task which updates temperature and humidity bound variables which may be
accessed "instantly".
The chip takes on the order of 120ms to acquire both data items. The driver
works asynchronously by triggering the acquisition and using
await asyncio.sleep(t) prior to reading the data. This allows other coros to
run while acquisition is in progress.
Hanging usually occurs because a task has blocked without yielding: this will hang the entire system. When developing it is useful to have a coro which periodically toggles an onboard LED. This provides confirmation that the scheduler is running.
When running programs using uasyncio at the REPL, issue a soft reset
(ctrl-D) between runs. This is because uasyncio retains state between runs
which can lead to confusing behaviour.
You may want to consider running a coro which issues:
gc.collect()
gc.threshold(gc.mem_free() // 4 + gc.mem_alloc())This assumes import gc has been issued. The purpose of this is discussed
here
in the section on the heap.
It's advisable to test that a device driver yields control when you intend it to. This can be done by running one or more instances of a dummy coro which runs a loop printing a message, and checking that it runs in the periods when the driver is blocking:
async def rr(n):
while True:
print('Roundrobin ', n)
await asyncio.sleep(0)As an example of the type of hazard which can occur, in the RecordOrientedUart
example above the __await__ method was originally written as:
def __await__(self):
data = b''
while not data.endswith(self.DELIMITER):
while not self.uart.any():
yield from asyncio.sleep(0)
data = b''.join((data, self.uart.read(self.uart.any())))
self.data = dataIn testing this hogged execution until an entire record was received. This was
because uart.any() always returned a nonzero quantity. By the time it was
called, characters had been received. The solution was to yield execution in
the outer loop:
def __await__(self):
data = b''
while not data.endswith(self.DELIMITER):
yield from asyncio.sleep(0) # Necessary because:
while not self.uart.any():
yield from asyncio.sleep(0) # timing may mean this is never called
data = b''.join((data, self.uart.read(self.uart.any())))
self.data = dataIt is perhaps worth noting that this error would not have been apparent had data been sent to the UART at a slow rate rather than via a loopback test. Welcome to the joys of realtime programming.
If a function or method is defined with async def and subsequently called as
if it were a regular (synchronous) callable, MicroPython does not issue an
error message. This is by design.
It typically leads to a program silently failing to run correctly:
async def foo():
# code
loop.create_task(foo) # Case 1: foo will never run
foo() # Case 2: Likewise.I have a PR which proposes a fix for case 1. The fast_io version implements this.
The script check_async_code.py attempts to locate
instances of questionable use of coros. It is intended to be run on a PC and
uses Python3. It takes a single argument, a path to a MicroPython sourcefile
(or --help). It is designed for use on scripts written according to the
guidelines in this tutorial, with coros declared using async def.
Note it is somewhat crude and intended to be used on a syntactically correct
file which is silently failing to run. Use a tool such as pylint for general
syntax checking (pylint currently misses this error).
The script produces false positives. This is by design: coros are first class
objects; you can pass them to functions and can store them in data structures.
Depending on the program logic you may intend to store the function or the
outcome of its execution. The script can't deduce the intent. It aims to ignore
cases which appear correct while identifying other instances for review.
Assume foo is a coro declared with async def:
loop.run_until_complete(foo()) # No warning
bar(foo) # These lines will warn but may or may not be correct
bar(foo())
z = (foo,)
z = (foo(),)
foo() # Will warn: is surely wrong.I find it useful as-is but improvements are always welcome.
The use of nonblocking sockets requires some attention to detail. If a nonblocking read is performed, because of server latency, there is no guarantee that all (or any) of the requested data is returned. Likewise writes may not proceed to completion.
Hence asynchronous read and write methods need to iteratively perform the nonblocking operation until the required data has been read or written. In practice a timeout is likely to be required to cope with server outages.
A further complication is that, at the time of writing, the ESP32 port has issues which require rather unpleasant hacks for error-free operation.
The file sock_nonblock.py illustrates the sort of techniques required. It is not a working demo, and solutions are likely to be application dependent.
An alternative approach is to use blocking sockets with StreamReader and
StreamWriter instances to control polling.
A subtle bug can arise if you need to instantiate the event loop with non
default values. Instantiation should be performed before running any other
asyncio code. This is because the code may acquire the event loop. In
doing so it initialises it to the default values:
import uasyncio as asyncio
import some_module
bar = some_module.Bar() # Constructor calls get_event_loop()
# and renders these args inoperative
loop = asyncio.get_event_loop(runq_len=40, waitq_len=40)Given that importing a module can run code the only safe way is to instantiate
the event loop immediately after importing uasyncio.
import uasyncio as asyncio
loop = asyncio.get_event_loop(runq_len=40, waitq_len=40)
import some_module
bar = some_module.Bar() # The get_event_loop() call is now safeRef this issue.
These notes are intended for those new to asynchronous code. They start by
outlining the problems which schedulers seek to solve, and give an overview of
the uasyncio approach to a solution.
Section 7.5
discusses the relative merits of uasyncio and the _thread module and why
you may prefer use cooperative (uasyncio) over pre-emptive (_thread)
scheduling.
A typical firmware application runs continuously and is required to respond to external events. These might include a voltage change on an ADC, the arrival of a hard interrupt, a character arriving on a UART, or data being available on a socket. These events occur asynchronously and the code must be able to respond regardless of the order in which they occur. Further the application may be required to perform time-dependent tasks such as flashing LED's.
The obvious way to do this is with an event loop. The following is not practical code but serves to illustrate the general form of an event loop.
def event_loop():
led_1_time = 0
led_2_time = 0
switch_state = switch.state() # Current state of a switch
while True:
time_now = utime.time()
if time_now >= led_1_time: # Flash LED #1
led1.toggle()
led_1_time = time_now + led_1_period
if time_now >= led_2_time: # Flash LED #2
led2.toggle()
led_2_time = time_now + led_2_period
# Handle LEDs 3 upwards
if switch.value() != switch_state:
switch_state = switch.value()
# do something
if uart.any():
# handle UART inputThis works for simple examples but event loops rapidly become unwieldy as the number of events increases. They also violate the principles of object oriented programming by lumping much of the program logic in one place rather than associating code with the object being controlled. We want to design a class for an LED capable of flashing which could be put in a module and imported. An OOP approach to flashing an LED might look like this:
import pyb
class LED_flashable():
def __init__(self, led_no):
self.led = pyb.LED(led_no)
def flash(self, period):
while True:
self.led.toggle()
# somehow wait for period but allow other
# things to happen at the same timeA cooperative scheduler such as uasyncio enables classes such as this to be
created.
Assume you need to read a number of bytes from a socket. If you call
socket.read(n) with a default blocking socket it will "block" (i.e. fail to
return) until n bytes have been received. During this period the application
will be unresponsive to other events.
With uasyncio and a non-blocking socket you can write an asynchronous read
method. The task requiring the data will (necessarily) block until it is
received but during that period other tasks will be scheduled enabling the
application to remain responsive.
The following class provides for an LED which can be turned on and off, and
which can also be made to flash at an arbitrary rate. A LED_async instance
has a run method which can be considered to run continuously. The LED's
behaviour can be controlled by methods on(), off() and flash(secs).
import pyb
import uasyncio as asyncio
class LED_async():
def __init__(self, led_no):
self.led = pyb.LED(led_no)
self.rate = 0
loop = asyncio.get_event_loop()
loop.create_task(self.run())
async def run(self):
while True:
if self.rate <= 0:
await asyncio.sleep_ms(200)
else:
self.led.toggle()
await asyncio.sleep_ms(int(500 / self.rate))
def flash(self, rate):
self.rate = rate
def on(self):
self.led.on()
self.rate = 0
def off(self):
self.led.off()
self.rate = 0Note that on(), off() and flash() are conventional synchronous methods.
They change the behaviour of the LED but return immediately. The flashing
occurs "in the background". This is explained in detail in the next section.
The class conforms with the OOP principle of keeping the logic associated with
the device within the class. Further, the way uasyncio works ensures that
while the LED is flashing the application can respond to other events. The
example below flashes the four Pyboard LED's at different rates while also
responding to the USR button which terminates the program.
import pyb
import uasyncio as asyncio
from led_async import LED_async # Class as listed above
async def killer():
sw = pyb.Switch()
while not sw.value():
await asyncio.sleep_ms(100)
leds = [LED_async(n) for n in range(1, 4)]
for n, led in enumerate(leds):
led.flash(0.7 + n/4)
loop = asyncio.get_event_loop()
loop.run_until_complete(killer())In contrast to the event loop example the logic associated with the switch is in a function separate from the LED functionality. Note the code used to start the scheduler:
loop = asyncio.get_event_loop()
loop.run_until_complete(killer()) # Execution passes to coroutines.
# It only continues here once killer() terminates, when the
# scheduler has stopped.Python 3.5 and MicroPython support the notion of an asynchronous function,
also known as a coroutine (coro) or task. A coro must include at least one
await statement.
async def hello():
for _ in range(10):
print('Hello world.')
await asyncio.sleep(1)This function prints the message ten times at one second intervals. While the function is paused pending the time delay asyncio will schedule other tasks, providing an illusion of concurrency.
When a coro issues await asyncio.sleep_ms() or await asyncio.sleep() the
current task pauses: it is placed on a queue which is ordered on time due, and
execution passes to the task at the top of the queue. The queue is designed so
that even if the specified sleep is zero other due tasks will run before the
current one is resumed. This is "fair round-robin" scheduling. It is common
practice to issue await asyncio.sleep(0) in loops to ensure a task doesn't
hog execution. The following shows a busy-wait loop which waits for another
task to set the global flag. Alas it monopolises the CPU preventing other
coros from running:
async def bad_code():
global flag
while not flag:
pass
flag = False
# code omittedThe problem here is that while the flag is False the loop never yields to
the scheduler so no other task will get to run. The correct approach is:
async def good_code():
global flag
while not flag:
await asyncio.sleep(0)
flag = False
# code omittedFor the same reason it's bad practice to issue delays like utime.sleep(1)
because that will lock out other tasks for 1s; use await asyncio.sleep(1).
Note that the delays implied by uasyncio methods sleep and sleep_ms can
overrun the specified time. This is because while the delay is in progress
other tasks will run. When the delay period completes, execution will not
resume until the running task issues await or terminates. A well-behaved coro
will always issue await at regular intervals. Where a precise delay is
required, especially one below a few ms, it may be necessary to use
utime.sleep_us(us).
The initial reaction of beginners to the idea of cooperative multi-tasking is often one of disappointment. Surely pre-emptive is better? Why should I have to explicitly yield control when the Python virtual machine can do it for me?
When it comes to embedded systems the cooperative model has two advantages. Firstly, it is lightweight. It is possible to have large numbers of coroutines because unlike descheduled threads, paused coroutines contain little state. Secondly it avoids some of the subtle problems associated with pre-emptive scheduling. In practice cooperative multi-tasking is widely used, notably in user interface applications.
To make a case for the defence a pre-emptive model has one advantage: if someone writes
for x in range(1000000):
# do something time consumingit won't lock out other threads. Under cooperative schedulers the loop must
explicitly yield control every so many iterations e.g. by putting the code in
a coro and periodically issuing await asyncio.sleep(0).
Alas this benefit of pre-emption pales into insignificance compared to the drawbacks. Some of these are covered in the documentation on writing interrupt handlers. In a pre-emptive model every thread can interrupt every other thread, changing data which might be used in other threads. It is generally much easier to find and fix a lockup resulting from a coro which fails to yield than locating the sometimes deeply subtle and rarely occurring bugs which can occur in pre-emptive code.
To put this in simple terms, if you write a MicroPython coroutine, you can be
sure that variables won't suddenly be changed by another coro: your coro has
complete control until it issues await asyncio.sleep(0).
Bear in mind that interrupt handlers are pre-emptive. This applies to both hard and soft interrupts, either of which can occur at any point in your code.
An eloquent discussion of the evils of threading may be found in threads are bad.
In non-trivial applications coroutines need to communicate. Conventional Python techniques can be employed. These include the use of global variables or declaring coros as object methods: these can then share instance variables. Alternatively a mutable object may be passed as a coro argument.
Pre-emptive systems mandate specialist classes to achieve "thread safe" communications; in a cooperative system these are seldom required.
Some hardware devices such as the Pyboard accelerometer don't support
interrupts, and therefore must be polled (i.e. checked periodically). Polling
can also be used in conjunction with interrupt handlers: the interrupt handler
services the hardware and sets a flag. A coro polls the flag: if it's set it
handles the data and clears the flag. A better approach is to use an Event.