asyncio coroutines

As mentioned in “Coroutine-Based Async Architectures”, the core concept used in asyncio is that of a coroutine—a function able to suspend its execution, preserving its internal state; explicitly hand control back to the framework that’s orchestrating it (in asyncio’s case, that’s the event loop, covered in “asyncio’s Event Loop”); and resume execution from the point at which it had suspended, when the framework tells it to resume.

Just as a generator function immediately executes, runs no user code, and returns a generator object, a coroutine function immediately executes, runs no user code, and returns a coroutine object. In fact, up to Python 3.4, a coroutine function was just a specific kind of generator function: a generator function is any function containing one or more occurrences of the yield keyword, and a coroutine function was one specifically containing one or more occurrences of the expression yield from iterable…although such a statement could also occur in a generator function not intended for use as a coroutine function, as covered in “yield from (v3-only)” (the resulting ambiguity, although marginal, was suboptimal, whence the introduction in Python 3.5 of async and await, which remove any ambiguity). To mark such a yield from–based coroutine function explicitly as being a coroutine, and ensure it was usable in asyncio, you decorated it with @asyncio.coroutine. For example, here’s a coroutine that sleeps for delay seconds, then returns a result:

@asyncio.coroutine
def delayed_result(delay, result):
  yield from asyncio.sleep(delay)
  return result

Calling, for example, delayed_result(1.5, 23), gives you a generator object (specifically a coroutine object) poised to wait a second and a half, then return 23. Note that the generator object in question won’t execute until it’s properly connected to an event loop, covered in “asyncio’s Event Loop”, and the event loop runs; for example,

loop = asyncio.get_event_loop()
x = loop.run_until_complete(delayed_result(1.5, 23))

sets x to 23 after a delay of a second and a half.

Mostly for debugging purposes, you can check whether some object f is a coroutine function with asyncio.iscoroutinefunction(f); you can check whether some object x is a coroutine object with asyncio.iscoroutine(x).

async and await

To allow more explicit and readable code, Python 3.5 introduces the keywords async and await.¹ Use async to create coroutines (with async def), to create asynchronous context managers (with async with), and for asynchronous iteration (with async for).

When you use async def instead of def, and await instead of yield from,² the function becomes an explicit coroutine function, with no need for decoration (and without being a generator function as a side effect). For example:

async def new_way(delay, result):
  await asyncio.sleep(delay)
  return result

gives you a coroutine function new_way that’s fully equivalent to the delayed_result in the previous example (you could decorate new_way with @asyncio.coroutine, but this is not required nor recommended usage). In this way, you can also write a coroutine function that never awaits anything. (It may seem peculiar to need that, but it comes in handy in prototyping, testing, and refactoring.) An async def coroutine function cannot contain any yield; conversely, you can use await only within an async def coroutine function.

Within a native coroutine function (meaning one defined with async def) only, you can also use two useful new constructs: async with and async for.

Loop state and debugging mode

loop can be in one of three states: stopped (that’s the state loop is when just created: nothing yet runs on loop), running (all functionality runs on loop), or closed (loop is irreversibly terminated, and cannot be started again). Independently, loop can be in debug mode (checking sanity of operations, and giving ample information to help you develop code) or not (faster and quieter operation; that’s the normal mode to use “in production,” as opposed to development, debugging, and testing), as covered in “asyncio developing and debugging”. Regarding state and mode, loop supplies the following methods:

Calls, time, and sleep

A key functionality of the asyncio event loop is to schedule calls to functions (see Table 18-1)—either “as soon as convenient” or with specified delays. For the latter purpose, loop maintains its own internal clock (in seconds and fractions), not necessarily coincident with the system clock covered in Chapter 12.

Besides loop’s own methods, the current event loop’s internal time is also used by one module-level function (i.e., a function supplied directly by the module asyncio):

Tasks

An asyncio task (asyncio.Task, covered in “Tasks”, is a subclass of asyncio.Future, covered in “Futures”) wraps a coroutine object and orchestrates its execution. loop offers a method to create a task:

You can also customize the factory loop uses to create tasks, but that is rarely needed except to write custom implementations of event loops, so we don’t cover it.

Another roughly equivalent way to create a task is to call asyncio.ensure_future with a single argument that is a coroutine object or other awaitable; the function in this case creates and returns a Task instance wrapping the coroutine object. (If you call asyncio.ensure_future with an argument that’s a Future instance, it returns the argument unchanged.)

Watching file descriptors

You can choose to use loop at a somewhat-low abstraction level, watching for file descriptors to become ready for reading or writing, and calling callback functions when they do. (On Windows, with the default SelectorEventLoop implementation of loop, you can use these methods only on file descriptors representing sockets; with the alternative ProactorEventLoop implementation that you can choose to explicitly instantiate, you cannot use these methods at all.) loop supplies the following methods related to watching file descriptors:

socket operations; hostnames

Also at a low level of abstraction, loop supplies four coroutine-function methods corresponding to methods on socket objects covered in Chapter 17:

It’s often necessary to perform DNS lookups. loop supplies two coroutine-function methods that work like the same-name functions covered in Table 17-1, but in an async, nonblocking way:

Unix signals

On Unix-like platforms, loop (when run on the main thread) supplies two methods to add and remove handlers for signals (a Unix-specific, limited form of inter-process communication, well covered on Wikipedia) the process may receive:

Executor

loop can arrange for a function to run in an executor, a pool of threads or processes as covered in “The concurrent.futures Module”: that’s useful when you must do some blocking I/O, or CPU-intensive operations (in the latter case, use as executor an instance of concurrent.futures.ProcessPoolExecutor). The two relevant methods are:

Error handling

You can customize exception handling in the event loop; loop supplies four methods for this purpose:

context is a dict with the following contents (more keys may be added in future releases). All keys, except message, are optional; use context.get(key) to avoid a KeyError on accessing some key in the context:

exception

Exception instance

future

asyncio.Future instance

handle

asyncio.Handle instance

message

str instance, the error message

protocol

asyncio.Protocol instance

socket

socket.socket instance

transport

asyncio.Transport instance

The following sections cover three more concepts you need in order to use asyncio, and functionality supplied by asyncio for each of these concepts.

Futures

The asyncio.Future class is almost compatible with the Future class supplied by the module concurrent.futures and covered in Table 14-1 (in some future version of Python, the intention is to fully unify the two Future interfaces, but this goal cannot be guaranteed). The main differences between an instance af of asyncio.Future and an instance cf of concurrent.futures.Future are:

• af is not thread-safe

• af can’t be passed to functions wait and as_completed of module concurrent.futures

• Methods af.result and af.exception don’t take a timeout argument, and can only be called when af.done() is True

• There is no method af.running()

For thread-safety, callbacks added with af.add_done_callback get scheduled, once af is done, via loop.call_soon_threadsafe.

af also supplies three extra methods over and above those of cf:

(In fact, cf has methods set_exception and set_result, too, but in cf’s case they’re meant to be called strictly and exclusively by unit tests and Executor implementations; af’s identical methods do not have such constraints.)

The best way to create a Future in asyncio is with loop’s create_future method, which takes no arguments; at worst, loop.create_future() just performs exactly the same as return futures.Future(loop), but, this way, alternative loop implementations get a chance to override the method and provide a better implementation of futures.

Tasks

asyncio.Task is a subclass of asyncio.Future: an instance at of asyncio.Task wraps a coroutine object and schedules its execution in loop.

The class defines two class methods: all_tasks(), which returns the set of all tasks defined on loop; and current_task(), which returns the task currently executing (None if no task is executing).

at.cancel() has slightly different semantics from the cancel method of other futures: it does not guarantee the cancellation of the task, but rather raises a CancelledError inside the wrapped coroutine—the latter may intercept the exception (intended to enable clean-up work, but also makes it possible for the coroutine to refuse cancellation). at.cancelled() returns True only when the wrapped coroutine has propagated (or spontaneously raised) CancelledError.

asyncio supplies several functions to ease working with tasks and other futures. All accept an optional (named-only) argument loop=None to use an event loop different from the current default one (None means to use the current default event loop); for all, arguments that are futures can also be coroutine objects (which get automatically wrapped into instances of asyncio.Task). The functions are:

try:
    with asyncio.timeout(0.5):
        await first()
        await second()
except asyncio.TimeoutError:
    print('Alas, too slow!')
else:
    print('Made it!')

Transports and protocols

For details about transports and protocols, see the section about them in the online docs. In this section, we’re offering just the conceptual basis, some core details about working with them, and two examples. The core idea is that a transport does all that’s needed to ensure that a stream (or datagram) of “raw,” uninterpreted bytes is pushed to an external system, or pulled from an external system; a protocol translates those bytes to and from semantically meaningful messages.

A transport class is a class supplied by asyncio to abstract any one of various kinds of communication channels (TCP, UDP, SSL, pipes, etc.). You don’t directly instantiate a transport class: rather, you call loop methods that create the transport instance and the underlying channel, and provide the transport instance when done.

A protocol class is one supplied by asyncio to abstract various kinds of protocols (streaming, datagram-based, subprocess-pipes). Extend the appropriate one of those base classes, overriding the callback methods in which you want to perform some action (the base classes supply empty default implementations for such methods, so just don’t override methods for events you don’t care about). Then, pass your class as the protocol_factory argument to loop methods.

A protocol instance p always has an associated transport instance t, in 1-to-1 correspondence. As soon as the connection is established, loop calls p.connection_made(t): p must save t as an attribute of self, and may perform some initialization-setting method calls on t.

When the connection is lost or closed, loop calls p.connection_lost(exc), where exc is None to indicate a regular closing (typically via end-of-file, EOF), or else an Exception instance recording what error caused the connection to be lost.

Each of connection_made and connection_lost gets called exactly once on each protocol instance p. All other callbacks to p’s methods happen between those two calls; during such other callbacks, p gets informed by t about data or EOF being received, and/or asks t to send data out. All interactions between p and t occur via callbacks by each other on the other one’s methods.

import asyncio

data = """A few lines of text
including non-ASCII characters: €£
to test the operation
of both server
and client."""

class EchoClient(asyncio.Protocol):

  def __init__(self):
      self.data_iter = iter(data.splitlines())

  def write_one(self):
      chunk = next(self.data_iter, None)
      if chunk is None:
          self.transport.write_eof()
      else:
          line = chunk.encode()
          self.transport.write(line)
          print('Sent:', chunk)

  def connection_made(self, transport):
      self.transport = transport
      print('Connected to server')
      self.write_one()

  def connection_lost(self, exc):
      loop.stop()
      print('Disconnected from server')

  def data_received(self, data):
      print('Recv:', data.decode())
      self.write_one()

loop = asyncio.get_event_loop()
echo = loop.create_connection(EchoClient, 'localhost', 8881)
transport, protocol = loop.run_until_complete(echo)
loop.run_forever()
loop.close()

import asyncio

class EchoServer(asyncio.Protocol):

  def connection_made(self, transport):
      self.transport = transport
      self.peer = transport.get_extra_info('peername')
      print('Connected from', self.peer)

  def connection_lost(self, exc):
      print('Disconnected from', self.peer)

  def data_received(self, data):
      print('Recv:', data.decode())
      self.transport.write(data)
      print('Echo:', data.decode())

loop = asyncio.get_event_loop()
echo = loop.create_server(EchoServer, 'localhost', 8881)
server = loop.run_until_complete(echo)
print('Serving at', server.sockets[0].getsockname())
loop.run_forever()

import asyncio

data = """A few lines of data
including non-ASCII characters: €£
to test the operation
of both server
and client."""

@asyncio.coroutine
def echo_client(data):
    reader, writer = yield from asyncio.open_connection(
        'localhost', 8881)
    print('Connected to server')
    for line in data:
        writer.write(line.encode())
        print('Sent:', line)
        response = yield from reader.read(1024)
        print('Recv:', response.decode())
    writer.close()
    print('Disconnected from server')

loop = asyncio.get_event_loop()
loop.run_until_complete(echo_client(data.splitlines()))
loop.close()

import asyncio

data = """A few lines of data
including non-ASCII characters: €£
to test the operation
of both server
and client."""

async def echo_client(data):
    reader, writer = await asyncio.open_connection(
        'localhost', 8881)
    print('Connected to server')
    for line in data:
        writer.write(line.encode())
        print('Sent:', line)
        response = await reader.read(1024)
        print('Recv:', response.decode())
    writer.close()
    print('Disconnected from server')

loop = asyncio.get_event_loop()
loop.run_until_complete(echo_client(data.splitlines()))
loop.close()

import asyncio

@asyncio.coroutine
def handle(reader, writer):
    address = writer.get_extra_info('peername')
    print('Connected from', address)

    while True:
        data = yield from reader.read(1024)
        if not data: break
        s = data.decode()
        print('Recv:', s)
        writer.write(data)
        yield from writer.drain()
        print('Echo:', s)

    writer.close()
    print('Disconnected from', address)

loop = asyncio.get_event_loop()
echo = asyncio.start_server(handle, 'localhost', 8881)
server = loop.run_until_complete(echo)

print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

import asyncio

async def handle(reader, writer):
    address = writer.get_extra_info('peername')
    print('Connected from', address)

    while True:
        data = await reader.read(1024)
        if not data: break
        s = data.decode()
        print('Recv:', s)
        writer.write(data)
        await writer.drain()
        print('Echo:', s)

    writer.close()
    print('Disconnected from', address)

loop = asyncio.get_event_loop()
echo = asyncio.start_server(handle, 'localhost', 8881)
server = loop.run_until_complete(echo)

print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

Asyncio synchronization primitives

asyncio supplies the classes BoundedSemaphore, Condition, Event, Loop, and Semaphore, very similar to the classes of the same names covered in “Thread Synchronization Objects”, except of course that the classes in asyncio synchronize coroutines, not threads. The asyncio synchronization classes and their methods do not accept a timeout argument (use asyncio.wait_for instead to apply a timeout to a coroutine or asyncio future).

Several methods of asyncio synchronization classes (all that deal with acquiring or waiting for a synchronization class’s instance, and therefore could possibly block) are coroutine methods returning coroutine objects. Specifically, this applies to BoundedSemaphore.acquire, Condition.acquire, Condition.wait, Condi⁠tion.wait_​for, Event.wait, Loop.acquire, and Semaphore.acquire.

Asyncio queues

asyncio supplies the class Queue, and its subclasses LifoQueue and PriorityQueue, very similar to the classes of the same names covered in “The queue Module”, except of course that the classes in asyncio queue data between coroutines, not between threads. The asyncio queue classes and their methods do not accept a timeout argument (use asyncio.wait_for instead to apply a timeout to a coroutine or asyncio future).

Several methods of asyncio queue classes (all that could possibly block) are coroutine methods, returning coroutine objects. Specifically, this applies to the methods get, join, and put, of the class Queue and of both subclasses.

selectors echo server example

Here is a selectors-based example server for the simplistic echo protocol shown in “A Connection-Oriented Socket Client”. This example is fully coded to run equally well in v2 (as long as you’ve installed the selectors34 backport; see “The concurrent.futures Module”) and v3; it provides more informational prints than example “A Connection-Oriented Socket Server”, to better show the low-level mechanisms involved in communication with multiple clients “at once,” in byte chunks of whatever size optimizes all I/O operations:

from __future__ import print_function
import socket
try:
    import selectors
except ImportError:
    import selectors34 as selectors
ALL = selectors.EVENT_READ | selectors.EVENT_WRITE

# mapping socket -> bytes to send on that socket when feasible
data = {}

# set of sockets to be closed and unregistered when feasible
socks_to_close = set()

def accept(sel, data, servsock, unused_events):
    # accept a new connection, register it with the selector
    sock, address = servsock.accept()
    print('Connected from', address)
    sel.register(sock, ALL, handle)

def finish(sel, data, sock):
    # sock needs to be closed & unregistered if it's in the selector
    if sock in sel.get_map():
        ad = sock.getpeername()
        print('Disconnected from {}'.format(ad))
        sel.unregister(sock)
        sock.close()
    data.pop(sock, None)

def handle(sel, data, sock, events):
    # a client socket just became ready to write and/or read
    ad = sock.getpeername()
    if events & selectors.EVENT_WRITE:
        tosend = data.get(sock)
        if tosend:
            # send as much as the client's ready to receive
            nsent = sock.send(tosend)
            s = tosend.decode('utf-8', errors='replace')
            print(u'{} bytes sent to {} ({})'.format(nsent, ad, s))
            # bytes to be sent later, if any
            tosend = tosend[nsent:]
        if tosend:
            # more bytes to send later, keep track of them
            print('{} bytes remain for {}'.format(len(tosend), ad))
            data[sock] = tosend
        else:
            # no more bytes to send -> ignore EVENT_WRITE for now
            print('No bytes remain for {}'.format(ad))
            data.pop(sock, None)
            sel.modify(sock, selectors.EVENT_READ, handle)
    if events & selectors.EVENT_READ:
        # receive up to 1024 bytes at a time
        newdata = sock.recv(1024)
        if newdata:
            # add new data received as needing to be sent back
            s = newdata.decode('utf-8', errors='replace')
            print(u'Got {} bytes from {} ({})'.format(
                  len(newdata), ad, s))
            data[sock] = data.get(sock, b'') + newdata
            sel.modify(sock, ALL, handle)
        else:
            # recv empty string means client disconnected
            socks_to_close.add(sock)

try:
    servsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    servsock.bind(('', 8881))
    servsock.listen(5)
    print('Serving at', servsock.getsockname())
    with selectors.DefaultSelector() as sel:
        sel.register(servsock, selectors.EVENT_READ, accept)
        while True:
            for sk, events in sel.select():
                sk.data(sel, data, sk.fileobj, events)
            while socks_to_close:
                sock = socks_to_close.pop()
                finish(sel, data, sock)
except KeyboardInterrupt:
    pass
finally:
    servsock.close()

selectors does not supply an event loop; the event loop needed in async programming, in the preceding example, is the simple loop using while True:.

The low-level approach, as shown in this example, makes us write a lot of code compared to the simplicity of the task, a task that is made especially simple by the fact that the server does not need to examine or process incoming bytes (for most network jobs, incoming streams or chunks of bytes do need to be parsed into semantically meaningful messages). The payback comes in the high performance one can attain, with such a no-overhead approach, consuming minimal resources.

Moreover, it’s worth noticing (as is emphasized by the detailed informational prints) that this example does not share the limitations of “A Connection-Oriented Socket Server”: the code, as presented, can handle any number of clients, up to the limits of available memory, as well as arbitrary network fragmentation of data along the arbitrarily high number of simultaneous connections.