Example #1: asyncio, with callbacks

Let's start by showing what this looks like using a traditional callback approach. (The two examples after this will demonstrate curio and asyncio's version of async/await-based APIs, which is what most people will want to use – this first example is to provide context for those.) I'll demonstrate with asyncio's "protocol" API, though the basic design here is inherited almost directly from the immensely influential Twisted (a Twisted version is also available for the curious). Here's the complete code, and then I'll give some commentary on how it works:

 1import sys
 2from functools import partial
 3import traceback
 4import asyncio
 5
 6class OneWayProxySource(asyncio.Protocol):
 7    def __init__(self, loop, server_task_container, dest_host, dest_port):
 8        self.loop = loop
 9        self.server_task = server_task_container[0]
10        self.dest_host = dest_host
11        self.dest_port = dest_port
12
13    def connection_made(self, transport):
14        # Stop listening for new connections
15        self.server_task.cancel()
16
17        # Save our transport
18        self.transport = transport
19
20        # Disable reading until the destination is ready.
21        self.transport.pause_reading()
22
23        # Connect to the destination
24        self.dest_protocol = OneWayProxyDest(self.loop, self.transport)
25        coro = self.loop.create_connection(lambda: self.dest_protocol,
26                                           self.dest_host, self.dest_port)
27        task = self.loop.create_task(coro)
28        def connection_check_for_failure(fut):
29            exc = fut.exception()
30            if exc is not None:
31                print("Failed to connect:")
32                # This isn't really right -- it doesn't handle exception
33                # chaining etc. I lack the will to worry about it.
34                traceback.print_tb(exc.__traceback__)
35                self.transport.abort()
36                self.loop.stop()
37        task.add_done_callback(connection_check_for_failure)
38
39    def data_received(self, data):
40        self.dest_protocol.send_data(data)
41
42    def connection_lost(self, exc):
43        self.dest_protocol.close()
44
45class OneWayProxyDest(asyncio.Protocol):
46    def __init__(self, loop, source_transport):
47        self.loop = loop
48        self.source_transport = source_transport
49
50    def connection_made(self, transport):
51        self.transport = transport
52        # Okay, now we're ready for data to start flowing
53        self.source_transport.resume_reading()
54
55    def send_data(self, data):
56        self.transport.write(data)
57
58    def close(self):
59        self.transport.write_eof()
60
61    def connection_lost(self, exc):
62        self.source_transport.abort()
63        self.loop.stop()
64
65def main(source_port, dest_host, dest_port):
66    loop = asyncio.get_event_loop()
67    server_task_container = []
68    coro = loop.create_server(
69        lambda: OneWayProxySource(loop, server_task_container,
70                                  dest_host, dest_port),
71        "localhost", source_port)
72    server_task = loop.create_task(coro)
73    server_task_container.append(server_task)
74    loop.run_forever()
75    loop.close()
76
77if __name__ == "__main__":
78    try:
79        args = [int(sys.argv[1]), sys.argv[2], int(sys.argv[3])]
80    except Exception:
81        print("Usage: {} SOURCE_PORT DEST_HOST DEST_PORT".format(__file__))
82    else:
83        main(*args)

There's a lot going on here, and the details aren't that important; like I said, this is mostly here to provide context for the next two examples. As a rough outline to get the idea, though:

• OneWayProxySource manages the incoming connection. When a connection is made (line 13) it first does some bookkeeping, then starts the outgoing connection (lines 24-36). Then when incoming data is received, or the incoming connection is closed, it forwards that on to the outgoing connection (lines 38-42).
• OneWayProxyDest manages the outgoing connection. In particular, it's responsible for actually sending data (lines 54-58), and shutting down the program once all the data has been sent (lines 60-61).
• main has the job of setting up the listening socket and arranging for incoming connections to be allocated a OneWayProxySource object (lines 66-69).
• And then there's the usual if __name__ == __main__ boilerplate at the bottom.

There's two main things I want to take away here:

1. The control flow is not at all straightforward or easy to follow.
2. All the actual reading and writing takes place "off-screen". By the time our data_received callback is run, someone else has already done the work of reading data off the network and into memory, and when we send data using self.transport.write, that doesn't actually do any sending. (How could it? Writing data takes time, and we aren't allowed to block.) Instead, what it does is queue the data to be sent later. This is also why we can't just shut down after calling self.transport.write_eof() – that just schedules the socket to be closed later, and we have to wait until OneWayProxyDest.connection_lost is called to let us know that the closure has actually happened.

Example #2: curio, with async/await

Now here's the equivalent program, but with curio. It looks very different, and I'll go through it in more detail:

 1import sys
 2from functools import partial
 3import curio
 4
 5READ_SIZE = 20000
 6
 7async def main(source_port, dest_host, dest_port):
 8    main_task = await curio.current_task()
 9    bound_cb = partial(proxy, dest_host, dest_port, main_task)
10    await curio.tcp_server("localhost", source_port, bound_cb)
11
12async def proxy(dest_host, dest_port, main_task, source_sock, addr):
13    await main_task.cancel()
14    dest_sock = await curio.open_connection(dest_host, dest_port)
15    async with dest_sock:
16        await copy_all(source_sock, dest_sock)
17
18async def copy_all(source_sock, dest_sock):
19    while True:
20        data = await source_sock.recv(READ_SIZE)
21        if not data:  # EOF
22            return
23        await dest_sock.sendall(data)
24
25if __name__ == "__main__":
26    try:
27        args = [int(sys.argv[1]), sys.argv[2], int(sys.argv[3])]
28    except Exception:
29        print("Usage: {} SOURCE_PORT DEST_HOST DEST_PORT".format(__file__))
30    else:
31        curio.run(main(*args))

First our main function sets up a listening socket, and arranges for the proxy function to be invoked on each incoming connection:

 7async def main(source_port, dest_host, dest_port):
 8    main_task = await curio.current_task()
 9    bound_cb = partial(proxy, dest_host, dest_port, main_task)
10    await curio.tcp_server("localhost", source_port, bound_cb)

When a connection arrives, proxy first tells main to stop listening (line 13), since we only want to handle one connection. Then it sets up the outgoing connection (line 14), and invokes copy_all:

12async def proxy(dest_host, dest_port, main_task, source_sock, addr):
13    await main_task.cancel()
14    dest_sock = await curio.open_connection(dest_host, dest_port)
15    async with dest_sock:
16        await copy_all(source_sock, dest_sock)

And copy_all, finally, implements the core logic of a proxy: copying data from one socket to another in a loop:

18async def copy_all(source_sock, dest_sock):
19    while True:
20        data = await source_sock.recv(READ_SIZE)
21        if not data:  # EOF
22            return
23        await dest_sock.sendall(data)

I hope we can all agree that in terms of readability, this is a huge improvement over the callback-based version. Ignoring imports and the __main__ boilerplate, we've gone from 67 lines of code down to 17 (four times shorter!), and the logic is now straightforward and procedural. Instead of having to manually check for everything that could go wrong and abort connections, we can just use with blocks and let exceptions propagate. That's the power of async/await.

Example #3: asyncio, with async/await

We can also write this example using asyncio's async/await-based "streams" API layer, and it looks very similar to the curio version:

 1import sys
 2from functools import partial
 3from contextlib import closing
 4import asyncio
 5
 6READ_SIZE = 20000
 7
 8async def main(loop, source_port, dest_host, dest_port):
 9    connect_event = asyncio.Event()
10    server_closed_event = asyncio.Event()
11    bound_cb = partial(proxy,
12                       loop, connect_event, server_done_event,
13                       dest_host, dest_port)
14    server = await asyncio.start_server(bound_cb, "localhost", source_port,
15                                        loop=loop)
16    await connect_event.wait()
17    server.close()
18    await server.wait_closed()
19    server_done_event.set()
20
21async def proxy(loop, connect_event, server_closed_event,
22                dest_host, dest_port,
23                source_reader, source_writer):
24    connect_event.set()
25    try:
26        with closing(source_writer):
27            tmp = await asyncio.open_connection(dest_host, dest_port, loop=loop)
28            dest_reader, dest_writer = tmp
29            with closing(dest_writer):
30                await copy_all(source_reader, dest_writer)
31    finally:
32        await server_done_event.wait()
33        loop.stop()
34
35async def copy_all(source_reader, dest_writer):
36    while True:
37        data = await source_reader.read(READ_SIZE)
38        if not data:  # EOF
39            return
40        dest_writer.write(data)
41
42if __name__ == "__main__":
43    try:
44        args = [int(sys.argv[1]), sys.argv[2], int(sys.argv[3])]
45    except Exception:
46        print("Usage: {} SOURCE_PORT DEST_HOST DEST_PORT".format(__file__))
47    else:
48        loop = asyncio.get_event_loop()
49        loop.create_task(main(loop, *args))
50        loop.run_forever()

In particular, notice how the core copy_all function here is almost identical to the curio version, modulo some spelling adjustments like read versus recv.

There is one source of extra complexity that ends up making the core logic here almost twice as long as in the curio version: the need to figure out when everything has completed so that the event loop can be safely shut down. In curio, the general rule is straightforward: the event loop exits when all (non-daemonic) tasks have finished. Here we have two tasks (main and proxy), so when they're both done, the loop exits. Asyncio doesn't provide any equivalent – we can use run_until_complete to run the loop until any one task exits, but this may leave arbitrary other tasks and callbacks unfinished. Instead, our two tasks have to manually coordinate to make sure that both have finished their work and cleaned up before we call loop.stop(), and this takes a bit of doing (lines 9-10, 16-19, 24, 32-33).

There's also another difference that doesn't show up on the page: the curio code pretty much directly does what it says, e.g. await sock.recv initiates a read from the socket and suspends until it completes. The asyncio code, on the other hand, is written as a layer on top of the protocol/transport system we saw in the first example. Now you see why we needed that first example! In the asyncio code, even though it doesn't look like we're using protocols, there's still a data_received callback somewhere that's stashing data in an internal buffer where source_reader.read can find it, and when we call dest_writer.write that ultimately turns into a call to a transport.write method, which schedules the writing to happen off-screen. The idea is that this is something the end-user doesn't have to think about; unfortunately, as we'll see, it doesn't quite work out that way.

And finally, there's one more extremely important difference between these examples: the asyncio protocols code has a showstopper bug. The asyncio async/await code has the same showstopper bug, plus a less important second bug... and a third, different, showstopper bug. Yet, remarkably, the curio code – despite being shorter and easier to understand – is correct as originally written.

Bug #1: backpressure

This bug affects both of the asyncio examples, but not the curio example.

Imagine that our example code is being used to proxy between two different networks that run at different speeds: data is arriving on the incoming socket at 3 MB/s, but the outgoing socket is only transmitting at 1 MB/s. What happens?

Artist's impression: Lord OOM rising from the deep and turning its baleful gaze upon a containerized app.

Remember what we said about how transport.write doesn't actually send any data, but rather just adds it to a buffer to be sent later? With the two asyncio examples shown above, that's exactly what will happen: each second we'll add 3 MB of data to the buffer, while 1 MB are removed, so on net our buffer will grow by 2 MB/s second until we eventually wake the out-of-memory killer. Of course, we probably won't catch this in testing, so it'll make a nice 2 am surprise someday.

And even if we don't actually run out of memory, we'll introduce potentially epic amounts of latency: after 10 seconds, we'll have accumulated 20 MB in our buffer, which means that a byte that arrives from the sender will sit in our process for 20 seconds before being forwarded on to the receiver; after 10 minutes, well... you get the idea. We are failing to apply backpressure, one of the cardinal sins of network programming. The elements of a distributed system can't function well if they aren't getting accurate feedback from their peers – and even just a single TCP connection is already a complex distributed system involving two userspaces, two kernel socket layers, two kernel packet layers, and who knows how many routers and competing flows. Backpressure is important.

For the asyncio protocols-based implementation, this can be fixed through some judicious use of the flow control callbacks and flow control commands. I'll leave the details as an exercise for the reader, but basically we want to call pause_reading after each chunk is received, and then define a resume_writing callback that calls resume_reading. (The Twisted protocol implementation also has this bug, and can be fixed in a similar way, albeit using some essentially undocumented APIs; issues around backpressure seem to be a perennial challenge for Twisted.)

For the asyncio streams-based implementation, StreamReader automatically uses the {pause,resume}_reading methods to transmit backpressure upstream, and StreamWriter provides a friendly wrapper around {pause,resume}_writing to help us accept backpressure from downstream: the drain method – we just have to remember to use it. So in order to fix our proxy to transmit backpressure, all we need to do is to add one line of code to copy_all. Specifically, this line:

35async def copy_all(source_reader, dest_writer):
36    while True:
37        await dest_writer.drain()
38        data = await source_reader.read(READ_SIZE)
39        if not data:  # EOF
40            return

In curio, things are different, because of the critical await that was already present:

18async def copy_all(source_sock, dest_sock):
19    while True:
20        data = await source_sock.recv(READ_SIZE)
21        if not data:  # EOF
22            return
23        await dest_sock.sendall(data)

In curio, there are no hidden buffers – sendall doesn't return until the OS has accepted the data to be sent. This has the effect of automatically propagating backpressure, without our having to remember to do anything. Our original example worked because in curio, it would take actual effort to get this wrong.

Bug #2: read-side buffering

This bug mostly affects the asyncio + async/await example, somewhat affects the asyncio + callbacks example, and doesn't affect the curio example.

We mentioned above how we should try to minimize buffering in order to keep latency down. Unfortunately, there's another source of extraneous buffering in the streams-based asyncio code, on the StreamReader side.

The asyncio internals can be difficult to follow, but I believe that at a steady state with data coming in faster than it can be processed, on Unix-like system, the user-space buffer in StreamReader will oscillate between 128 KiB and 384 KiB. (I calculated this as 128 KiB = twice the _DEFAULT_LIMIT; 384 Kib = 128 KiB + 256 KiB.) For our proxy example, this is pretty bad – it's pure "bufferbloat", adding latency without adding any value. In our example with 1 MB/s outgoing bandwidth, this buffer adds an extra ~130-400 ms of steady-state latency for no good reason; on a slower connection (e.g. mobile) this could easily become multiple seconds.

In the curio version, this problem doesn't arise, again because of curio's lack of userspace buffering: curio doesn't issue any recv syscalls until you actually call await sock.recv. This is really the right thing to do, because reading from a socket is (paradoxically!) a data transmitting action: after you call the recv syscall, the kernel will literally send out a packet to the remote peer asking them to send more data ("opening the receive window"). So you shouldn't call recv until you really are ready for more data, and you should always request exactly as much data as you're ready to process, no more. Curio makes this natural.

As bugs go, StreamReader's buffering is not as severe as the other two we discuss here – it won't cause crashes or data corruption, and receive-side backpressure isn't as universal a concern as send-side backpressure. (Everyone has to be prepared to handle a slow client, but perhaps some programs can process incoming data so quickly that they don't need to worry about fast clients.) Still, proxies are one place where this matters, and if we were seriously implementing some kind of proxy – like a SOCKS or HTTP proxy, or ssh port-forwarding – then asyncio's streams layer is probably not a good choice as it currently stands. (Working directly at the protocol layer would be somewhat better, because the buffer is reduced, but it's still not ideal in this respect – the curio style still gives us more control over the receive window while being simpler to use.)

I don't have a fix to show here. This seems to be an intrinsic limitation of the hybrid design strategy – because of the way the async/await-centric StreamReader is built on top of asyncio's callback-centric protocol layer, it's not obvious how or whether asyncio could switch to the just-in-time-recv model.

Bug #3: closing time

There's a final showstopper bug in the asyncio streams example (only), which is pretty obvious once you see it, but not so obvious how to fix. Here's the core of the proxy function again for reference. After the incoming connection has been closed, copy_all returns and the with closing(dest_writer) block calls dest_writer.close(), and then we stop the event loop:

try:
    with closing(source_writer):
        tmp = await asyncio.open_connection(dest_host, dest_port, loop=loop)
        dest_reader, dest_writer = tmp
        with closing(dest_writer):
            await copy_all(source_reader, dest_writer)
finally:
    await server_done_event.wait()
    loop.stop()

The problem is that again, it's dest_writer.close(), not await dest_writer.close() – we don't wait for the socket to actually close, we just make a note to close the socket later, once the send buffer has finished emptying out. But then we immediately stop the event loop before that can happen (maybe – it's a race condition), so some data will get dropped on the floor and lost. We need to wait for the close to complete before stopping the loop.

But how? Unless I've missed something, the StreamWriter API actually does not provide any mechanism for detecting when the stream has been closed (!). But we might reason that since the close is delayed until all data has been written, we can trick the close into happening promptly by draining the send buffer first:

try:
    with closing(source_writer):
        tmp = await asyncio.open_connection(dest_host, dest_port, loop=loop)
        dest_reader, dest_writer = tmp
        try:
            await copy_all(source_reader, dest_writer)
        finally:
            await dest_writer.drain()
            dest_writer.close()
finally:
    await server_done_event.wait()
    loop.stop()

Unfortunately, this isn't enough. As the docs warn us, drain doesn't actually block until all data is written; it only guarantees that the unwritten data is less than the "high water mark", whose default is undocumented but currently appears to be 64 KiB, and specifically tries to make sure that there's at least a "low water mark" worth of unsent data (default 16 KiB). So adding the drain call makes this bug harder to hit, and it might seem to work in testing (especially since you need a slow network connection to really increase the odds, and who runs their tests over a slow network connection?), but sooner or later we're going to randomly lose data.

To really fix the problem, we need to get rid of this high-water mark thing. This can be done by calling transport.set_write_buffer_limits(0) on the underlying transport object; then drain will only return once the send buffer is completely empty. Unfortunately, the only supported way to get access to the transport object is to copy-paste the implementation of asyncio.open_connection into our code, and add the call there. (It's lucky we aren't implementing a bi-directional proxy; if we were, then we'd also have to do this to the implementation of asyncio.start_server – in general, all of the stream helper functions are seem to currently be unusable if you need to be able to cleanly shut down a write stream in a protocol where you have the last word.)

Is this safe, though? The documentation specifically warns us not to do this: "Use of zero for either limit is generally sub-optimal as it reduces opportunities for doing I/O and computation concurrently."

Superficially, this warning seems to make a lot of sense. Our network card can only send data at certain moments, and in a modern operating system with pre-emptive multi-tasking, there's no way to guarantee that our application will be ready to hand off a packet at the exact moment when the network card wants it. So we need some sort of buffer that we load up with data ahead of time, and that the network card can read at its leisure. And if this buffer ever runs completely empty, the network card will go idle, which would be bad and waste bandwidth – so our application needs to wake up to refill the buffer before it goes empty, to give us a bit of a cushion to keep things going while we're getting the new data ready. That's the concurrent computation and I/O that the docs are referring to, and low-water/high-water logic provides this cushion.

But... while I'm not an expert on networking (certainly not like these folks!), and this stuff can be hella subtle... I'm pretty sure this reasoning is all completely wrong. Because the thing is... we already have a buffer that does all that stuff: the kernel's socket send buffer. The send syscall doesn't actually write packets to the network – it just enqueues our data into a buffer inside the kernel, where the lower-level networking stack knows to look for them when it wants them. select and friends don't wait for the send buffer to be empty before marking the socket writeable – they implement some low-water/high-water logic. And on modern systems, the kernel will even do fancy stuff like automatically tuning the buffer size depending on the speed of the connection and the amount of memory pressure the system is under. Plus, for various reasons the kernel buffer is usually too big. Adding a second user-space send buffer on top of this seems entirely superfluous. (Especially since the way asynchronous I/O works, whenever our application is blocked on the CPU then it means we aren't running through the event loop and letting it hand off data from the userspace buffer to the kernel. So how would this concurrent I/O and computation even work?)

As far as I can tell, the historical reason the asyncio userspace send buffer exists has nothing to do with performance. It's a necessary evil motivated purely by the need to let transport.write be non-blocking, so that callback-based programming becomes less painful. The asyncio low-watermark/high-watermark logic seems to be based on it then being confused for a different kind of buffer. So the docs are wrong: for optimal performance the watermarks should always be set to zero, to reduce bufferbloat and let the kernel buffer do its job.

Okay, phew. Having dealt with that, are we finally finished? Not quite.

We can now be confident that all our data will be transmitted to the outgoing socket, but it's still possible that the socket itself will remain open at the time our event loop exits. This is a bit of a nit-picky point. In our example, it actually doesn't matter, because as soon as the event loop exits then the program exits as well, and that will close down the socket. But in general this can be important. For example, in your test suite you should probably be running each test with its own isolated event loop, to make sure that different tests can't interfere with each other – but if each event loop you shut down leaves behind some dangling resources, multiply that by the number of tests you run and you might have a problem. Twisted's unit test framework goes to great lengths to annoy people into getting this right.

So, let's assume we'd like to actually, finally, for real, close our socket before we stop the event loop. We're calling close. What else do we need? Well, if we grovel around in the asyncio source for long enough, we discover that StreamWriter.close calls transport.close, but transport.close doesn't actually close the socket. Instead, it schedules a call to _call_connection_lost on the next event loop iteration, and that's what actually closes the socket. So if we want to actually close the socket, we have to yield to the event loop between our call to dest_stream.close() and our call to loop.stop(). I believe that yielding once should be enough to make this happen deterministically, since even if on the next iteration we get resumed first and call loop.stop() before the socket has closed, the event loop should still finish calling all currently-scheduled callbacks before it actually stops. I think. So long as those internal details don't change.

I also wonder what about what happens if some error is detected during socket shutdown, and how that's supposed to propagate out through this API.

Anyway.

In conclusion, here's a final version of our simple asyncio streams-based proxy server. This version still has extraneous buffering on the receive side, but it now does transmit back-pressure and (I think) reliably and correctly cleans up after itself. The lines that had to be added/changed are highlighted:

 1import sys
 2from functools import partial
 3from contextlib import closing
 4import asyncio
 5
 6READ_SIZE = 20000
 7
 8# Contains code derived (and modified) from the asyncio library, which is
 9# distributed under the Apache 2 license:
10#   https://github.com/python/asyncio/blob/master/COPYING
11# asyncio is copyright its authors:
12#   https://github.com/python/asyncio/blob/master/AUTHORS
13@asyncio.coroutine
14def fixed_open_connection(host=None, port=None, *,
15                          loop=None, limit=65536, **kwds):
16    if loop is None:
17        loop = asyncio.get_event_loop()
18    reader = asyncio.StreamReader(limit=limit, loop=loop)
19    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
20    transport, _ = yield from loop.create_connection(
21        lambda: protocol, host, port, **kwds)
22    ###### Following line added to fix buffering issues:
23    transport.set_write_buffer_limits(0)
24    ######
25    writer = asyncio.StreamWriter(transport, protocol, reader, loop)
26    return reader, writer
27
28async def main(loop, source_port, dest_host, dest_port):
29    connect_event = asyncio.Event()
30    server_closed_event = asyncio.Event()
31    bound_cb = partial(proxy,
32                       loop, connect_event, server_closed_event,
33                       dest_host, dest_port)
34    server = await asyncio.start_server(bound_cb, "localhost", source_port,
35                                        loop=loop)
36    await connect_event.wait()
37    server.close()
38    await server.wait_closed()
39    server_closed_event.set()
40
41async def proxy(loop, connect_event, server_closed_event,
42                dest_host, dest_port,
43                source_reader, source_writer):
44    connect_event.set()
45    try:
46        with closing(source_writer):
47            tmp = await fixed_open_connection(dest_host, dest_port, loop=loop)
48            dest_reader, dest_writer = tmp
49            try:
50                await copy_all(source_reader, dest_writer)
51            finally:
52                await dest_writer.drain()
53                dest_writer.close()
54                # To let the socket actually close
55                await asyncio.sleep(0, loop=loop)
56    finally:
57        await server_closed_event.wait()
58        loop.stop()
59
60async def copy_all(source_reader, dest_writer):
61    while True:
62        await dest_writer.drain()
63        data = await source_reader.read(READ_SIZE)
64        if not data:  # EOF
65            return
66        dest_writer.write(data)
67
68if __name__ == "__main__":
69    try:
70        args = [int(sys.argv[1]), sys.argv[2], int(sys.argv[3])]
71    except Exception:
72        print("Usage: {} SOURCE_PORT DEST_HOST DEST_PORT".format(__file__))
73    else:
74        loop = asyncio.get_event_loop()
75        loop.create_task(main(loop, *args))
76        loop.run_forever()

HTTP servers

Here's a simple stress test for an HTTP server: send it lots of GET requests, without ever reading the responses:

import sys, socket

host, port = sys.argv[1], int(sys.argv[2])

with socket.create_connection((host, port)) as sock:
    get = b"GET / HTTP/1.1\r\nHost: " + host.encode("ascii") + b"\r\n\r\n"
    while True:
        sock.sendall(get)

It turns out that if you point this at a server running twisted.web, then our client will upload a few megabytes of data and then the server will crash. (Here's an example server in case you want to try at home; make sure to hit control-C before your laptop starts Death Swapping.) This is a backpressure bug: the server reads the first GET, generates the response, and writes that to its send buffer. Where it sits, because the client isn't reading. But the server isn't paying attention to the buffer, so as far as it's concerned the data has been sent, and it goes on to process the next request, generates another response, and appends that to the send buffer. Repeat until the send buffer swallows all available memory and the server falls over. We shouldn't exactly panic over this – this just is a denial-of-service attack, and it's impossible to fully defend against DoS. What makes this a bit embarrassing is the degree of amplification: if there's some URL on a server that returns, say, a 100 KB response body, then each ~1 MB of upload from a single client will permanently swallow ~2 GB of memory.

Aiohttp (example server) is more robust against this particular attack – it will also crash eventually, but doesn't show the severe amplification. The reason is that aiohttp calls StreamWriter.drain after processing each request, so the send buffer can't grow to unbounded sizes. The reason it still crashes eventually is a bit weird: it never applies back-pressure on the receive side, even though that's the case that StreamReader will actually handle for you automatically... but aiohttp turns out to use the protocol API instead and its own ad hoc buffering code, which happily accepts and queues infinitely many request bodies without ever pushing back. So the amplification factor here is just 1x – this is a bug, but a relatively minor one. If an attacker is willing to upload a few gigabytes of data to an aiohttp server, they can crash it – but there are probably other things an attacker can do with a few gigabytes of data that are even worse (e.g. opening tens of thousands of connections), so meh.

On the other hand, remember the problems we had with data being lost at shutdown? It turns out that aiohttp has a "graceful shutdown" mode, which gives all current connections time to finish before exiting – but as far as I can tell this uses StreamWriter.close and doesn't disable the low-water mark stuff, so it actually has no way to know when the connections have finished closing. I haven't verified this experimentally, but I strongly suspect that the "graceful" shutdown is randomly chopping off some of those connections before they're finished.

If we turn to look at the the toy curio-based HTTP server that I wrote for some docs (yes, this says something about the relative maturity of the twisted/asyncio/curio ecosystems), then it avoids all of these problems. (Well, technically, it doesn't implement graceful shutdown, but if it did then it wouldn't run into the StreamWriter.close bugs.) Now, you've probably figured out by now that I'm the kind of paranoid person who worries about these things, so I won't pretend that I didn't think about this at all while writing that code. But when I thought about it, I realized that with curio, the most obvious naive implementation actually was correct, so I didn't need to do anything special.

With twisted and asyncio, you have to do something special. And everyone makes mistakes.

Websocket servers

Consider a websocket server that accepts connections and then sends the client an ongoing, infinite stream of messages. This is a pretty common configuration – examples would include IRC-style chat apps, or a live twitter feed viewer. Now imagine what happens if we connect to such a websocket and then go idle, never reading:

import sys, asyncio, aiohttp

URL = sys.argv[1]

async def constipate(url):
    session = aiohttp.ClientSession()
    async with session.ws_connect(url) as ws:
        await asyncio.sleep(100000000)

asyncio.get_event_loop().run_until_complete(constipate(URL))

A naive server that doesn't respond to backpressure will keep trying to send us messages, buffer them in memory, and eventually crash. This attack effectively has an amplification factor of infinity: all we have to do is send two packets to set up the connection, and then we can walk away while the server slowly leaks to death. In fact, this can easily happen by accident when a client establishes a connection and then crashes or otherwise goes offline.

Do any Python websocket servers implement this naive buffering algorithm? As far as I can tell, yes, by default, they all do. In fact, they all use a very similar API for sending messages that makes it difficult to avoid: they provide some kind of synchronous-colored write_message method that queues a message and then returns immediately, and it causes the same kind of troubles that we had with asyncio's synchronous-colored write method above. [Edit: it turns out there's an exception that I missed – the websockets package provides exactly the API that I advocate. Awesome!]

More specifically:

• aiohttp (example server): Doesn't respond to backpressure from slow clients by default. AFAICT has no API for doing so, so handling the disappearing client case is currently not possible without somehow changing the protocol use itself. The API for receiving messages does use await and thus should be able to transmit backpressure upstream to clients who send too fast (though I haven't tested whether this works myself).
• autobahn (example server): Doesn't respond to or transmit backpressure by default. When running on twisted, then it is possible to get backpressure notifications for slow clients using the twisted API I mentioned earlier. When running on asyncio, this seems to be unimplemented, so responding to backpressure is impossible. Doesn't have any API for transmitting backpressure upstream to clients who are sending data too quickly.
• tornado (example server): Doesn't respond to or transmit backpressure by default. Has APIs available that allow a sufficiently dedicated programmer to implement both upstream and downstream backpressure. (Tornado FWIW is also immune to the GET flood attack described above – they've clearly put a lot of thought into these kinds of issues.)

Even for the libraries where handling websocket backpressure is possible in theory, I couldn't find any mention of this in their examples, so it's doubtful that there are many deployments that actually take advantage of this.

My goal isn't to shame the authors of these packages – obviously they're helping people solve real problems right now, and all I have to offer is some hypothetically-improved vaporware! Rather, I want to point out that there's a whole set of nasty edge-cases that are very difficult to handle when using the conventions of traditional callback-based APIs. But if these servers had exposed a causality-respecting API for websockets – one with methods like async def receive_message and async def write_message – then these issues would simply go away. It's very rare that we can solve genuinely hard problems like this just by changing some API conventions – we should be excited!

The tornado API is particularly instructive, because it's so close to what I recommend: their write_message method returns a future. This means that in the asyncio/tornado async/await integration that allows await to be applied to Futures, one can write code like await websock.write_message(...) and get correct backpressure handling. But, if we forget, and write websock.write_message(...) instead... then the code will still seem to work! So despite the similarity, this isn't really a causality-preserving method, the way it would be if it were implemented as an async method that required await to run. It's a method that spawns a concurrent thread of execution to do the actual work, while also providing an option to join that thread. Much better than nothing! But the end result is that if you look at their official examples, they don't actually check the return value.

Of course tornado itself can't easily fix this, because they have to worry about backcompat and Python 2 support. But looking to the future, if we want to habitually write reliable software without breaking our brains, then causality needs to be opt-out, not opt-in, and the great thing about async/await is that it makes causal APIs just as easy to use as non-causal ones.

Orphan tasks

Curio obviously supports spawning new Tasks, but once spawned it can be rather tricky to manage them properly. In particular, the event loop's supervision guarantees that they will eventually clean up and exit. But unless you're very careful it's easy to get into situations where a task has crashed and no-one notices, or where a parent task has been cancelled but the child task continues on oblivious. For example, a common pattern I've run into is where I want to spawn several worker tasks that act like "part of" the parent task: if any of them raises an exception then all of them should be cancelled + the parent raise an exception; if the parent is cancelled then they should be cancelled too. We need ergonomic tools for handling these kinds of patterns robustly.

Fortunately, this is something that's easy to experiment with, and there's lots of inspiration we can draw from existing systems: Erlang certainly has some good ideas here. Or, curio makes much of the analogy between its event loop and an OS kernel; maybe there should be a way to let certain tasks sign up to act as PID 1 and catch failures in orphan tasks? I think we'll see rapid progress here.

Cleanup in generators and async generators

Generators and async generators present a somewhat stickier problem. If you think about it, a generator is something very like an independent thread of execution that runs arbitrary code. Given our analysis above, this should make us nervous!

Fortunately, stepping through a generator with __iter__ and __next__ turns out to be compatible with our rules, because while the generator acts somewhat like an independent thread semantically, each step gets executed as regular function call that's sort of grafted onto a regular callstack – so the context is correct, exceptions will propagate, etc.

The problem is that there's another piece to the generator API, that not everyone realizes is even there. It's the __del__ method. If we have a generator with some sort of cleanup code, like:

def some_generator(path):
    try:
        handle = open(path)
        yield ...
    finally:
        handle.close()

and we iterate it, but stop before reaching the end:

for obj in some_generator(path):
    break

# or

for obj in some_generator(path):
    raise ...

then eventually that finally block will be executed by the generator's __del__ method (see PEP 342 for details).

And if we think about how __del__ works, we realize: it's another sneaky, non-causal implicit-threading API! __del__ does not get executed in the context of the callstack that's using the generator – it happens at some arbitrary time and place.

In the special case where you're using CPython, and there are no reference loops involving your generator object, then CPython's use of reference counting does at least guarantee that __del__ will be called at the right time. I.e., as soon as the last reference is dropped then CPython will immediately pause the thread that dropped that reference and execute __del__ right there. However, this still takes places in a special context where exceptions are discarded. Besides which, in most general purpose code you probably shouldn't assume that you're on CPython and that there are no reference loops, in which case all bets are off: generator __del__ methods can easily end up executing arbitrary code without respecting causality, exception propagation, access to the correct task-local storage, timeout restrictions, ... basically all of our rules and the guarantees they're trying to provide just go out the window.

That's for regular, synchronous generators. Async generators are slightly different, because the reference counting GC part doesn't apply. Even if we're in the happy case on CPython where __del__ gets called synchronously on our callstack, then it still can't actually run async cleanup code directly, because __del__ is sync-colored. (This is a consequence of the weird environment where __del__ methods run, similar to the reason they have to discard exceptions.) PEP 525 provides an API for async generator __del__ methods to hand off to an event loop to spawn a full-fledged Task to run the actual cleanup. Compared to synchronous generators this is kind of an improvement, since regular __del__ methods can run at any moment, like pre-emptively scheduled threads – which Glyph told us is bad – while this async generator cleanup code at least gets scheduled on a cooperative thread and thus respects the use of yield points as a synchronization mechanism. But on the other hand, an implicitly spawned thread is still an implicitly spawned thread, and the fact that this is the only way to run async generator __del__ methods means that we lose even CPython's weak guarantees about when they'll run: so they will never respect causality, exception propagation, access to the correct task-local storage, timeout restrictions, etc.

This one worries me, because it's basically the one remaining hole in the lovely interlocking set of rules described above – and here it's the Python language itself that's fighting us.

For now, the only solution seems to be to make sure that you never, ever call a generator without explicitly pinning its lifetime with a with block. For synchronous generators, this looks like:

def some_sync_generator(path):
    with open(path) as ...:
        yield ...

# DON'T do this
for obj in some_sync_generator(path):
    ...

# DO do this
from contextlib import closing
with closing(some_sync_generator(path)) as tmp:
    for obj in tmp:
        ...

And for async generators, this looks like:

async def some_async_generator(hostname, port):
    async with open_connection(hostname, port) as ...:
        yield ...

# DON'T do this
async for obj in some_async_generator(hostname, port):
    ...

# DO do this
class aclosing:
    def __init__(self, agen):
        self._agen = agen

    def __aenter__(self):
        return self._agen

    def __aclose__(self, *args):
        await self._agen.aclose()

async with aclosing(some_async_generator(hostname, port)) as tmp:
    async for obj in tmp:
        ...

It might be possible for curio to subvert the PEP 525 __del__ hooks to at least catch cases where async generators are accidentally used without with blocks and signal some kind of error.

PEP 533 is one possible proposal for fixing this at the language level, by adding an explicit __iterclose__ method to the iterator protocol and adapting Python's iteration constructs like for accordingly.

Do you really think everyone's going to abandon callbacks?

I hope so! Apparently the C# world has done this and they seem to be doing OK. Arguably it's how golang works too, if you squint. My guess is that Python will get there eventually. But... it certainly won't happen immediately.

There's a lot of code out there using callbacks, and for all its flaws, it's a very well-understood paradigm that has been used for lots of large, successful systems. There are also, I think, some very compelling arguments for getting rid of them – hence this essay – but the async/await-native paradigm is still fairly immature and will take some time to settle down and prove itself. My suspicion is that anything you can do with callbacks can be done better without them, but I can't prove it until someone tries.

This is another place where the goto/structured programming analogy is surprisingly apt. By 1968, the structured programming folks could show that all programs could be written without goto – but the construction was pretty ugly, and it left open the question of whether programs could be written elegantly without goto. And the answer was not obvious, especially considering that structured programming advocates of the time also liked to forbid the use of break, continue, and mid-function return on the grounds that they were too goto-like. (Exceptions were right out.)

Similarly, one could implement an asyncio event loop on top of curio to prove that the curio paradigm is at least as powerful in principle – but this wouldn't really answer the question, and it's entirely possible that current implementations like curio are missing some harmless quality-of-life measures like break. It's pretty exciting: we're on the cusp of learning things! But there are a lot of open questions about what exactly this future looks like.

So should I drop asyncio/twisted/etc. and rewrite everything using curio tomorrow?

Well... that's a complicated question.

If you want to start using the async/await-native approach today, then curio is currently the only game in town.

But even if you agree that this is where we want to end up eventually, there are still very good reasons why you might decide not to switch yet. Twisted and tornado are extremely mature, asyncio is in the standard library, and curio is neither of those things. All have seen years of intensive development by lots of very smart people and are in production use at companies you've heard of; curio is currently experimental alpha-status software by basically one guy. There's a much larger ecosystem of supporting libraries around twisted/tornado/asyncio than curio. And while the callback-based paradigm has its faults, those faults and their magnitude is well-understood with known workarounds, while the "curio paradigm" is still under heavy development, and curio-the-software doesn't yet make any promises about API stability.

On the other hand, if you find the async/await-native programming model compelling, and want to help flesh out a new paradigm, aren't reliant on the existing ecosystems (or are excited to help build a new one), and are comfortable with the risks, then you should totally go for it. Help us stride forward into a glorious Future-free future! Even if curio doesn't end up being the async/await-native API to end all APIs, we'll still learn something from the attempt.

For me personally, it helps that (a) the programs I'm working with are on the smaller side; there are no 20-person teams and big budgets depending on my technology choices; I could always port to something else if I had to. And (b) based on my experience so far in submitting patches to curio versus just trying to understand asyncio's edge case semantics, I'm honestly uncertain whether – if worst came to worst – it would be more work to maintain a personal fork of curio or to use upstream asyncio. Asyncio isn't bad; async/await is just so good that it radically changes the usual maturity calculus.

Should asyncio be "fixed" to have a curio-style async/await-native API?

In many ways this would be the ideal solution, since it would let us keep a single standard-bearer library with its interoperability story and more-developed ecosystem. And there are APIs in asyncio, like loop.sock_sendall, that seem like they could be first steps in this direction. But there are also some significant challenges:

• I can't see how this could be done without substantially throwing out and rewriting most of asyncio. Currently, transports are asyncio's fundamental abstraction layer: this is the layer that abstracts across different kinds of communication channels (e.g. sockets versus processes) and it also does a lot of the heavy lifting in abstracting across different underlying APIs (e.g. Unix-style polling versus Windows IOCP). The transport layer is also, as we saw above, the layer whose callback-centric abstractions cause so many problems for async/await code. And below the transport layer we have the event loop itself, where the mismatch isn't quite as great, but it still isn't the most natural fit. I'm not an asyncio developer, so I could be missing something... but the callback chaining parts are pretty deeply baked into asyncio as it currently exists.
• As mentioned above, the whole concept of a "curio-style API" is still undergoing heavy development, while asyncio's position in the stdlib makes it a poor place to experiment with new paradigms. This seems like the kind of thing where the ecosystem may be better off letting it stew in a 3rd-party lib for a while.
• Presumably the point of giving asyncio a curio-style API is that it would make it easier to mix together curio-style code and callback-style code in the same program (since I assume asyncio's current APIs wouldn't be going away – this would be supplemental). But – my whole argument in this essay is that for the most part, you don't want to mix these. So it's not clear what the point would be, really. If we're going to end up with two separate-but-equal APIs that only communicate at arms-length, then sticking them into two separate namespaces seems like it would be a lot less confusing.

So maybe this wouldn't be the best idea? I'll be very interested to see what the asyncio developers think.

One possible future would be: asyncio remains as the standard-bearer for the callback/hybrid approach – which is obviously going to remain viable and in use indefinitely – while eventually fading into becoming a legacy library as async/await-native approach matures and takes over for new code. (This is a not-unfamiliar trajectory for stdlib libraries – see urllib2.)

Okay, then should curio switch to using asyncio as a backend? Or what will the story be on cross-event-loop compatibility? I thought asyncio was supposed to be the event loop to end all event loops!

Indeed, one of the original, compelling motivations for adding asyncio to the stdlib is that it could become a standard foundation layer, so we could start up one event loop in one thread and then use it to simultaneously run libraries written for twisted, tornado, ... well, mostly twisted and tornado. And as a secondary benefit, we could swap in different backends to work on Unix vs. Windows, or on a headless server vs. embedded in a GUI app where the GUI framework imposes a particular event loop. Plus it's nice if different libraries can share code, so not everyone has to implement, say, IOCP handling from scratch.

Of course, this vision is rather predicated on the fact that until async/await came along, all these different event loops basically worked the same way, and the idea that there are lots of existing libraries that we want to use together.

It's not clear that curio could run nicely on top of asyncio – in particular, it seems difficult to reconcile their different ideas of how to manage the lifecycle of the event loop itself, and as we've seen, asyncio's higher-level abstractions are not useful to curio. They do already share code in the form of the selectors module (which is a great addition that asyncio brought to the stdlib!). Unfortunately, selectors isn't high-level enough to abstract over the differences between Unix and Windows, and as a result curio doesn't currently have great Windows support... but unfortunately there currently is no such abstraction layer that could be shared between curio and asyncio, because as mentioned above, asyncio's IOCP abstractions rely on the transport interfaces, and those are not useful to curio. It would be great if there were a library that abstracted over these platform differences at a lower level than asyncio does – maybe libuv could serve as an inspiration.

In addition, the main theme of this essay is that libraries written using async/await natively will be simpler and higher-quality than libraries written using callbacks, with or without an async/await layer added on top. So ideally we want to throw out those old libraries using the old APIs and replace them! This might be particularly true given the current interest in migrating to Sans I/O-style protocol libraries – which, in addition to their advantages in terms of design and maintainability, also make it much easier to migrate between incompatible I/O APIs, which makes direct interoperability less urgent.

...but of course, while the "throw out all the legacy code" strategy might work okay for green-field projects using popular protocols like HTTP, it doesn't help with that legacy tornado app, and it's probably going to be a while until we see an async/await-native client for NNTP or IMAP4. So it would be nice to have some kind of interopability story. One approach would be: start two OS threads. In one thread, run your async/await-native event loop; in the other thread, run your twisted reactor. Communicate by message passing. This approach is really crude... but, the programming models are different enough that message-passing might be what you want to use anyway. (I mean, curio doesn't even have a Deferred / Future concept, and for good reasons.) So even this crude approach might give you 90% of what you'd get by merging the underlying event loops, and with much less fuss?

This suggests that a library for cross-event-loop message passing might be an interesting short-term target for those who are worried about interoperability.

As for plugging in different backends, like for GUI framework interoperability: I'm not sure how that might work, and am not enough of a GUI programmer to have any useful insight into how async/await affect GUI programming. Definitely an interesting open question.