What we will build
We are going to make a metronome using Python’s
asyncio module. Our metronome is going tick at a steady rate, executing a collection of subroutines at each tick. I’ve used this module to control a light & music show – but you can use this code for running any operation at regular intervals. The code is relatively straight-forward – I just think it’s a neat introdoction to the power of async IO!
Intro to async IO
asyncio module is an everything-and-the-kitchen-sink package of parallel & concurrent programming tools. We’re just going to look at a very simple use case for this library: running an infinite loop with a sleep in each iteration.
In synchronous single-threaded programs, calling
sleep() does just that – it puts the thread to sleep, and the thread does no work until it wakes up. With Python’s
asyncio there is a kind of ‘sleep’ where the thread does not stop working, but instead gets a free pass for a time to do work elsewhere. With calling
asyncio.sleep(), the thread refrains from working at the current routine, and makes itself available to do work elswhere. The single thread does not go to sleep – it rushes off to do as much work as it can before being yanked back ‘awake’ to the place it briefly adjourned.
asyncio.sleep(), if we wanted our program to keep working during a sleep, we would need more than 1 thread! As anyone who has tried it knows: multithreaded programming is neither easy nor simple. But, as we’ll see, getting work out your program while putting a thread ‘to sleep’ is a cinch with
The upshot: we’re going to write code with infinite loops, 1 thread, and high-performance.
Metronome spec time
We are going to make a metronome. A metronome is used typically in music to synchronize a performance to a regular tempo. Traditionally, a tempo is defined in terms of beats-per-minute. It terms of code, our metronome is going to beat at regular intervals, and at each beat it will execute a collection of callbacks.
All our metronome needs is (1) the beats-per-minute tempo, and (2) a list of callbacks. We can take a first pass at the code
import asyncio from typing import Callable, Any, List CallBack = Callable[None, Any] CallBacks = List[Callback] class Metronome: def __init__(tempo: int, callbacks: CallBacks): self.tempo = tempo self.callbacks = callbacks async def start(self): while True: # tempo is beats-per-minute. 60bpm is a beat every 1 second sleep_time = 60 / self.tempo asyncio.sleep(sleep_time) for callback in self.callbacks: callback()
This is a good start! It’s a minimum viable metronome, for sure. We instantiate the class with 2 arguments: a tempo and a list of callbacks. The tempo is used for determining the sleep time, and at each beat in the loop, our metronome executes all the callbacks.
How can we improve it? Let’s make 2 changes. The first will improve performance, and the second will improve the regularity of our sleep interval, so that the metronome does not drift much.
1. Async everywhere
I want to use my metronome for sending commands to things in the outside world. Maybe my code will send MIDI messages, or control an array of LEDs. This IO-bound work takes time – time which can block our main thread. This sort of IO-focused processing is what Python’s
asyncio library is made for! If we ensure that our callbacks are all async subroutines, then we can mitigate the impact of IO on our code’s performance. This is just a simple change to our metronome’s API & implementation.
Now with all async callbacks:
import asyncio from typing import Callable, Any, List, Awaitable CallBack = Callable[None, Awaitable[Any]] CallBacks = List[Callback] class Metronome: def __init__(tempo: int, callbacks: CallBacks): self.tempo = tempo self.callbacks = callbacks async def start(self): while True: # tempo is beats-per-minute. 60bpm is a beat every 1 second sleep_time = 60 / self.tempo asyncio.sleep(sleep_time) for callback in self.callbacks: await callback()
It’s a simple change with big consequences. It ensures that our callbacks will not block our thread while waiting on IO to complete.
2. Prevent drift over repeated intervals
Making our callbacks asynchronous does not make them run faster. The IO operations will still take time. This time will create a drift in our metronome intervals, where one beat might take longer than another owing to how long it needed to wait for its callbacks to complete. We need to keep track of the time it takes to run a beat’s worth of callbacks to completion. This is a simple enough feature to implement:
import asyncio import time from typing import Callable, Any, List, Awaitable CallBack = Callable[None, Awaitable[Any]] CallBacks = List[Callback] class Metronome: def __init__(tempo: int, callbacks: CallBacks): self.tempo = tempo self.callbacks = callbacks async def start(self): offset = 0 while True: sleep_time = max(0, (60 / self.bpm - offset)) asyncio.sleep(sleep_time) # start timer t0 = time.time() for callback in self.callbacks: await callback() t1 = time.time() # calculate offset offset = t1 - t0
Now we are timing our callbacks, and reduce our sleep time accordingly. This won’t perfectly eliminate drift, but the minor variations between beats should be well within what a human ear can detect.
Under the hood, our metronome is an infinite loop. We can run this infinite loop in the midst of a single-threaded program because we utilize
asyncio.sleep() in our metronome’s loop, which frees up our thread to do other work while it is wating for the next beat.
.start() method of the metronome returns a coroutine. This coroutine needs to be passed off to an
asyncio event loop before it will run. Here is a simple example usage:
import asyncio from metronome import Metronome # define a callback async def do_beat(): print('BEAT') # collect all the metronome's callbacks cbs = [do_beat] # set tempo to 120 beats-per-minute tempo = 120 # create the metronome metronome = Metronome(tempo=tempo, callbacks=cbs) # produce the coroutine metronome_coro = metronome.start() # asyncio boilerplate to run the coroutine loop = asyncio.get_event_loop() loop.run_until_complete(metronome_coro)