Writing/The Unbearable Lightness of Coding
Chapter 28 · Draft · 77 min read
Async Programming: Select, Callbacks, and Coroutines
Three Banks
One of the most important principles of good design, by Dieter Rams, is that good design is environmentally friendly. For software, I like to think of it in the sense of good design being considerate of its environment: it doesn't use more resources than it absolutely has to. And by this measure, the end of our last chapter was pretty depressing, wasn't it? We talked about machines, virtual machines, containers, processes, threads — and most of the time, they're just there; it's like we've surrounded ourselves with a lot of cars, which spend most of their time parked — entire fleets of expensive machinery, idling and waiting.
The key insight behind asynchronous programming is that these things are mostly waiting for I/O — and while they are, their hardware can be doing something else on behalf of someone else. A web server, for example, can generate a response in a few microseconds, and then spend orders of magnitude more time sending it, and waiting for the next request. While our one client is waiting, we can be doing useful work for other clients, even within the same thread.
So, remember the bank clerks from the end of the last chapter? The single British clerk, with his long, neat queue of people patiently waiting for their turn (none of them would dream of cutting in line), is synchronous code; and twenty British clerks are multithreading, or really, a thread pool — because customer number twenty-one still has to wait, and the clerks keep bumping into each other around the same shared filing cabinet, and that's where we need to use locks.
But the Israeli clerk: you go up to him, and he hands you a form; but before you even start filling it out, he's already addressing the next person in line: "sit aside, you, what do you want?" — and while you're still writing, he's helping others, and when you're done, you get his attention again, and he gets back to you, and so on. In other words, he's zig-zagging: whenever a customer is busy doing something, like filling out a form, he uses that time to focus his attention elsewhere; this is what we call asynchronous programming.
This isn't better or worse — it's just different. If what the customers want is long explanations (that is, CPU-bound work), the clerk can't explain two things at once — and the only thing that helps is real parallelism, with multiple clerks. Also, there's a difference in who's doing the juggling: when we use threads, we delegate it to the operating system ("please run these things in parallel, I don't care how"), but with asynchronous programming, we do the juggling ourselves, so we have to engineer it accordingly.
In fact, asynchronous programming is the oldest of the three, and it's how servers were written when machines only had a single core. It's also the most counterintuitive paradigm of the three, so it's OK if this chapter melts your brain a little. It usually takes people a few iterations to really get it, so if at any point you feel like this is too much, just take a break and go over things again as many times as you need.
Here's the plan, and also the chapter's title: first, we'll cover select, which is the most manual way of doing asynchronous programming that you can imagine. Then, we'll talk about callbacks — a framework built on top of it, which we'll implement ourselves, before we check out the real one, called twisted. Next, we'll cover coroutines, and develop our own framework based on generators; and then we'll discuss asyncio, the standard library version, from yield from to async and await. Finally, we'll go on a deep dive of asyncio itself: we'll learn how to run it, and get to know its futures and tasks and exceptions and cancellations and timeouts and threads and streams and so on.
Select
Let's start with the echo server from the previous chapter, which achieved concurrency by dedicating a thread per connection, only to run into the limits of this approach: thousands of threads, most of them idle, waiting for recv to return, and all of them bringing with them various races and deadlocks, which are notoriously hard to reason about.
So let's turn to select, and its more scalable successors, poll and epoll (on Linux) and kqueue (on BSD and macOS). It's pretty much the most low-level, manual approach you could imagine to a problem as fundamental as network programming.
According to its man page, select lets you monitor multiple file descriptors, waiting until one or more of them become "ready" for some class of I/O operation. In case of a single socket, it's pretty pointless — but if you have many sockets, it lets you block until any one of them is ready, at which point you can handle just this one socket, and only for as long as it's absolutely necessary to service it. If this is the first time you're hearing about this, take a moment, and read carefully — it's important.
What "ready" means is that a readable listener has a connection pending, so calling accept on it will return instantly; a readable connection has some data available (or the client closed it), so calling recv on it will return instantly; and a writable socket has some room in its kernel send buffer, so calling send on it won't block. The rule is always the same — you ask the operating system, "can I act on this socket? Or rather, can I act on any of these sockets?", and only go ahead with the ones on which you can do something immediately, without having to wait.
Let's play with it a little in the interpreter, using a small helper function, ready, which selects a given socket with a timeout of zero, which means "don't wait at all — just check and tell me". It returns a boolean indicating whether the socket is readable, and another one indicating whether it's writable:
>>> import select, socket
>>> def ready(sock):
... readable, writable, errors = select.select([sock], [sock], [sock], 0)
... return sock in readable, sock in writable
>>> listener = socket.socket()
>>> listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
>>> listener.bind(('127.0.0.1', 8000)); listener.listen(5)
>>> ready(listener)
(False, False)
>>> client = socket.create_connection(('127.0.0.1', 8000))
>>> ready(listener)
(True, False)
>>> connection, address = listener.accept()
>>> ready(connection)
(False, True)
>>> client.sendall(b'Hello, ')
>>> ready(connection)
(True, True)
>>> connection.recv(4096)
b'Hello, '
The first thing we do is check whether our fresh listener is readable; and indeed, it's not. But once a client connects, it is — and calling accept returns instantly. The new connection is writable right away (that is, its send buffer is empty), but not readable — until the client sends us something, like "Hello, ", at which point it becomes both, and recv returns the data without having to wait for it.
The writable part is a bit subtle: it doesn't mean that the other side is actually reading anything, just that the kernel buffer has room — so let's fill it up:
>>> connection.setblocking(False)
>>> sent = 0
>>> try:
... while True:
... sent += connection.send(b'x' * 4096)
... except BlockingIOError:
... print(f'the buffers are full after {sent:,} bytes')
the buffers are full after 544,248 bytes
>>> ready(connection)
(False, False)
>>> client.recv(65536)[:10]
b'xxxxxxxxxx'
>>> ready(connection)
(False, True)
Because the socket is in non-blocking mode, send raises a BlockingIOError instead of blocking; and after sending about half a megabyte of data (it depends on the exact size of the buffers, both ours and the receiver's), it becomes unwritable — until the client reads something, and makes some room, at which point it becomes writable again. This is also the reason sendall can still block: if the client doesn't read anything, sendall will just wait until there's room.
Now, let's use it to implement a single-threaded echo server. We start with a single thread, a set of active sockets with just our listener, and a forever loop that selects the readable ones and deals with them:
>>> class EchoServer:
... def __init__(self, host, port):
... self.host = host
... self.port = port
... def start(self):
... listener = socket.socket()
... listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
... listener.bind((self.host, self.port))
... listener.listen(5)
... active = {listener}
... with listener:
... while True:
... ready, _, _ = select.select(active, [], [])
... for sock in ready:
... if sock is listener: # accept won't block
... connection, address = sock.accept()
... active.add(connection)
... continue
... data = sock.recv(4096) # recv won't block
... if data:
... sock.sendall(data) # but this might!
... else:
... sock.close()
... active.remove(sock)
>>> server = EchoServer('127.0.0.1', 8000)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> a = socket.create_connection(('127.0.0.1', 8000))
>>> b = socket.create_connection(('127.0.0.1', 8000))
>>> a.sendall(b'hello'); a.recv(4096)
b'hello'
>>> b.sendall(b'world'); b.recv(4096)
b'world'
>>> threading.active_count()
2
If the ready socket is our listener, that means a new connection arrived, so we accept it and add it to the active set; and if it's a connection, we receive data from it — which, thanks to our selection strategy, shouldn't block — and echo it back (or, if we get nothing, that means the client closed the connection, so we close ours and remove it from the active set). Two clients are served, seemingly concurrently, by a single thread; and if we call threading.active_count(), we get 2, indicating that it's just the REPL's thread and our server's thread. We actually only made the receiving part asynchronous, though: sendall can still block, and a fully asynchronous implementation would also buffer the outgoing data and select for writability, too.
What we're really doing here is coding the operating system's scheduling logic with tweezers. When the operating system schedules threads, it allocates them, monitors them, and wakes them up when the result they're waiting for arrives; here, we're just telling it "let us know when something happens, and we'll do the rest".
A Slightly Harder Problem
Our echo server is actually about as simple as a protocol gets — so let's make it slightly more complicated, just for fun. Before the client gets its echo service, we'll require it to provide a four-byte password — 1234, say, very secure — and here's how it might look:
>>> class SecureEchoServer:
... def __init__(self, host, port):
... self.host = host
... self.port = port
... def start(self):
... listener = socket.socket()
... listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
... listener.bind((self.host, self.port))
... listener.listen(5)
... active = {listener: None}
... with listener:
... while True:
... ready, _, _ = select.select(active, [], [])
... for sock in ready:
... if sock is listener:
... connection, address = sock.accept()
... active[connection] = 'auth'
... elif active[sock] == 'auth':
... data = sock.recv(4)
... if data == b'1234':
... active[sock] = 'echo'
... else:
... sock.close()
... del active[sock]
... elif data := sock.recv(4096):
... sock.sendall(data)
... else:
... sock.close()
... del active[sock]
>>> server = SecureEchoServer('127.0.0.1', 8001)
>>> threading.Thread(target=server.start, daemon=True).start()
Our active set now becomes a dictionary of states; and since every socket now requires explicit state, I've decided to map each connection to 'auth' or 'echo', while the listener gets None. In the auth state, we receive four bytes; and if they are indeed the password, we promote the connection to the echo state, and otherwise we close it. In this state, we behave as before (using the walrus, just to make it a bit shorter: receive something, and if you get something, send it back).
Already, it's getting ugly. Imagine how a real conversation would look: login, commands and replies, maybe some file transfers — all this baked monolithically into this loop, into this single state machine. Anything more complicated than this, and we'd end up with quite a mess.
And it's not only ugly, it's wrong — let's see how:
>>> a = socket.create_connection(('127.0.0.1', 8001))
>>> a.sendall(b'1234'); a.sendall(b'hello'); a.recv(4096)
b'hello'
>>> b = socket.create_connection(('127.0.0.1', 8001))
>>> b.sendall(b'12')
>>> b.sendall(b'34'); b.sendall(b'hello'); b.recv(4096)
b''
In this example, the first client sends "1234" and "hello", and gets its echo; the second one sends "12" and then "34", which is a perfectly valid password, but recv(4) doesn't guarantee that it'll get all four bytes at once. Remember how we said, in the Socket Programming chapter, that TCP is not a sequence of messages, but a stream of bytes? This means that recv(4) will actually return what is available so far — in our case, "12" and not "1234". This is not our password, so the server hangs up; the b'' is the client seeing its connection closed.
To fix it, we'd have to introduce a buffer to every connection, so we can accumulate its bytes and not lose them; and augment our state machine to remember that we're in the middle of receiving something, and how much of it is left. It's a pretty terrible way to live, but it's how things were done in the days of single-core machines, and it actually works: the CPU is so much faster than the I/O that it can handle thousands of such sessions at the same time, with a single thread.
And that's a big deal, really: no locks, no race conditions, and a log that actually reads in its proper order, since nothing is actually happening at the same time. You could also combine this technique with threads, for example, and have twenty cores, each one running a thread that juggles a thousand clients. But select is an engineering utility, not an abstraction — cumbersome and low-level, not something you'd want to build on top of. As our logic grows more complex, our complexity grows with it; we need a framework.
Callbacks
So let's build one on top of select. The first thing we need to consider is the concept of a callback: a function that we pass in to someone else, and ask them to call us back when they're ready — in other words, a function pointer, which we've already seen when we talked about sorted and its key function, or the publish/subscribe pattern of "call me back when this happens". We've also seen it in the higher-order functions and decorators of the previous chapters.
Usually, callbacks have a fixed signature: it's part of the interface, in fact, that says, "I'll call you like so, on such and such — so make sure you handle it thusly". In our case, we want to let the framework deal with the nitty-gritty details of juggling the sockets behind the scenes, and just call us back when there's some data to handle. This way, the framework owns the select loop, while we provide it with a handler for each connection:
>>> class Server:
... def __init__(self, host, port, Handler):
... self.host = host
... self.port = port
... self.Handler = Handler
... def start(self):
... listener = socket.socket()
... listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
... listener.bind((self.host, self.port))
... listener.listen(5)
... handlers = {listener: None}
... with listener:
... while True:
... ready, _, _ = select.select(handlers, [], [])
... for sock in ready:
... if sock is listener:
... connection, address = sock.accept()
... handlers[connection] = self.Handler(connection)
... elif handlers[sock].run() == 'done':
... sock.close()
... del handlers[sock]
>>> class Handler:
... def __init__(self, connection):
... self.connection = connection
... def run(self):
... raise NotImplementedError()
Let's break this down: we have a dictionary that maps every socket to its handler (and the listener to None); whenever a connection is readable, the server calls its handler's run method — and if it returns 'done', the connection is closed. As for the handler, it doesn't need to know much: it's only interested in its connection, and what to do when some data arrives. Admittedly, in an earlier version of this example, the server and the handler were glued together with inheritance — but I rewrote it to composition, passing the handler class into the server, so that the networking and the protocol can vary independently; the same lesson as before, when we didn't mix the thread with the service.
For example, we can have an echo handler, or a secure echo handler:
>>> class EchoHandler(Handler):
... def run(self):
... data = self.connection.recv(4096)
... if not data:
... return 'done'
... self.connection.sendall(data)
>>> class SecureEchoHandler(EchoHandler):
... def __init__(self, connection):
... super().__init__(connection)
... self.state = 'auth'
... def run(self):
... if self.state == 'echo':
... return super().run()
... if self.connection.recv(4) != b'1234': # still optimistic
... return 'done'
... self.state = 'echo'
>>> server = Server('127.0.0.1', 8002, SecureEchoHandler)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> a = socket.create_connection(('127.0.0.1', 8002))
>>> a.sendall(b'1234'); a.sendall(b'hello'); a.recv(4096)
b'hello'
>>> b = socket.create_connection(('127.0.0.1', 8002))
>>> b.sendall(b'4321'); b.recv(4096)
b''
The first one is trivial: it receives some data, and either echoes it back or says it's done. The second one, however, keeps some state in an attribute, and delegates to the first one's behavior once the client has authenticated; the right password results in an echo, the wrong one in disconnection. It's better, in that the protocol logic is separated from the networking, and we have nice little objects to handle the connections; but it's still somewhat optimistic about recv(4), and we'd have to implement the same accumulation logic as before — only now we'd have to do it in every handler that needs it.
And this points to a deeper problem: every recv has to be the last one, so to speak. The handler can't say "please receive four bytes, check them out, and if they're OK, keep echoing data indefinitely"; all it can do is say "here's what I want to do with whatever data arrives this time", and then remember its place in the conversation by storing it as part of self.state. Also, any handler that does something CPU-heavy will block everyone else, since the whole thing is done in a single thread.
Twisted
And it's not some toy idea, either; this is actually how it works in twisted, one of the oldest and most battle-tested networking frameworks in Python, first released in 2002. In its terminology, handlers are called protocols, and the select loop is called the reactor; but the idea is the same:
# twisted_echo.py
from twisted.internet import protocol, reactor
class EchoProtocol(protocol.Protocol):
def dataReceived(self, data):
self.transport.write(data)
class EchoFactory(protocol.Factory):
protocol = EchoProtocol
reactor.listenTCP(8003, EchoFactory())
reactor.run()
That is, a protocol subclasses protocol.Protocol and implements dataReceived, which the reactor calls with whatever data arrives (with self.transport representing the connection, and its write method sending data asynchronously — that is, twisted buffers the data for us, and writes it to the socket when it's writable, something our server didn't bother with). To connect it to a port, we create a factory, and use listenTCP to register it there; and to start the loop, we call reactor.run.
To test it, we can use the same client as before:
>>> import socket
>>> a = socket.create_connection(('127.0.0.1', 8003))
>>> a.sendall(b'hello'); a.recv(4096)
b'hello'
The only thing I haven't explained is why we need a factory; and the reason is that we need a new protocol for every new connection, with their own buffer and state — so the factory's protocol attribute is used to configure which class to instantiate for this purpose. The factory can also contain whatever the protocols share in common — like configuration, or a list of all the connected clients.
Now, let's make it secure. Since twisted doesn't give us four bytes, but rather whatever arrives, we'll have to accumulate it ourselves in self.data; but here, I'd like to demonstrate a neat little trick. Rather than having a state string, and an if-statement to handle it accordingly, we can make the state into the next method to be called:
# twisted_secure.py
from twisted.internet import protocol, reactor
class SecureEchoProtocol(protocol.Protocol):
def __init__(self):
self.data = b''
self.next = self.auth
def dataReceived(self, data):
self.data += data
self.next = self.next()
def auth(self):
if len(self.data) < 4:
return self.auth
password, self.data = self.data[:4], self.data[4:]
if password != b'1234':
self.transport.loseConnection()
return None
return self.echo()
def echo(self):
if self.data:
self.transport.write(self.data)
self.data = b''
return self.echo
class SecureEchoFactory(protocol.Factory):
protocol = SecureEchoProtocol
reactor.listenTCP(8004, SecureEchoFactory())
reactor.run()
That's SecureEchoProtocol, in twisted_secure.py. In dataReceived, we add the data to the buffer, call self.next, and whatever it returns, we set as the new self.next to be used for the next call. The auth method simply returns itself as long as less than four bytes have been accumulated, and once enough is available, peels them off and either hangs up on a wrong password or hands over to echo (which also takes care of any additional bytes that might have arrived with the password). Finally, echo writes out the buffer, and returns itself, so that's what dataReceived sets as the new self.next.
This self.next = self.next() idiom deserves a second look: because functions are objects, after all (and if you remember our discussion in the Functions chapters, that should ring some bells), then it's quite natural to have a state represented by the function that handles it, where each state simply determines the next one to transition into. This way, we can add a state simply by adding a method — no growing if-chain to maintain.
And it works:
>>> a = socket.create_connection(('127.0.0.1', 8004))
>>> a.sendall(b'12')
>>> a.sendall(b'34hello'); a.recv(4096)
b'hello'
>>> b = socket.create_connection(('127.0.0.1', 8004))
>>> b.sendall(b'4321'); b.recv(4096)
b''
The first one sends the password in two parts, so we see that it can indeed be accumulated across calls — and whatever arrives after that is echoed back. The second one sends a wrong password, which results in the connection being closed. As an aside, when I first showed this in class, a student caught a bug in it; these accumulate-and-dispatch logics are so easy to get subtly wrong that it's almost scary, and that's something to consider, too.
To be fair, twisted solved all the plumbing (buffered writes, the reactor, dozens of useful protocols already implemented) — and its deferreds, a chain of callbacks for results that don't exist yet, are an ancestor of the futures we'll see when we talk about asyncio. Still, it's pretty fragmented code; the logic of one conversation is spread over several methods, connected together by return values.
Callback Hell, and a Way Out
To demonstrate this cost, take the simplest possible conversation with an echo server: we send "1", receive a byte and print it, then send "2", receive a byte and print it. If we do it synchronously, it's just four lines, which you read top to bottom:
>>> sock = socket.create_connection(('127.0.0.1', 8005))
>>> sock.sendall(b'1')
>>> print(sock.recv(1))
b'1'
>>> sock.sendall(b'2')
>>> print(sock.recv(1))
b'2'
Now, imagine we'd have to do this with a framework that doesn't let us wait, but instead requires us to pass in a function to be called when the operation is complete. To test this out, I wrote a little stand-in, CallbackSocket, with async_send and async_recv — it just performs the operation and calls you back, but the overall shape is the same as with any other framework:
>>> class CallbackSocket: # a stand-in for a framework's socket
... def __init__(self, sock):
... self.sock = sock
... def async_send(self, data, callback):
... self.sock.sendall(data)
... callback()
... def async_recv(self, size, callback):
... callback(self.sock.recv(size))
>>> csock = CallbackSocket(sock)
>>> def f1():
... csock.async_send(b'1', f2)
>>> def f2():
... csock.async_recv(1, f3)
>>> def f3(data):
... print(data)
... csock.async_send(b'2', f4)
>>> def f4():
... csock.async_recv(1, f5)
>>> def f5(data):
... print(data)
>>> f1()
b'1'
b'2'
Five functions for four lines; and we're just getting started. Each function ends by passing the next one as a callback, so the logic is scattered across all of them, and you need to jump back and forth to piece together what happens. You could nest them, I guess, and end up with the infamous pyramid of doom:
>>> csock.async_send(b'1', lambda:
... csock.async_recv(1, lambda data:
... print(data) or csock.async_send(b'2', lambda:
... csock.async_recv(1, lambda data:
... print(data)))))
b'1'
b'2'
With print(data) or ... just to squeeze two statements in the same lambda. This is callback hell: highly fragmented code, with an execution flow that's unpredictable and hard to follow — and that's not even considering how error handling has to be propagated through all of it.
The way out is coroutines: functions with multiple entry points. On the face of it, it might sound a bit counterintuitive — if a function has x = 1 at the beginning, and we enter it somewhere else, what is x? But in fact, we've already seen such a situation in the Generators chapter: when a generator is advanced, it runs until it hits a yield statement and pauses, then resumes from this position when we pass in a value — all the way until the next yield, or until it returns and terminates, all the while retaining its local state.
Here's a reminder of how that looks:
>>> def gen(x):
... print(f'x: {x}')
... y = yield x
... print(f'y: {y}')
... z = yield y
... print(f'z: {z}')
... return z
>>> g = gen(1)
>>> g.send(2)
TypeError: can't send non-None value to a just-started generator
>>> next(g)
x: 1
1
>>> g.send(2)
y: 2
2
>>> g.send(3)
z: 3
StopIteration: 3
When we call the generator function, no code is actually executed — it just returns a new generator object. To activate it, we pass it to next (or send(None)); this runs it until it hits a yield statement, at which point it pauses, and returns the yielded value. Sending a value into the generator, on the other hand, will resume it from the same place and inject that value as the result of the yield expression (or None, under next).
So in our case, sending a value right away is an error, since there's no yield yet to receive it; we first run it with next, which prints x: 1 and yields 1. Then, we send(2), which resumes the generator from this position and assigns it to y, which becomes 2. We send(3) again, which assigns it to z and makes it 3 — and then it returns, so we get StopIteration: 3, the value being the return value of the generator.
Another interesting feature of generators is that throw can raise an exception in their flow, as if the yield statement had raised it:
>>> class Oops(Exception): pass
>>> def gen():
... while True:
... try:
... data = yield
... print('got', data)
... except Oops:
... print('caught it')
... return 'bye'
>>> g = gen(); next(g)
>>> g.send('hello')
got hello
>>> g.throw(Oops)
caught it
StopIteration: bye
This is as if we had thrown the exception from inside the generator, on the line of the yield statement where it's paused; and just like we can catch such an exception, so can the generator — in which case, it can return a value, so we get StopIteration: bye.
So as you can see, generators and coroutines are in fact the same mechanism — we call it a coroutine when we use the values that are sent in.
Daydreaming
Let me daydream for a moment: if we could somehow write asynchronous code in a way that's almost synchronous, how would our secure echo handler look? It's going to be a slow read, because this is the part where the brain really starts melting.
>>> class ConnectionClosed(Exception):
... pass
>>> def handle(connection):
... data = yield 4 # wake me up when 4 bytes are available
... if data != b'1234':
... return
... try:
... while True:
... data = yield # wake me up when anything is available
... connection.sendall(data)
... except ConnectionClosed:
... return
Read it as if the rules were already true: data = yield 4 means "call me again when you've accumulated four bytes"; a bare yield means "call me again when anything arrives"; and ConnectionClosed is some exception of our own, which the framework throws in when the client disconnects. I just made this all up on the spot; at the moment, it's pure fantasy. But if you compare it to the state machines we had before, it's a doll, if only we could make it run.
Here's how it might work: we'd have one thread, which would hold a bunch of generators, one for each connection, and run select on their behalf; essentially, a hundred logical threads that are not actually threads. The generators wouldn't do the blocking operation themselves — they'd yield it to someone else, because actually doing it would block everyone.
For every connection, the framework would keep a triple: the generator that describes its flow, the data that has been accumulated so far, and the amount of data it expects to arrive. Initially, it would be a fresh generator, None, and zero:
>>> class Server:
... def __init__(self, host, port, handle):
... self.host = host
... self.port = port
... self.handle = handle
... self.handlers = {}
... def start(self):
... listener = socket.socket()
... listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
... listener.bind((self.host, self.port))
... listener.listen(5)
... with listener:
... while True:
... ready, _, _ = select.select([listener, *self.handlers], [], [])
... for sock in ready:
... if sock is listener:
... connection, address = listener.accept()
... self.handlers[connection] = self.handle(connection), None, 0
... self.advance(connection if sock is listener else sock)
It would be pretty much the same as before, except that when a new connection arrives, we'd want to advance it right away, so that its generator gets a chance to run until its first request:
>>> def advance(self, connection):
... generator, data, size = self.handlers[connection]
... try:
... if data is None: # just started: run it until it asks for something
... size, data = next(generator), b''
... else: # it's readable, so this won't block
... chunk = connection.recv(4096 if size is None else size - len(data))
... if not chunk:
... generator.throw(ConnectionClosed)
... raise StopIteration() # in case it caught the exception and kept going
... data += chunk
... if size is None or len(data) >= size:
... size, data = generator.send(data), b''
... self.handlers[connection] = generator, data, size
... except StopIteration:
... connection.close()
... del self.handlers[connection]
>>> Server.advance = advance
This is where the fun begins: if the data is None, it means that the generator hasn't started yet, so we use next to get it going until its first yield; and that, we know, should be the size of the data it expects. Otherwise, the socket is readable, so receiving from it won't block; we receive just what's missing (or up to 4096 bytes, if it asked for any amount). If we get an empty chunk, it means that the client left, so we throw ConnectionClosed into the mix, and if the generator survives that, we stop it anyway. If we get enough bytes, we send them in, and whatever it yields next is what it expects to receive next. When the generator returns, that's StopIteration, so we close the connection and drop it.
It looks brain-breaking, but it isn't really that complex. And notice how the accumulation, which we'd otherwise have to rewrite in every handler, is now part of the framework; it's just boilerplate code that always looks the same. We can see it in action by sending the password in two parts, and peeking at the handler's state in between:
>>> server = Server('127.0.0.1', 8006, handle)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> a = socket.create_connection(('127.0.0.1', 8006))
>>> a.sendall(b'12')
>>> list(server.handlers.values())
[(<generator object handle at 0x101e87510>, b'12', 4)]
>>> a.sendall(b'34')
>>> a.sendall(b'hello'); a.recv(4096)
b'hello'
>>> b = socket.create_connection(('127.0.0.1', 8006))
>>> b.sendall(b'4321'); b.recv(4096)
b''
>>> a.close()
>>> server.handlers
{}
After "12", the state is the generator, the two bytes, and the four it's waiting for: the handler is asleep. When "34" arrives, the generator wakes up with b'1234', and from then on we get our echo; and if we provide a wrong password, we get disconnected. When the first client leaves, the handlers dictionary is empty again.
Now, with our little framework in place, a simple echo server is almost a non-event:
>>> def echo(connection):
... try:
... while True:
... connection.sendall((yield))
... except ConnectionClosed:
... pass
>>> server = Server('127.0.0.1', 8007, echo)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> c = socket.create_connection(('127.0.0.1', 8007))
>>> c.sendall(b'hello'); c.recv(4096)
b'hello'
And notice that we went from a class to a function; a single flow, interrupted by a single yield, is all we need. Just look at the lovely code we end up with. (Note the parentheses in sendall((yield)): a yield expression used as an argument needs to be enclosed in them.)
The Limits of Our Framework
Now, this is nice; but it also relies on all our handlers being well-behaved. What if some of them don't yield every time they're about to do something slow?
>>> def slow_echo(connection):
... try:
... while True:
... data = yield
... time.sleep(2) # or any heavy computation
... connection.sendall(data)
... except ConnectionClosed:
... pass
>>> server = Server('127.0.0.1', 8008, slow_echo)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> a = socket.create_connection(('127.0.0.1', 8008))
>>> b = socket.create_connection(('127.0.0.1', 8008))
>>> a.sendall(b'hello')
>>> with Timer():
... b.sendall(b'world')
... print(b.recv(4096))
b'world'
3901.978 ms
Let's see what happens if we try to use it. Client a sends "hello", and the handler sleeps for two seconds, which is a pretty long time. During this time, client b sends "world" and waits for a response; we time its execution with the Timer context manager from our performance chapter. And it waits for almost four seconds: first, for a's handler to wake up, because the server is stuck and can't even notice b's data; and then, for its own handler to finish. If a handler would call time.sleep(10), the entire server would be stuck for ten seconds.
A similar thing would happen if a handler would do any other kind of heavy work, like check whether some number a client sent is prime using our naive is_prime. In that case, everyone would wait, because the entire power of our single thread was in its zig-zagging, and a computation, unlike a network operation, can never be zig-zagged — it simply has to happen, and the thread has to plow through it. The solution would be to delegate any such heavy work to a thread pool (or even a process pool), and await its result asynchronously, so that even if all our workers are busy, the server is still able to serve others. But then, we need to equip our handlers with a richer vocabulary than "give me n bytes, please". Let's daydream a little more: what if our handlers would yield actual instruction objects, and the framework would dispatch them based on their type?
>>> import concurrent.futures, queue, select, socket, threading
>>> from prime_utils import is_prime
>>> class Receive:
... def __init__(self, size=None):
... self.size = size
>>> class Execute:
... def __init__(self, function, *args):
... self.function = function
... self.args = args
>>> def handle(connection):
... while data := (yield Receive()):
... n = int(data)
... prime = yield Execute(is_prime, n)
... connection.sendall(f'{n} is {"" if prime else "not "}prime'.encode())
This handler receives a number using yield Receive(), then does yield Execute(is_prime, n), and gets the result as if it has called the function — it can then send back whether the number is prime or not. When the client disconnects, Receive will return empty bytes, and the loop will end; for simplicity's sake, I didn't implement the accumulation of the requested size like before.
To make this work, we'd have to rewrite Server to handle such instructions:
>>> class Server:
... def __init__(self, host, port, handle):
... self.host, self.port, self.handle = host, port, handle
... self.waiting = {} # socket: (generator, instruction)
... self.done = queue.SimpleQueue() # results that came back from the threads
... self.wakeup, self.notify = socket.socketpair()
... self.executor = concurrent.futures.ThreadPoolExecutor()
... def start(self):
... listener = socket.create_server((self.host, self.port))
... while True:
... ready, _, _ = select.select([listener, self.wakeup, *self.waiting], [], [])
... for sock in ready:
... if sock is listener:
... connection, address = listener.accept()
... self.step(connection, self.handle(connection), None)
... elif sock is self.wakeup:
... sock.recv(4096)
... while not self.done.empty():
... self.step(*self.done.get())
... else:
... generator, instruction = self.waiting.pop(sock)
... self.step(sock, generator, sock.recv(instruction.size or 4096))
And add the step and finish methods:
>>> def step(self, connection, generator, value):
... try:
... instruction = generator.send(value)
... except StopIteration:
... return connection.close()
... if isinstance(instruction, Receive):
... self.waiting[connection] = generator, instruction
... elif isinstance(instruction, Execute):
... future = self.executor.submit(instruction.function, *instruction.args)
... future.add_done_callback(lambda future: self.finish(connection, generator, future.result()))
>>> def finish(self, connection, generator, result): # runs in the pool's thread
... self.done.put((connection, generator, result))
... self.notify.send(b'!') # wake the select up
>>> Server.step, Server.finish = step, finish
The step method, then, sends a value into a generator and peers at what it yields next. If it's a Receive object, it parks the generator in waiting, associated with its socket, until select tells us this socket is ready to read. If it's an Execute object, it submits it to the thread pool, providing a callback to be invoked when it's done.
That's the tricky part. This callback happens in a thread of the thread pool, while the generators should only be touched by the thread of our event loop (or we'd be back to races); and this thread is stuck in a select. So finish puts the result in a thread-safe queue, and also writes a byte to one end of a socket pair, whose other end is among the sockets we're selecting on. This wakes our event loop, which then drains the queue and resumes the generators accordingly. This self-pipe trick is actually how event loops in real-life frameworks let other threads wake them up.
So even if we have a client that tries to check a really big prime number, while another client asks about a small one:
>>> server = Server('127.0.0.1', 8009, handle)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> a = socket.create_connection(('127.0.0.1', 8009))
>>> b = socket.create_connection(('127.0.0.1', 8009))
>>> a.sendall(b'15485867')
>>> with Timer():
... b.sendall(b'15')
... print(b.recv(4096))
b'15 is not prime'
43.287 ms
>>> with Timer():
... print(a.recv(4096))
b'15485867 is prime'
491.788 ms
While a's number is being checked in a thread, b asks about 15, and gets an answer within a few dozen milliseconds; a's answer takes another half a second to arrive, but the point is that the server didn't stop serving anyone. With the GIL, it's still true that the computation competes with the event loop for the interpreter, but the interval at which they switch hands is such that the loop gets a turn every few milliseconds; and if we want the computation to be faster as well, we can use a process pool instead.
Now that you know what's possible, I can tell you what's actually out there. We've just written a basic framework that does something pretty cool: you write such a thing once, and everyone else can use it as a library. And it does exist, in the standard library, even: it's the next thing we're going to cover. But before we get there, I'd like to suggest an exercise: try to rebuild this framework from scratch yourself, because you'll get to know its design by heart; and getting it to actually work is a great exercise, and not an easy one at that. If you'd like a live coding session of a similar idea, you can watch David Beazley's talk on concurrency from the ground up.
Asyncio: from Yield From to Await
Admittedly, this is a pretty common problem — so much so, that it was solved once already, in the standard library: as part of the asyncio module, designed by Guido himself (and that's always a good sign), and added in Python 3.4. It's not exactly the same as what we did — it also supports writing, and a whole lot more — but the general philosophy is similar: it provides a way to run coroutines, which are suspended as they wait for something to happen, delegating control to an event loop that runs select (or something better) on their behalf.
In version 3.4, these coroutines were just generators decorated with @asyncio.coroutine, and they would wait using yield from. This expression, as we've seen, is used to delegate to another generator, passing values in and out of it; and in this case, it lets us compose coroutines in a way that we didn't have an answer to before — "how do you make one handler call another one?" data = yield from reader.read(4) was asyncio's version of our data = yield 4.
In 3.5, it was upgraded to a whole new syntax: async def to define such an asynchronous function, and await to pause it for some time — "this might take a while, so suspend me here and resume me later, when it's ready". Introducing new keywords is always a breaking change (since 3.7, async and await are fully reserved, and code that used them as variable names broke), so it's not a decision that was taken lightly, and it shows just how important asynchronous programming has become. The old decorator was deprecated, and it's gone since 3.11, as we can see:
>>> import asyncio
>>> asyncio.coroutine
AttributeError: module 'asyncio' has no attribute 'coroutine'. Did you mean: 'coroutines'?
>>> class Wait:
... def __init__(self, size):
... self.size = size
... def __await__(self):
... data = yield self.size
... return data
>>> async def handle():
... data = await Wait(4)
... print('woken up with', data)
>>> coroutine = handle()
>>> coroutine
<coroutine object handle at 0x10507f920>
>>> coroutine.send(None)
4
>>> coroutine.send(b'1234')
woken up with b'1234'
StopIteration
And await isn't magic — this shows it. Wait is actually an awaitable, an object whose __await__ method is a generator; and await Wait(4) actually delegates to it, the same way yield from would. When we drive the coroutine manually, we see that send(None) starts it, and we get the 4 it yielded, all the way up through the await; then, we can send in b'1234' to resume the coroutine with it, which it prints and finishes with StopIteration. This is basically our very own protocol, only with the new shiny keywords — and the whole point is that there's an event loop in the background, doing the sending for us. To see how it can be used, check out this simple example:
>>> import asyncio
>>> async def count():
... print(1)
... await asyncio.sleep(1)
... print(2)
>>> async def main():
... await asyncio.gather(count(), count(), count())
>>> with Timer():
... asyncio.run(main())
1
1
1
2
2
2
1002.289 ms
Each of the count coroutines prints 1, then sleeps for a second, then prints 2 — but thanks to asyncio.gather, which runs them concurrently, all three happen together. All three print 1 together, then all three sleep together, and all three print 2 together — so it takes just one second, not three. As for asyncio.sleep — that's the asynchronous version of time.sleep, which yields to the loop instead of blocking it.
>>> async def echo(reader, writer):
... while data := await reader.read(4096):
... writer.write(data)
... await writer.drain()
... writer.close()
>>> async def main():
... server = await asyncio.start_server(echo, '127.0.0.1', 8010)
... async with server:
... await server.serve_forever()
>>> threading.Thread(target=asyncio.run, args=(main(),), daemon=True).start()
>>> a = socket.create_connection(('127.0.0.1', 8010))
>>> b = socket.create_connection(('127.0.0.1', 8010))
>>> a.sendall(b'hello'); a.recv(4096)
b'hello'
>>> b.sendall(b'world'); b.recv(4096)
b'world'
This is a full-fledged asynchronous echo server. start_server invokes our handler for every new connection it receives, passing in a dedicated reader and a dedicated writer. await reader.read(4096) suspends the handler until some data arrives (or an empty result if the client was closed); and then it uses writer.write to buffer that data for sending, and await writer.drain() to wait until its buffer is emptied enough. This is called backpressure — a client that doesn't read, and lets us buffer up, can't get us to store unlimited amounts of data for them. Two clients, and only one thread (I ran asyncio.run(main()) in a background thread just to keep the REPL free for playing the clients).
main also showcases two more useful pieces: serve_forever runs our server until we cancel it, and the server object itself is an asynchronous context manager (we'll encounter those in a moment), which we use to clean up the server when we're done. Finally, our handler is also careful to close the writer when the client leaves, so the connection doesn't linger.
But before we dive in, I have one more tool for us to consider: Python actually comes equipped with its own asyncio REPL, python -m asyncio. In it, we can await directly at the top level (it actually runs an event loop for us; we'll see what that means in a moment), so we don't have to set up an event loop to play with it. From now on, the examples will run there:
$ python3.13 -m asyncio
asyncio REPL 3.13.9 (main, Oct 14 2025, 21:10:40) [Clang 20.1.4 ] on darwin
Use "await" directly instead of "asyncio.run()".
Type "help", "copyright", "credits" or "license" for more information.
>>> import asyncio
>>> await asyncio.sleep(1, result='hello')
'hello'
And note that the asyncio REPL imports asyncio for us — but if we import it again out of habit, it doesn't mind. Note also that sleep can even return a result. In this REPL, the loop runs in the main thread, and every input it gets it runs as part of the event loop, which is why if we type a blocking call here it will freeze everything else that's running in the background.
Tasks, Async For and Async With
Before we can use a coroutine, however, we need to await it — which is what brings it to life. In fact, the first lesson we learn is that calling an async def function doesn't actually run it: it returns a coroutine object, just like calling a generator function returns a generator object. If we forget to await it, it won't do anything at all:
>>> async def f():
... print('started')
... asyncio.sleep(1)
... print('done')
>>> await f()
started
<python-input-0>:3: RuntimeWarning: coroutine 'sleep' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
done
So f doesn't actually sleep for a second: it just creates a coroutine that, if properly executed, would sleep for a second. Python even warns us that our coroutine was never awaited, which is nice; it's a pretty classic mistake, so there's a whole warning dedicated to it (although the part about tracemalloc just tells us how to find where the object was created).
So we definitely need to await it — but how exactly does it work? Suppose we await it twice in a row, or use two tasks instead:
>>> async def f():
... print('started')
... await asyncio.sleep(1)
... print('done')
>>> async def g():
... await f()
... await f()
>>> with Timer():
... await g()
started
done
started
done
2002.455 ms
>>> async def g():
... task1 = asyncio.create_task(f())
... task2 = asyncio.create_task(f())
... print('tasks created')
... await asyncio.gather(task1, task2)
>>> with Timer():
... await g()
tasks created
started
started
done
done
1001.024 ms
The first version waits for one second, and then for another; the second only takes a second, not two. You see, await means "run this as part of me, and wait for it to finish", so when we use it twice in a row, one after another, that's exactly what happens: the two things happen sequentially. A task, however, is a different beast: it's a coroutine wrapped by asyncio.create_task, which schedules it on the event loop to run concurrently with whoever created it, like a logical thread. Also notice how we see "tasks created" even before either task starts, since all we've done so far is schedule them — and if we want them to start before we gather them, we have to:
>>> async def g():
... task1 = asyncio.create_task(f())
... task2 = asyncio.create_task(f())
... print('tasks created')
... await asyncio.sleep(0)
... print('tasks running')
... await asyncio.gather(task1, task2)
>>> await g()
tasks created
started
started
tasks running
done
done
In other words, a task only starts when the current coroutine yields control to the event loop — and await asyncio.sleep(0) is the idiom we use to say "yield, but come right back", effectively giving everyone else a chance to get in line and start doing something. And this is important, because nothing actually preempts a coroutine: it keeps running until it awaits something, which is why a coroutine that never awaits anything will simply block everyone else.
The order in which things happen, however, is deterministic: both callbacks and tasks are executed in the order in which they become ready, first come first served. With threads, it's up to the operating system to decide what to do when, but with coroutines, we are in control.
The iteration and context management protocols have asynchronous versions, too: __aiter__, __anext__, __aenter__ and __aexit__. These are used when something like getting the next item, or entering and exiting a context, can take a while — but then, we can do it in an asynchronous manner. For example:
>>> class Countdown:
... def __init__(self, n):
... self.n = n
... async def __aiter__(self):
... for i in range(self.n, 0, -1):
... await asyncio.sleep(1)
... yield i
>>> async for i in Countdown(3):
... print(i)
3
2
1
>>> async def countdown(name):
... async for i in Countdown(3):
... print(name, i)
>>> await asyncio.gather(countdown('a'), countdown('b'))
a 3
b 3
a 2
b 2
a 1
b 1
[None, None]
As you can see, async for works pretty much like for, but it calls __aiter__, then awaits __anext__ for every item. But since __aiter__ itself is an async def with a yield in it, that's an async generator — and it gives us both at once, just like generator functions gave us iterators. Note how our two countdowns are neatly interleaved: first a 3, then b 3, then a 2, then b 2 — since each sleeps, and then gets back in line. Also note that gather returns the list of results, which are two Nones. We can also do:
>>> class Connection:
... async def __aenter__(self):
... await asyncio.sleep(1)
... print('connected')
... return self
... async def __aexit__(self, exception, error, traceback):
... await asyncio.sleep(1)
... print('disconnected')
>>> async with Connection() as connection:
... print('inside')
connected
inside
disconnected
In which case async with will await __aenter__ when we go in, and await __aexit__ when we go out (just like the context managers we saw in Objects 5, with the same exit signature). This way, we can connect and disconnect from a network asynchronously, without blocking anyone else — and we've already seen it in action in our echo server, when we did async with server: and the server was closed for us as part of exiting the context. Like async for, it only works inside coroutines — or in the asyncio REPL.
Running Asyncio
So who runs the coroutines? The event loop, and it's the same role as our engine: a reactor that holds all the registered callbacks, tasks and sockets, and juggles them around. In Linux, it uses epoll; in macOS and BSD, kqueue; and as a last resort, plain select (through the selectors module). On Windows, the default event loop uses I/O completion ports instead.
Here's how asyncio.run looks in the regular Python REPL, and what it does if we ask for a loop when there isn't one running:
>>> import asyncio
>>> async def hello():
... await asyncio.sleep(1)
... print('hello world')
>>> asyncio.run(hello())
hello world
>>> asyncio.get_running_loop()
RuntimeError: no running event loop
>>> asyncio.get_event_loop()
RuntimeError: There is no current event loop in thread 'MainThread'.
So asyncio.run is the reactor being turned on: it creates a loop, runs the coroutine until it's done, and closes it. Outside of this, there is no loop at all: get_running_loop raises an error, and so does the older get_event_loop here (because asyncio.run didn't leave one behind). get_event_loop used to quietly create a loop if there wasn't one, which led to all sorts of confusing bugs; this behavior was deprecated in 3.12, and eventually removed in 3.14, so inside coroutines we should use get_running_loop.
In the asyncio REPL, however, a loop is already running (that's how we get that magical top-level await), and asyncio.run refuses to start a new one in the same thread:
>>> async def hello():
... await asyncio.sleep(1)
... print('hello world')
>>> asyncio.run(hello())
RuntimeError: asyncio.run() cannot be called from a running event loop
>>> loop = asyncio.get_running_loop()
>>> loop
<_UnixSelectorEventLoop running=True closed=False debug=False>
>>> loop.is_running()
True
Note that the loop is a selector event loop — our select from the beginning of the chapter, all grown up and ready to party. Don't call stop on a loop that you didn't start, though: in the REPL (or in Jupyter notebooks, where a similar loop is running in the background of every cell), it tends to break stuff. There can be at most one loop running at a time per thread, although different threads can run loops of their own. To illustrate that asyncio.run is not magic, here's a simplified version of it, written by hand:
>>> def run(coroutine):
... loop = asyncio.new_event_loop()
... asyncio.set_event_loop(loop)
... try:
... return loop.run_until_complete(coroutine)
... finally:
... pending = asyncio.all_tasks(loop)
... for task in pending:
... task.cancel()
... loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
... loop.run_until_complete(loop.shutdown_asyncgens())
... asyncio.set_event_loop(None)
... loop.close()
>>> run(hello())
hello world
Let's go over it, since we haven't seen anything like it before: first, we create a new loop and make it current. Then, we run the coroutine until it completes, and return its result; and finally, we do some cleanup that's easy to forget about: cancel any tasks that are still pending, run the loop until they've had a chance to handle it, finalize asynchronous generators, and close the loop. The original version also handles signals, debug modes and contexts, but you get the idea.
The run_until_complete method we saw runs the loop until a particular awaitable is complete; a server, however, would probably use run_forever, which does what it says on the tin — until someone calls stop:
>>> loop = asyncio.new_event_loop()
>>> handle = loop.call_later(1.5, loop.stop)
>>> task = loop.create_task(hello())
>>> with Timer():
... loop.run_forever()
hello world
1501.020 ms
>>> loop.is_running(), loop.is_closed()
(False, False)
>>> loop.close()
This runs our task, which prints hello world, and then stops the loop after 1.5 seconds, as instructed; and after it's over, the loop is neither running nor closed (until we close it ourselves).
If we'd like to run several top-level coroutines one after the other on the same loop (for example, as part of a test), there's a context manager for that:
>>> with asyncio.Runner() as runner:
... runner.run(hello())
... runner.run(hello())
hello world
hello world
So why don't we just await everything at the top level, all the time? Because without a running loop, there's nothing to drive the coroutine forward. Some environments run one for you: python -m asyncio, Jupyter, or pytest (if you install the pytest-asyncio plugin and mark tests with @pytest.mark.asyncio). And while the asyncio documentation throws you right in the deep end, with dozens of utilities and no quickstart, once you realize that it's all the API of our engine, it should start making a lot more sense.
Futures and Awaitables
At a lower level, there's also a future object. It's pretty much exactly what we called a future in concurrent.futures: a placeholder for a result that doesn't exist yet, albeit in this case it's designed for the event loop rather than threads (so it's not thread-safe). We can register callbacks on it, set its result (or exception) once, and even await it:
>>> future = asyncio.Future()
>>> future
<Future pending>
>>> future.add_done_callback(lambda future: print('callback:', future.result()))
>>> future.set_result(1)
callback: 1
>>> future
<Future finished result=1>
>>> future.set_result(2)
asyncio.exceptions.InvalidStateError: invalid state
The future starts out in a pending state; add_done_callback lets us register a function to be invoked with this future once it's done; and set_result marks it as such, causing the callback to be invoked (although, since it's scheduled on the loop, it actually happens as soon as the loop gets a chance to do so). Marking a future done more than once is an InvalidStateError: a future is done once and for all.
>>> future = asyncio.Future()
>>> async def waiter():
... print('waiting')
... result = await future
... print('got', result)
>>> task = asyncio.create_task(waiter())
waiting
>>> task
<Task pending name='Task-1' coro=<waiter() running at <python-input-1>:3> wait_for=<Future pending cb=[Task.task_wakeup()]>>
>>> future.set_result(42)
got 42
>>> task
<Task finished name='Task-1' coro=<waiter() done, defined at <python-input-1>:1> result=None>
This is the glue that connects callbacks to coroutines. If some function returns a pending future, which it resolves later, we can await it; our task will be suspended (its repr even tells us it's wait_for=<Future pending ...>), and when this future is resolved, our task will be woken up with the result. This is exactly what would have happened in our framework, had we implemented an instruction object for it.
So why do we need futures, if we have tasks? Because sometimes, the result we're interested in isn't even produced by a coroutine at all: a callback-based library, a thread, or even the operating system (like a socket becoming readable) might be the one to produce it — and a future is how the event loop represents "something that will be ready later". Even asyncio.sleep is implemented in this way: it creates a future, schedules a callback with call_later to set its result, and awaits the future.
To recap, there are several awaitables in asyncio — let's go over the taxonomy of those again. We have coroutines, which are what we get when we call an async def function — these run when we await them, as part of whoever is awaiting them. Then we have tasks, which are scheduled when we create them and run concurrently, in parallel to the current coroutine. And we have futures, which are never scheduled at all; they're just a box that someone else will fill. In fact, anything with an __await__ method can be awaited:
>>> import inspect
>>> coroutine = hello()
>>> inspect.isawaitable(coroutine), inspect.isawaitable(asyncio.Future())
(True, True)
>>> isinstance(asyncio.create_task(coroutine), asyncio.Future)
True
hello world
Note that a task is actually a future, too; it's a special subclass thereof, in which its result is set by running a coroutine. So in fact, we can do anything to a task that we can do to a future — check if it's done, get its result or exception, register callbacks and more. We can even cancel it, although it's somewhat different from cancelling a future (we'll cover it in a moment). There's even a function, asyncio.ensure_future:
>>> future = asyncio.Future()
>>> asyncio.ensure_future(future) is future
True
>>> task = asyncio.create_task(hello())
>>> asyncio.ensure_future(task) is task
True
>>> asyncio.ensure_future(hello())
<Task pending name='Task-2' coro=<hello() running at <python-input-0>:1>>
>>> await task
hello world
hello world
So ensure_future is a way to coerce anything that is awaitable into a future object: futures and tasks can pass through it unchanged, while a coroutine will be wrapped in a new task and immediately scheduled (which is why we see two hello worlds: one from task, and one from the task created by ensure_future). This functionality exists to support frameworks that accept "anything awaitable" — but in your own code, I recommend sticking to the more explicit create_task.
This is the same design pattern as a helper that coerces anything into a list — we leave lists as is, wrap strings or bytes (which are iterable, but we almost never intend to iterate over their characters) or anything else that's not iterable in a list, and turn anything that is iterable into a list. It's a useful trick to have at the boundary of an API that accepts both one thing and a list thereof — but elsewhere, it's usually a smell. In fact, when I was writing this live during a class, I initially wrote isinstance((str, bytes)), absentmindedly forgetting the object; it should be isinstance(x, (str, bytes)).
Tasks, Task Groups and Exceptions
Let's go back to tasks, our trusty workhorse. asyncio provides the create_task function, which immediately schedules a coroutine and returns the task object, which we can then await, or gather with others:
>>> t1 = asyncio.create_task(hello())
>>> t2 = asyncio.create_task(hello())
>>> await asyncio.gather(t1, t2)
hello world
hello world
[None, None]
>>> t1
<Task finished name='Task-1' coro=<hello() done, defined at <python-input-0>:1> result=None>
>>> t1.done(), t1.result()
(True, None)
>>> t3 = asyncio.create_task(hello())
>>> t3.result()
asyncio.exceptions.InvalidStateError: Result is not set.
>>> await t3
hello world
Once a task is gathered, it's done, so done() is true, and we can retrieve its result, which is the return value of its coroutine, or in this case, None. If we were to ask for the result of a task that isn't done yet, InvalidStateError would be raised, because futures don't block — you're supposed to await them.
One thing to note, though, is that the event loop only keeps weak references to tasks; so if we create a task, but don't keep a reference to it anywhere, it might very well be garbage collected in the middle of its execution. So it's important to keep them in a variable or a set — or better yet, in a task group.
asyncio.TaskGroup is a new addition to Python 3.11, and it's the recommended, structured way of dealing with tasks. The tasks created through a group are all awaited when its async with block is exited, so none of them can outlive the block in which they were created:
>>> async def double(x):
... await asyncio.sleep(1)
... return x * 2
>>> async with asyncio.TaskGroup() as group:
... t1 = group.create_task(double(1))
... t2 = group.create_task(double(2))
... print('tasks scheduled')
tasks scheduled
>>> t1.result(), t2.result()
(2, 4)
Notice how "tasks scheduled" is printed first, because creating them only schedules them to happen; and indeed, we wait for both of them at the end of the block, so afterwards we can retrieve their results.
What about exceptions? If a task raises one, it doesn't crash anything right away, because like a result, an exception is also stored in the task object:
>>> async def fail(error):
... await asyncio.sleep(1)
... raise error
>>> task = asyncio.create_task(fail(ValueError('oops')))
>>> task.exception()
asyncio.exceptions.InvalidStateError: Exception is not set.
>>> await asyncio.sleep(1.1)
>>> task.exception()
ValueError('oops')
>>> await task
ValueError: oops
Before the task is finished, exception() raises InvalidStateError too; and afterwards, it returns the exception, so if we await the task we get the same error (if nobody ever retrieves it, asyncio at least logs a warning when the task is garbage collected: "Task exception was never retrieved").
However, if we gather such tasks:
>>> await asyncio.gather(fail(ValueError(1)), fail(TypeError(2)))
ValueError: 1
>>> await asyncio.gather(fail(ValueError(1)), fail(TypeError(2)), return_exceptions=True)
[ValueError(1), TypeError(2)]
By default, gather raises the first exception to happen — which means we lose the other one, and not only that: we don't cancel the rest of the awaitables either, so they keep running. We could pass return_exceptions=True to have them returned instead of raised, but then we'd have to sift through the list ourselves. What we'd really want in such a situation is to raise all of them, and in Python 3.11 this was added: ExceptionGroup, a new exception that can contain multiple exceptions, and a new syntax to handle them.
If we implement our own version of gather, we can raise an ExceptionGroup with all the failures, which would have this traceback:
>>> async def gather(*awaitables):
... results = await asyncio.gather(*awaitables, return_exceptions=True)
... errors = [result for result in results if isinstance(result, Exception)]
... if errors:
... raise ExceptionGroup('some tasks failed', errors)
... return results
>>> await gather(fail(ValueError(1)), fail(ValueError(2)), fail(TypeError(3)))
+ Exception Group Traceback (most recent call last):
| File "<python-input-2>", line 1, in <module>
| await gather(fail(ValueError(1)), fail(ValueError(2)), fail(TypeError(3)))
| File "<python-input-1>", line 5, in gather
| raise ExceptionGroup('some tasks failed', errors)
| ExceptionGroup: some tasks failed (3 sub-exceptions)
+-+---------------- 1 ----------------
| Traceback (most recent call last):
| File "<python-input-0>", line 3, in fail
| raise error
| ValueError: 1
+---------------- 2 ----------------
| Traceback (most recent call last):
| File "<python-input-0>", line 3, in fail
| raise error
| ValueError: 2
+---------------- 3 ----------------
| Traceback (most recent call last):
| File "<python-input-0>", line 3, in fail
| raise error
| TypeError: 3
+------------------------------------
The traceback is pretty interesting: it presents the group as a tree, with the group's traceback on top, and each sub-exception underneath it, along with their traceback. We can then use except* to handle value errors and type errors separately:
>>> try:
... await gather(fail(ValueError(1)), fail(ValueError(2)), fail(TypeError(3)))
... except* ValueError as group:
... print('value errors:', group.exceptions)
... except* TypeError as group:
... print('type errors:', group.exceptions)
value errors: (ValueError(1), ValueError(2))
type errors: (TypeError(3),)
>>> async def slow():
... try:
... await asyncio.sleep(10)
... except asyncio.CancelledError:
... print('slow was cancelled')
... raise
>>> try:
... async with asyncio.TaskGroup() as group:
... t1 = group.create_task(slow())
... t2 = group.create_task(fail(ValueError(1)))
... except* ValueError as group:
... print('caught', group.exceptions)
slow was cancelled
caught (ValueError(1),)
Each clause matches all the exceptions of a particular type — in our case, the first clause gets the two value errors, the second one the type error, and both clauses are executed (unlike except, which only runs the first clause that fits). That's also how TaskGroup works: if any task fails, it cancels the others (as we see slow complaining that it was cancelled), then waits for them to finish, and raises an ExceptionGroup of everything that failed; that's what structured concurrency is about — making sure that no task is left behind.
Cancellation and Timeouts
Any task can be cancelled, simply by calling cancel() on it: the next time its loop would resume it, a CancelledError is raised inside it, at the await statement where it's currently suspended. If it's awaiting another coroutine, this error will surface in the innermost one first, and propagate outwards from there:
>>> async def g():
... try:
... await asyncio.sleep(1)
... print('g done')
... except asyncio.CancelledError:
... print('g cancelled')
... raise
>>> async def f():
... try:
... await g()
... print('f done')
... except asyncio.CancelledError:
... print('f cancelled')
>>> task = asyncio.create_task(f()); task.cancel()
True
>>> await task
asyncio.exceptions.CancelledError
>>> task = asyncio.create_task(f())
>>> await asyncio.sleep(0); task.cancel()
True
g cancelled
f cancelled
>>> await task
>>> task.cancelled()
False
If we cancel the task before it's even started, it's not going to happen at all — not a single line of f will execute, and trying to get a result out of it will raise a CancelledError. But if we cancel it after it's already started, the error will surface as soon as asyncio.sleep(0) lets it happen — first inside g's sleep, where it will say "well, I'm cancelled, then, let's get it over with", and re-raise it, and then in f, where it will be caught again. And note how here, f actually swallows the cancellation — so the task is considered to have ended "successfully" (that is, cancelled() is False). Don't do that; if you catch a CancelledError, it's OK to use it for some cleanup, but you should re-raise it, because whoever cancelled you — whether it's a timeout, or a task group, or simply the shutdown of asyncio.run — is unable to do its job if you ignore this request. This is exactly why it's a BaseException rather than an Exception — so you don't accidentally swallow it with an except Exception clause.
Sometimes, however, it really is necessary to protect some operation from being interrupted halfway through, like when committing a transaction; for that, asyncio provides the asyncio.shield function:
>>> async def f():
... try:
... await asyncio.shield(g())
... print('f done')
... except asyncio.CancelledError:
... print('f cancelled')
>>> task = asyncio.create_task(f())
>>> await asyncio.sleep(0); task.cancel()
True
f cancelled
>>> await task
g done
This creates a shield around g, so cancelling f (from its perspective, anyway — the await statement raising a CancelledError) doesn't necessarily mean g is cancelled: it continues running, undisturbed, in its own task, and announces its completion a second later. This doesn't mean, however, that the outer operation is protected — the shield protects the inner one, not the other way around.
Timeouts are just a particular case of cancellation, really — one initiated by a clock rather than an explicit request. Since version 3.11, we have the asyncio.timeout context manager: if we do something inside it, and it takes too long, our task is cancelled, and this is turned into a TimeoutError as soon as we exit the context:
>>> async def work(n):
... await asyncio.sleep(n)
... return n
>>> async with asyncio.timeout(2):
... print(await work(1))
1
>>> async with asyncio.timeout(1):
... print(await work(2))
TimeoutError
>>> await asyncio.wait_for(work(2), timeout=1)
TimeoutError
Here, one second of work is definitely within our two second budget, so the task completes successfully — but two seconds of work within one second raises a TimeoutError (as of 3.11, this is the actual exception raised by asyncio's timeout, rather than a separate asyncio.TimeoutError class). Similarly, the last line uses wait_for, which is the older and less elegant version of the same thing, accepting just one awaitable object and cancelling it after the given timeout.
This is useful, but sometimes we just want to stop waiting, without cancelling anything — for this, we have asyncio.wait. Like its name implies, it simply waits for a set of tasks (optionally, with a timeout), and returns two sets: those that have completed, and those that are still pending — without cancelling either. We can then do something with the done tasks, and retry the pending ones later (or cancel them, if we want):
>>> tasks = [asyncio.create_task(work(n)) for n in (0, 2)]
>>> with Timer():
... done, pending = await asyncio.wait(tasks, timeout=1)
1001.511 ms
>>> [task.result() for task in done], pending
([0], {<Task pending name='Task-2' coro=<work() running at <python-input-0>:2> wait_for=<Future pending cb=[Task.task_wakeup()]>>})
>>> tasks = [asyncio.create_task(work(n)) for n in (3, 1, 2)]
>>> for next_done in asyncio.as_completed(tasks):
... print(await next_done)
1
2
3
After one second, it's clear enough that the zero-second task is done, while the two-second task is still pending — but also still running, so we can wait for it again (or cancel it, if we change our mind). We can even pass return_when=FIRST_COMPLETED (or FIRST_EXCEPTION), in which case wait will return as soon as one of the tasks is done (or raises an exception). And if we don't want to deal with sets of tasks, we can use as_completed instead: it similarly accepts a list of tasks, but returns an iterator that yields awaitables in the order they finish — 1, 2, 3 — regardless of the order in which we pass them in, much like its namesake in concurrent.futures.
CPU-Bound Work, Threads and Scheduling
So far, we assumed our coroutines are spending their time awaiting — but what if they don't? To break this assumption, let's bring back our old friend is_prime, imported from the previous chapter's prime_utils module:
>>> from prime_utils import is_prime
>>> async def f():
... print('f started')
... print('f done:', is_prime(15485867))
>>> async def g():
... print('g started')
... await asyncio.sleep(0)
... print('g done')
>>> with Timer():
... await asyncio.gather(f(), g())
f started
f done: True
g started
g done
[None, None]
505.082 ms
Here, f gets to run first, and since it never suspends, g doesn't even get a chance to start until it's done — and that's half a second of a frozen event loop. In a webserver, that's half a second during which nobody is being served, no matter how many clients are waiting; the same lesson we got from our framework's sleepy handler. To avoid it, we could do:
>>> async def f():
... print('f started')
... print('f done:', await asyncio.to_thread(is_prime, 15485867))
>>> with Timer():
... await asyncio.gather(f(), g())
f started
g started
g done
f done: True
[None, None]
508.218 ms
>>> import concurrent.futures
>>> async def f():
... loop = asyncio.get_running_loop()
... with concurrent.futures.ProcessPoolExecutor() as executor:
... print('f done:', await loop.run_in_executor(executor, is_prime, 15485867))
>>> with Timer():
... await asyncio.gather(f(), g())
g started
g done
f done: True
[None, None]
1060.695 ms
Here, asyncio.to_thread (which was added in 3.9) runs the function in a separate thread, and returns an awaitable that we can suspend on to get its result; so f will pause, letting g run, and then resume once the thread is done. It uses the loop's default thread pool, and is the same trick that our instruction framework's Execute used, so it's good to know what it actually does under the hood; it also works, but with the GIL, the computation itself won't be faster — for that, we need loop.run_in_executor, which can accept any executor (including a ProcessPoolExecutor; note we use get_running_loop, not the old get_event_loop). Here, it took a bit more than a second, since most of it was spent starting up the worker processes.
This is what people mean when they say that asynchronous programming is infectious: any blocking call inside a coroutine will stall the entire thread, so anything it calls has to be asynchronous as well, or else be explicitly hoisted onto a thread.
Going in the opposite direction, what if we have code running in another thread, and we want it to communicate with the event loop? Such a thread doesn't have a running event loop of its own, so it can't just invoke asyncio directly:
>>> def thread():
... asyncio.get_running_loop()
>>> t = threading.Thread(target=thread); t.start(); t.join()
Exception in thread Thread-1 (thread):
Traceback (most recent call last):
...
File "<python-input-2>", line 2, in thread
asyncio.get_running_loop()
~~~~~~~~~~~~~~~~~~~~~~~~^^
RuntimeError: no running event loop
>>> loop = asyncio.get_running_loop()
>>> def thread():
... future = asyncio.run_coroutine_threadsafe(hello(), loop)
... print('submitted', future)
>>> t = threading.Thread(target=thread); t.start(); t.join()
submitted <Future at 0x105d72120 state=pending>
hello world
What it can do is use run_coroutine_threadsafe, which takes a coroutine and a loop, and makes sure it's scheduled on this loop in a safe manner from any thread. It returns a concurrent.futures.Future (note the different representation — this kind is thread-safe, and can be waited on with a timeout), which this thread can then wait for without blocking the loop:
>>> def thread():
... future = asyncio.run_coroutine_threadsafe(hello(), loop)
... done, pending = concurrent.futures.wait([future], timeout=2)
... print('done' if done else 'still waiting')
>>> t = threading.Thread(target=thread); t.start(); t.join()
still waiting
hello world
>>> t = threading.Thread(target=thread); t.start()
>>> await asyncio.to_thread(t.join)
hello world
done
>>> def thread():
... loop.call_soon_threadsafe(print, 'hello from the loop')
>>> t = threading.Thread(target=thread); t.start(); t.join()
hello from the loop
This is actually a trap, because it's a deadlock in disguise: the thread is waiting for the coroutine, which needs the loop, whose thread is blocked by t.join(), waiting for the thread. Eventually, the waiting times out ("still waiting"), at which point the loop is finally free to say hello; but if we were to join using await asyncio.to_thread(t.join) instead, the loop wouldn't be blocked, and the coroutine would run and the thread would finish. To schedule a plain function, we can use loop.call_soon_threadsafe, which registers a callback to be executed on the loop from another thread (by writing to its self-pipe to wake it up, much like the socket pair in our framework). Generally, asyncio's objects (futures, queues, locks, and most of the loop's methods) are not thread-safe; these _threadsafe functions are the bridges, and everything else should only be touched by the loop's thread.
Finally, the event loop also provides functionality to schedule plain callbacks, which is the basis on which futures and tasks are built:
>>> def hello(name):
... print('hello', name)
>>> loop = asyncio.get_running_loop()
>>> handles = [
... loop.call_later(2, hello, 'later'),
... loop.call_at(loop.time() + 1, hello, 'at'),
... loop.call_soon(hello, 'soon'),
... ]
hello soon
>>> await asyncio.sleep(2.1)
hello at
hello later
>>> loop.call_later(1, hello, 'never').cancel()
call_soon runs the callback as soon as the loop gets a chance (so here, as soon as the input finishes), call_at runs it at a particular timestamp (according to the loop's own clock, loop.time(), which is a monotonic one, not the wall clock), and call_later runs it after a certain delay. All three return handles, which can be cancelled (which is why "never" never prints).
Streams, Subprocesses, Queues and Locks
Asyncio also provides us with some high-level APIs for the things we most commonly need to wait for. For example, it has a stream abstraction, which is effectively its version of a socket. We can create a server with start_server and a client with open_connection: both give us a reader and a writer, through which we can communicate with the other party.
>>> async def handle(reader, writer):
... print('connection from', writer.get_extra_info('peername'))
... while line := await reader.readline():
... writer.write(line.upper())
... await writer.drain()
... writer.close()
... await writer.wait_closed()
>>> server = await asyncio.start_server(handle, '127.0.0.1', 8011)
>>> reader, writer = await asyncio.open_connection('127.0.0.1', 8011)
connection from ('127.0.0.1', 58180)
>>> writer.write(b'hello\nworld\n'); await writer.drain()
>>> await reader.readline()
b'HELLO\n'
>>> await reader.readexactly(3)
b'WOR'
>>> await reader.readuntil(b'\n')
b'LD\n'
>>> writer.close(); await writer.wait_closed()
>>> server.close(); await server.wait_closed()
This is an uppercasing line server, and a client that uses it. The server's handler receives a reader and a writer, and can get information about who connected to it by calling get_extra_info('peername'). The client writes two lines, then "drains" the connection — readline returns everything until a newline (including it); readexactly(3) returns exactly three bytes, which is something our framework did for us, but it's nice to see it here too (it raises IncompleteReadError if the stream ends too soon); and readuntil returns everything up to a certain separator. There's also read(n), which returns as soon as something is available (up to n bytes). As for closing the connection, it's a two-step affair: we call close to initiate it, and then await wait_closed(), both for the connection and for the server.
Under the hood, streams are implemented with transports and protocols, which is actually a design idea from twisted: a protocol that defines how we interact with the stream (by implementing methods like data_received), and a transport that handles the communication details, buffering our writes until they can be sent. The streams, then, wrap them in convenient coroutines; and there are also versions that work with Unix sockets, start_unix_server and open_unix_connection.
Similarly, asyncio provides us with a way to work with subprocesses, in a way that's pretty similar to what we've already seen: we call create_subprocess_exec (to run a program with some arguments) or create_subprocess_shell (to run a command line in a shell), and get back a process whose pipes are streams.
>>> process = await asyncio.create_subprocess_exec(
... 'echo', 'hello world', stdout=asyncio.subprocess.PIPE,
... )
>>> await process.stdout.readexactly(5)
b'hello'
>>> await process.stdout.read()
b' world\n'
>>> await process.wait()
0
>>> async def run(command):
... process = await asyncio.create_subprocess_shell(command, stdout=asyncio.subprocess.PIPE)
... stdout, _ = await process.communicate()
... return stdout
>>> with Timer():
... print(await asyncio.gather(run('sleep 1; echo a'), run('sleep 1; echo b')))
[b'a\n', b'b\n']
1018.848 ms
>>> process = await asyncio.create_subprocess_exec('sleep', '10')
>>> try:
... await asyncio.wait_for(process.wait(), timeout=1)
... except TimeoutError:
... process.terminate()
>>> await process.wait()
-15
readexactly(5) gets us "hello", read() gets us the rest, and wait gets us the exit code. The fact that two sleep 1 commands are executed in one second is also nice, since it means our loop is smart enough to wait for both processes at once. And lastly, this is the asynchronous version of execute from the previous chapter: wait_for is used to enforce the timeout, and terminate sends the process a SIGTERM (that's why its exit code is -15).
Finally, there's the question of synchronization: even though we're dealing with a single thread, different tasks can still interfere with each other, since they can be interleaved at any await. This means that shared state can still become corrupted, and asyncio provides us with its own version of the threading primitives we saw before: Queue, Lock, Event, Condition and Semaphore. They provide pretty much the same interface as before, except that you await them — and they're not thread-safe, since they're intended to synchronize tasks, not threads.
>>> async def producer(queue):
... for i in range(3):
... await asyncio.sleep(0.5)
... await queue.put(i)
... await queue.put(None)
>>> async def consumer(queue):
... while (item := await queue.get()) is not None:
... print('consumed', item)
>>> queue = asyncio.Queue()
>>> await asyncio.gather(producer(queue), consumer(queue))
consumed 0
consumed 1
consumed 2
[None, None]
Just like before, we have a producer that puts a number every half second, and a consumer that awaits to get it and print it; the None sentinel is used to shut the process down. It's exactly like the queue.Queue of the previous chapter — except, you know, with awaits.
>>> lock = asyncio.Lock()
>>> async def critical(name):
... async with lock:
... print(name, 'in')
... await asyncio.sleep(0.5)
... print(name, 'out')
>>> await asyncio.gather(critical('a'), critical('b'))
a in
a out
b in
b out
[None, None]
>>> async def producer(queue):
... async with lock:
... for i in range(3):
... await queue.put(i)
... await queue.put(None)
>>> async def consumer(queue):
... while True:
... async with lock:
... item = await queue.get()
... if item is None:
... break
... print('consumed', item)
>>> queue = asyncio.Queue()
>>> async with asyncio.timeout(2):
... await asyncio.gather(consumer(queue), producer(queue))
TimeoutError
We can also use an asyncio.Lock to serialize two critical sections: b won't get in until a is out, even though a is pausing in the middle. But we can also get deadlocks: here, the consumer holds the lock while it waits for get, and the producer can't acquire the same lock to put anything; so each waits for the other forever, until our timeout kicks in and throws a TimeoutError. The point from the previous chapter still holds, even for asynchronous programming: don't wait for something while holding a lock that might be needed for it to happen. And had the producer tried to acquire the lock first, everything would work fine — it's just that the moment you have something that can deadlock, it's usually a matter of timing, and that's what makes deadlocks so nasty.
A Way of Thinking About the World
So let me spell it out: we started with select, juggling sockets by hand. Then we wrapped it in callbacks, which broke our flow into fragments. Then we rearranged it into coroutines, which are just generators that yield what they're waiting for to an engine, and were delighted to rediscover the same design, but already polished, in asyncio, where async and await make asynchronous code almost indistinguishable from synchronous.
It's really cool, if you think about it: we get to use our CPU at full capacity while waiting for I/O, and run many execution flows on a single thread, without having to spawn any OS threads or worry about their preemption. Behind the scenes, the event loop is juggling everything — but our code is just invoked as callbacks on different events, goes dormant when it awaits something, and wakes up on the same line when it's ready, as if nothing had happened. I don't know if I'm conveying my excitement well enough, but this is really cool.
But it also has its drawbacks. First, it melts your brain: the order of execution is no longer the order of lines, and it takes some practice to reason about which task happens when (it took me several iterations to get it right, so it's OK if it takes you a few, too). Second, it's infectious: any blocking call inside a coroutine will clog the one and only thread that runs everything. If you write db.users.create(name='alice'), without giving it a second thought, what you don't see is that packets are actually being sent to the database process, which then opens a transaction, commits and replies — and all of this takes a moment, during which your entire system is stuck. So really, you have to write user = await db.users.create(name='alice') instead, which means create has to be an async def, as well as anything it calls — and if your driver happens to be blocking, like the classic psycopg2 for PostgreSQL, you can't use it, and have to find an asynchronous version.
Third, if it's all happening in a single process, it's all or nothing: if you want an asynchronous system, then everything that might be called from a coroutine has to be async, and pay this full price of complexity. That is, unless you use microservices: then, you can quarantine the infection — only the parts that need the performance of asynchronous execution actually have to be asynchronous: namely, the servers and gateways that are juggling thousands of connections. A parser that subscribes to a message queue, grinds the CPU to parse what it gets and publish the result — such a component doesn't benefit from being asynchronous (it's busy parsing, so it wouldn't be able to serve anyone else while parsing, anyway). So, it can remain synchronous, and the gateway can publish to the queue, while the parser subscribes to it, without ever directly calling each other; in fact, the parser could even be written in a different language! And that's also how you scale the gateway: instead of raising two hundred servers to handle two hundred clients, you raise one to juggle them all, and simply throw the heavy work into a queue to be handled however it's handled.
So really, asynchronous programming is not just a library or a couple of keywords, it's a way of thinking about the world: it's inherently asynchronous, things just happen, and it's just a question of whether you wait for each of them to happen in turn, or say "sit aside, you, what do you want?" and get back to them when they're ready. Even a microservice that subscribes to a queue and is called back whenever a message arrives is thinking in this way.
The place where you'll encounter all of this most, though, is the web: servers that juggle clients, frameworks with both synchronous and asynchronous flavors, and the protocol that they all speak. This is what we're going to cover next, starting with how the web actually works.