Python AsyncIO: Asynchronous IO

08-30-202008-30-2020 blog 16 minutes read (About 2473 words) visits

Introduction

Because breaking tasks into pieces and scheduling to run them asynchronously introduce runtime overheads, if the asynchronous program does not have too much I/O time to be saved from asyncio, the performance of executing those tasks asynchronously will be worse than executing those tasks synchronously. This means asynchronous I/O is the most important part of single-thread asynchronous application.

The question is how does asyncio save I/O time? A quick metaphor will be as follows. I have a candy factory and there are three production lines producing candies one by one. The candies from the production line will drop to the ground at the end of the production line, but dropping the candies to the ground will contaminate the candies. Unfortunately, I only got one worker to collect candies into the box. What’s his best strategy to collect as many candies as possible? Typically, he could have two strategies.

Collecting candies at one production line at one time.
Placing a box at the end of each production line. Rotating between the production lines from time to time and collecting the candies from the boxes.

Obviously, smart people will choose the second strategy. Basically, this first strategy is the single-thread synchronous application, and the second strategy is the single-thread asynchronous application.

In this blog post, I would like to discuss the asynchronous of IO for Python asynchronous programming using asyncio.

Asynchronous Sleep

A common I/O-bound task includes the sleep function. Using the synchronous and blocking time.sleep, we would have to wait and be blocked for whatever seconds we specified before it is returned.

In asyncio, we also have an asynchronous version of the sleep coroutine function, asyncio.sleep. By scheduling the coroutine to the event loop, during the sleep duration, we are not blocked from doing something else. Here is how asyncio.sleep is implemented.

async def sleep(delay, result=None, *, loop=None):
    """Coroutine that completes after a given time (in seconds)."""
    if delay <= 0:
        await __sleep0()
        return result

    if loop is None:
        loop = events.get_running_loop()
    else:
        warnings.warn("The loop argument is deprecated since Python 3.8, "
                      "and scheduled for removal in Python 3.10.",
                      DeprecationWarning, stacklevel=2)

    future = loop.create_future()
    h = loop.call_later(delay,
                        futures._set_result_unless_cancelled,
                        future, result)
    try:
        return await future
    finally:
        h.cancel()

Surprisingly, it is very simple. We create a Future and use a callback to set the value for this Future after some sleep duration delay. Once the result of Future got set, the sleep is over. During the the sleep duration delay, the event loop could schedule some other callbacks to do other tasks. It should be noted that such sleep duration is not precise in practice, due to the nature of event loop scheduling. For instance, if somehow the event loop scheduled a callback that takes extremely long to finish during the sleep duration, the actual sleep time would be much longer than what we were expecting. For example,

import asyncio
import time

async def block_func(block_duration: float = 5) -> None:

    # Blocking the event loop for 5 seconds
    time.sleep(block_duration)

async def sleep_func(sleep_duration: float = 1) -> None:

    start_time = time.time()
    await asyncio.sleep(delay=sleep_duration)
    end_time = time.time()
    elapsed_time = end_time - start_time

    print("Expected asynchronous sleep duration: {:.2f}s.".format(sleep_duration))
    print("Actual asynchronous sleep duration: {:.2f}s.".format(elapsed_time))

block_duration = 5
sleep_duration = 1

loop = asyncio.get_event_loop()

print("="*50)
print("Scheduled blocking first...")
print("-"*50)
loop.run_until_complete(
    asyncio.gather(
        block_func(block_duration),
        sleep_func(sleep_duration),
    )
)
print("="*50)
print("Scheduled sleeping first...")
print("-"*50)
loop.run_until_complete(
    asyncio.gather(
        sleep_func(sleep_duration),
        block_func(block_duration),
    )
)
print("="*50)

$ python long_sleep.py 
==================================================
Scheduled blocking first...
--------------------------------------------------
Expected asynchronous sleep duration: 1.00s.
Actual asynchronous sleep duration: 1.00s.
==================================================
Scheduled sleeping first...
--------------------------------------------------
Expected asynchronous sleep duration: 1.00s.
Actual asynchronous sleep duration: 5.00s.
==================================================

In this particular example, if we first schedule asynchronous sleep and then schedule a block execution, the finish of sleep has to wait until the the block execution finishes. This is also because the callback scheduling is sophisticated and not smart. Not sure if it is theoretically possible to have a smart event loop implementation.

Asynchronous Read

Although we have seen how sleep is scheduled asynchronously in the event loop, this does not help us understand how other IO-bounded tasks were scheduled very much. For example, if a server receives messages from multiple clients, how does it schedule the callbacks asynchronously?

This is a simple server that could asynchronously reading messages from and sending messages to multiple clients asynchronously.

import asyncio

async def handle_connection(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    addr = writer.get_extra_info("peername")
    while message := await reader.read(100):
        text = message.decode()
        print(f"Received {text!r} from {addr!r}")
        print(f"Sending {text!r}")
        writer.write(message)
        await writer.drain()
        if text == "quit\n":
            break
    
    print("Closing the connection")
    writer.close()

async def main() -> None:
    server = await asyncio.start_server(handle_connection, "127.0.0.1", 8888)
    addr = server.sockets[0].getsockname() if server.sockets else "unknown"
    print(f"Serving on {addr}")
    async with server:
        await server.serve_forever()

if __name__ == "__main__":
    
    asyncio.run(main())

The documentation of asyncio.start_server says “The client_connected_cb callback is called whenever a new client connection is established. It receives a (reader, writer) pair as two arguments, instances of the StreamReader and StreamWriter classes.” Therefore, the connection to the server is not blocking, allowing multiple clients connecting to the sever. If the connection were synchronous and blocking, only one connection could be established.

The data reading and writing to the streams have to be asynchronous as well. Different clients use different sockets in the same port. Each file from the client sent to the server socket will have an EOF (end-of-file), which is an indication of the end of transfer. If the data reading and writing is synchronous, the network connection between the server and one client is poor, even if the file being transferred is small, it might take a long time to receive the EOF, thus blocking the entire thread.

Using the asynchronous StreamReader.read method, we could switch between the multiple connections to the clients. Let’s check how it is achieved.

When we start an server, a StreamReader will be created and a StreamReaderProtocol will wrap the the StreamReader.

async def start_server(client_connected_cb, host=None, port=None, *,
                       loop=None, limit=_DEFAULT_LIMIT, **kwds):
    """Start a socket server, call back for each client connected.
    The first parameter, `client_connected_cb`, takes two parameters:
    client_reader, client_writer.  client_reader is a StreamReader
    object, while client_writer is a StreamWriter object.  This
    parameter can either be a plain callback function or a coroutine;
    if it is a coroutine, it will be automatically converted into a
    Task.
    The rest of the arguments are all the usual arguments to
    loop.create_server() except protocol_factory; most common are
    positional host and port, with various optional keyword arguments
    following.  The return value is the same as loop.create_server().
    Additional optional keyword arguments are loop (to set the event loop
    instance to use) and limit (to set the buffer limit passed to the
    StreamReader).
    The return value is the same as loop.create_server(), i.e. a
    Server object which can be used to stop the service.
    """
    if loop is None:
        loop = events.get_event_loop()
    else:
        warnings.warn("The loop argument is deprecated since Python 3.8, "
                      "and scheduled for removal in Python 3.10.",
                      DeprecationWarning, stacklevel=2)

    def factory():
        reader = StreamReader(limit=limit, loop=loop)
        protocol = StreamReaderProtocol(reader, client_connected_cb,
                                        loop=loop)
        return protocol

    return await loop.create_server(factory, host, port, **kwds)

The StreamReader.read method is implemented as follows. What it does basically is wait until the some data have been received via self._wait_for_data and the return at most n bytes of the data.

async def read(self, n=-1):
    """Read up to `n` bytes from the stream.
    If n is not provided, or set to -1, read until EOF and return all read
    bytes. If the EOF was received and the internal buffer is empty, return
    an empty bytes object.
    If n is zero, return empty bytes object immediately.
    If n is positive, this function try to read `n` bytes, and may return
    less or equal bytes than requested, but at least one byte. If EOF was
    received before any byte is read, this function returns empty byte
    object.
    Returned value is not limited with limit, configured at stream
    creation.
    If stream was paused, this function will automatically resume it if
    needed.
    """

    if self._exception is not None:
        raise self._exception

    if n == 0:
        return b''

    if n < 0:
        # This used to just loop creating a new waiter hoping to
        # collect everything in self._buffer, but that would
        # deadlock if the subprocess sends more than self.limit
        # bytes.  So just call self.read(self._limit) until EOF.
        blocks = []
        while True:
            block = await self.read(self._limit)
            if not block:
                break
            blocks.append(block)
        return b''.join(blocks)

    if not self._buffer and not self._eof:
        await self._wait_for_data('read')

    # This will work right even if buffer is less than n bytes
    data = bytes(self._buffer[:n])
    del self._buffer[:n]

    self._maybe_resume_transport()
    return data

The StreamReader._wait_for_data coroutine is basically waiting for a Future self._waiter to be set.

async def _wait_for_data(self, func_name):
    """Wait until feed_data() or feed_eof() is called.
    If stream was paused, automatically resume it.
    """
    # StreamReader uses a future to link the protocol feed_data() method
    # to a read coroutine. Running two read coroutines at the same time
    # would have an unexpected behaviour. It would not possible to know
    # which coroutine would get the next data.
    if self._waiter is not None:
        raise RuntimeError(
            f'{func_name}() called while another coroutine is '
            f'already waiting for incoming data')

    assert not self._eof, '_wait_for_data after EOF'

    # Waiting for data while paused will make deadlock, so prevent it.
    # This is essential for readexactly(n) for case when n > self._limit.
    if self._paused:
        self._paused = False
        self._transport.resume_reading()

    self._waiter = self._loop.create_future()
    try:
        await self._waiter
    finally:
        self._waiter = None

So who is going to set the value for self._waiter? It is StreamReader.feed_data which calls StreamReader._wakeup_waiter to set the value for self._waiter.

def feed_data(self, data):
    assert not self._eof, 'feed_data after feed_eof'

    if not data:
        return

    self._buffer.extend(data)
    self._wakeup_waiter()

    if (self._transport is not None and
            not self._paused and
            len(self._buffer) > 2 * self._limit):
        try:
            self._transport.pause_reading()
        except NotImplementedError:
            # The transport can't be paused.
            # We'll just have to buffer all data.
            # Forget the transport so we don't keep trying.
            self._transport = None
        else:
            self._paused = True

def _wakeup_waiter(self):
    """Wakeup read*() functions waiting for data or EOF."""
    waiter = self._waiter
    if waiter is not None:
        self._waiter = None
        if not waiter.cancelled():
            waiter.set_result(None)

Before the value of self._waiter is set, the data has been extended to self._buffer so that StreamReader.read could safely read the buffer.

So the final question is who will call StreamReader.feed_data? Remember the StreamReader is wrapped in a StreamReaderProtocol.

class StreamReaderProtocol(FlowControlMixin, protocols.Protocol):
    """Helper class to adapt between Protocol and StreamReader.
    (This is a helper class instead of making StreamReader itself a
    Protocol subclass, because the StreamReader has other potential
    uses, and to prevent the user of the StreamReader to accidentally
    call inappropriate methods of the protocol.)
    """

    _source_traceback = None

    def __init__(self, stream_reader, client_connected_cb=None, loop=None):
        super().__init__(loop=loop)
        if stream_reader is not None:
            self._stream_reader_wr = weakref.ref(stream_reader)
            self._source_traceback = stream_reader._source_traceback
        else:
            self._stream_reader_wr = None
        if client_connected_cb is not None:
            # This is a stream created by the `create_server()` function.
            # Keep a strong reference to the reader until a connection
            # is established.
            self._strong_reader = stream_reader
        self._reject_connection = False
        self._stream_writer = None
        self._transport = None
        self._client_connected_cb = client_connected_cb
        self._over_ssl = False
        self._closed = self._loop.create_future()

When StreamReaderProtocol.data_received is called, it actually calls StreamReader.feed_data.

def data_received(self, data):
    reader = self._stream_reader
    if reader is not None:
        reader.feed_data(data)

But who is calling StreamReaderProtocol.data_received? According to the base class protocols.Protocol for StreamReaderProtocol.

class Protocol(BaseProtocol):
    """Interface for stream protocol.
    The user should implement this interface.  They can inherit from
    this class but don't need to.  The implementations here do
    nothing (they don't raise exceptions).
    When the user wants to requests a transport, they pass a protocol
    factory to a utility function (e.g., EventLoop.create_connection()).
    When the connection is made successfully, connection_made() is
    called with a suitable transport object.  Then data_received()
    will be called 0 or more times with data (bytes) received from the
    transport; finally, connection_lost() will be called exactly once
    with either an exception object or None as an argument.
    State machine of calls:
      start -> CM [-> DR*] [-> ER?] -> CL -> end
    * CM: connection_made()
    * DR: data_received()
    * ER: eof_received()
    * CL: connection_lost()
    """

Essentially the state machine is

start -> connection_made
    [-> data_received]*
    [-> eof_received]?
-> connection_lost -> end

The data got received for one or more than one times, it could be less, equal, or larger than the number of bytes the user wanted. Once the user want to wait for the data via StreamReader.read, most likely the most recent data collected from StreamReaderProtocol.data_received will be the data returned. The StreamReaderProtocol.data_received is called at certain rate and the buffer got extended accordingly, it is how asynchronous read becomes possible.

Conclusions

The fundamentals of asynchronous IO in asynchronous programming are sometimes trivial if we understand the system we are working on very well. However, it could still be complicated to implement at the low-level.