A Simple Event Loop

In most cases, you do not need to create an event loop object yourself. You can get a BaseEventLoop object by using the asyncio.get_event_loop() function. What you will actually get will be a subclass; which subclass you get is platform-dependent. You do not need to worry about this implementation detail too much. The API between all of them is the same. However, a few platform-dependent limitations exist.

When you first get the loop object, it will not be running.

>>> loop = asyncio.get_event_loop()
>>> loop.is_running()
False

>>> loop.run_forever()

>>> import functools
>>> def hello_world():
...     print('Hello world!')
...
>>> def stop_loop(loop):
...     print('Stopping loop.')
...     loop.stop()
...
>>> loop.call_soon(hello_world)
Handle(<function hello_world at 0x1003c0b70>, ())
>>> loop.call_soon(functools.partial(stop_loop, loop))
Handle(functools.partial(<function stop_loop at 0x101ccf268>,
     <asyncio.unix_events._UnixSelectorEventLoop
     object at 0x1007399e8>), ())
>>> loop.run_forever()
Hello world!
Stopping loop.
>>> 

>>> loop.call_later(10, hello_world)
TimerHandle(60172.411042585, <function hello_world at 0x1003c0b70>, ())
>>> loop.call_later(20, functools.partial(stop_loop, loop))
TimerHandle(60194.829461844, functools.partial(
     <function stop_loop at 0x101ccf268>,
     <asyncio.unix_events._UnixSelectorEventLoop object at 0x1007399e8>),
     ())
>>> loop.run_forever()

>>> import functools
>>> def stop_loop(loop):
...     print('Stopping loop.')
...     loop.stop()
...
>>> loop.call_soon(functools.partial(print, 'Hello world! ')
Handle(functools.partial(<built-in function print>, 'Hello world'), ())
>>> loop.call_soon(functools.partial(stop_loop, loop))
Handle(functools.partial(<function stop_loop at 0x101ccf268>,
     <asyncio.unix_events._UnixSelectorEventLoop object
     at 0x1007399e8>), ())
>>> loop.run_forever()
Hello world!
Stopping loop.
>>> 

>>> partial = functools.partial(stop_loop, loop)
>>> partial.func
<function stop_loop at 0x10223e488>
>>> partial.args
(<asyncio.unix_events._UnixSelectorEventLoop object at 0x102238b70>,)

>>> @asyncio.coroutine
... def trivial():
...     return 'Hello world!'
...
>>> loop.run_until_complete(trivial())
'Hello world!'

>>> import asyncio
>>> import threading
>>>
>>> def run_loop_forever_in_background(loop):
...     def thread_func(l):
...         asyncio.set_event_loop(l)
...         l.run_forever()
...     thread = threading.Thread(target=thread_func, args=(loop,))
...     thread.start()
...     return thread
...
>>>
>>> loop = asyncio.get_event_loop()
>>> run_loop_forever_in_background(loop)
<Thread(Thread-1, started 4344254464)>
>>>
>>> loop.is_running()
True

>>> loop.call_soon_threadsafe(functools.partial(print, 'Hello world'))
Handle(functools.partial(<built-in function print>, 'Hello world'), ())
>>> Hello world

Nested Coroutines

Coroutines provide a special mechanism to call other coroutines (or Future instances, as discussed shortly) in a fashion that mimics that of sequential programming. By using the yield from statement, a coroutine can run another coroutine, and the statement returns the result. This is one mechanism available to write asynchronous code in a sequential manner.

The following simple coroutine calls another coroutine using yield from:

>>> import asyncio
>>> @asyncio.coroutine
... def nested(*args):
...     print('The 'nested' function ran with args: %r' % (args,))
...     return [i + 1 for i in args]
...
>>> @asyncio.coroutine
... def outer(*args):
...     print('The 'outer' function ran with args: %r' % (args,))
...     answer = yield from nested(*[i * 2 for i in args])
...     return answer
...
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(outer(2, 3, 5, 8))
The 'outer' function ran with args: (2, 3, 5, 8)
The 'nested' function ran with args: (4, 6, 10, 16)
[5, 7, 11, 17]

Here you have two coroutines, with the outer coroutine calling the nested coroutine using the yield from syntax. You can see from the output to standard out that both coroutines run, and the final result is returned at the end of outer.

Incidentally, what is happening here under the hood is that the outer coroutine is actually suspended when it encounters the yield from statement. The nested coroutine is then placed on the event loop and the event loop runs it. The outer coroutine does not continue until nested completes and a result is available.

A couple things are worth noting. First, the yield from statement returns the result of the coroutine it runs. That is why you see an assignment to a variable in the example.

Second, why would you not simply call the function directly? This would be fine if it were a procedural function, but this is a coroutine. Calling it directly would return a generator rather than the value. You could write nested as a standard function, but consider the following situation where you would also want to be able to assign it to the event loop directly.

>>> loop.run_until_complete(nested(5, 10, 15))
The 'nested' function ran with args: (5, 10, 15)
[6, 11, 16]

The capability to have a coroutine call another coroutine using yield from addresses this. It increases the capability to reuse coroutines.

Futures

A mechanism for dealing with this particular challenge is the Future object. Essentially, a Future is an object that is told about the status of an asynchronous function. This includes the status of the function—whether that function is running, has completed, or was canceled. This also includes the result of the function, or, if the function ended by raising an exception, the exception and traceback.

The Future is a standalone object. It is independent of the actual function that is running. It does nothing but store the state and result information.

Tasks

A Task is a subclass of Future, as well as what you will generally be using when programming with asyncio. Whenever a coroutine is scheduled on the event loop, that coroutine is wrapped in a Task. So, in the previous example, when you called run_until_complete and passed a coroutine, that coroutine was wrapped in a Task class and then executed. It was the Task that stored the result and handled providing it in the yield from statement.

The run_until_complete method is not the only way (or even the primary way) for a coroutine to be wrapped in a class, however. After all, in many applications, your event loop runs forever. How do tasks get placed on the event loop in such a system?

The primary way you do this is by using the asyncio.async method. This method will place a coroutine on the event loop, and return the associated Task.

To demonstrate this, first get the event loop and write a garden-variety coroutine, as shown here:

>>> import asyncio
>>>
>>> @asyncio.coroutine
... def make_tea(variety):
...     print('Now making %s tea.' % variety)
...     asyncio.get_event_loop().stop()
...     return '%s tea' % variety
...
>>>

This is still a trivial task, but one new thing here that you have not seen yet is that the task actually stops the event loop. This is simply a nice workaround to dodge the fact that when you start the loop (with run_forever), it will run forever.

Next, register the task with the event loop.

>>> task = asyncio.async(make_tea('chamomile'))

This is all you actually need to do to register the task with the loop, but because the loop is not running, the task is not going to execute for now. Indeed, you can inspect the task object using the done and result methods and see this.

>>> task.done()
False
>>> task.result()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/
       futures.py", line 237, in result
    raise InvalidStateError('Result is not ready.')
asyncio.futures.InvalidStateError: Result is not ready.

Next, you must start the loop. It is okay to start the loop with run_forever now; the actual task will stop it as soon as the task completes because of the call to loop.stop().

>>> loop = asyncio.get_event_loop()
>>> loop.run_forever()
Now making chamomile tea.
>>>

Sure enough, the loop starts, runs the task, and then immediately stops. Now if you inspect the task variable, you will get different results.

>>> task.done()
True
>>> task.result()
'chamomile tea'

Whenever you create a Task object with asyncio.async, you will get a Task object back. You can inspect that object at any time to get the status or result of the task.

No Guarantee of Success

There is one important thing to note. Simply because a Future is done does not guarantee that it ran successfully. This example simply assumes that future.result() will be populated, but that may not be the case. The Task could have ended in an exception, in which case, attempting to access future.result() will raise that exception.

Similarly, it is possible to cancel a task (using the Future.cancel() method or by other means). If this occurs, the task will be marked Cancelled, and the callbacks will be scheduled. In this case, attempting to access future.result() will raise CancelledError.

Under the Hood

Internally, asyncio informs the Future object that it is done. The Future object then takes each of the callbacks registered against it and calls call_soon_threadsafe on each of them.

Be aware that there is no guarantee of order when it comes to callbacks. It is entirely possible (and fine) to register multiple callbacks to the same task. However, you do not have any way of controlling which callbacks will be run in which order.

Callbacks with Arguments

One limitation of the callback system is that, as noted, the callback receives the Future as a positional argument, and accepts no other arguments.

It is possible to send other arguments to a callback through the use of functools.partial. If you do this, however, the callback must still accept the Future as a positional argument. In practice, the Future is appended to the end of the positional arguments list before the callback is called.

Consider the following case of a callback that expects another argument:

>>> import asyncio
>>> import functools
>>>
>>> loop = asyncio.get_event_loop()
>>>
>>> @asyncio.coroutine
... def make_tea(variety):
...     print('Now making %s tea.' % variety)
...     return '%s tea' % variety
...
>>> def add_ingredient(ingredient, future):
...     print('Now adding %s to the %s.' % (ingredient, future.result()))
...
>>>
>>> task = asyncio.async(make_tea('herbal'))
>>> task.add_done_callback(functools.partial(add_ingredient, 'honey'))
>>>
>>> loop.run_until_complete(task)
Now making herbal tea.
Now adding honey to the herbal tea.
'herbal tea'

This is mostly similar to the previous example. The only significant difference is in how the callback is registered. Instead of passing the function object directly (as you did in the previous example), you instantiate a functools.partial object with the positional argument you are sending ('honey').

Again, notice that the add_ingredient function is written to accept two positional arguments, but the partial only specifies one argument. The Future object is sent as the last positional argument in cases where a partial is used. The function signature for add_ingredient reflects this.

Gathering Tasks

The first mechanism that asyncio provides for this purpose is the gather function. The gather function takes a sequence of coroutines or tasks and returns a single task that aggregates all of them (wrapping any coroutines it receives in tasks as appropriate).

>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>>
>>> @asyncio.coroutine
... def make_tea(variety):
...     print('Now making %s tea.' % variety)
...     return '%s tea' % variety
...
>>> meta_task = asyncio.gather(
...     make_tea('chamomile'),
...     make_tea('green'),
...     make_tea('herbal')
... )
...
>>> meta_task.done()
False
>>>
>>> loop.run_until_complete(meta_task)
Now making chamomile tea.
Now making herbal tea.
Now making green tea.
['chamomile tea', 'green tea', 'herbal tea']
>>> meta_task.done()
True

In this case, the asyncio.gather function received three coroutine objects. It wrapped them all in tasks under the hood, and returned a single task that serves as an aggregation of all three.

Notice that scheduling the meta_task object effectively schedules the three tasks gathered underneath it. Once you run the loop, the three subtasks all run.

In the case of a task created with asyncio.gather, the result is always a list, and that list contains the results of the individual tasks that were gathered. The order of the list of results is guaranteed to be the same order in which the tasks were gathered (but the tasks are not guaranteed to be run in that order). Therefore, the list of strings you got back are in the same order as the registered coroutines in the asyncio.gather call.

The asyncio.gather paradigm also provides the opportunity to add a callback to the set of tasks as a whole, rather than the individual tasks. What if you only want a callback to run once all of the tasks are completed, but it does not matter to you in which order they complete?

>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>>
>>> @asyncio.coroutine
... def make_tea(variety):
...     print('Now making %s tea.' % variety)
...     return '%s tea' % variety
...
>>> def mix(future):
...     print('Mixing the %s together.' % ' and '.join(future.result()))
...
>>> meta_task = asyncio.gather(make_tea('herbal'), make_tea('green'))
>>> meta_task.add_done_callback(mix)
>>>
>>> loop.run_until_complete(meta_task)
Now making green tea.
Now making herbal tea.
Mixing the green tea and herbal tea together.
['green tea', 'herbal tea']

The first thing that happened when you called run_until_complete was that both of the individual tasks gathered into meta_task ran, individually. Finally, the mix function ran, only after both of the individual tasks had run. This is because the meta_task is not considered to be done until after all of its individual tasks are done, so only once both individual tasks complete does it trigger the callback.

You can also see that the Future object that the mix function received was meta_task, not the individual tasks, and, therefore, its result method returned a list of both of the individual results.

Waiting on Tasks

Another tool that the asyncio module provides is the built-in wait coroutine. The asyncio.wait coroutine takes a sequence of coroutines or tasks (wrapping any coroutines in tasks) and returns once they are done. Note that the signature here is distinct from asyncio.gather. gather takes each coroutine or task as a single positional argument, whereas wait expects a list.

Additionally, wait accepts a parameter to return when any of its tasks complete, rather than only returning when all of them do. Regardless of whether this flag is set, the wait method always returns a two-tuple, with the first element being the Future objects that have completed, and the second element being those that are still pending.

Consider the following example that is similar to how you previously used asyncio.gather:

>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>>
>>> @asyncio.coroutine
... def make_tea(variety):
...     print('Now making %s tea.' % variety)
...     return '%s tea' % variety
...
>>> coro = asyncio.wait([make_tea('chamomile'), make_tea('herbal')])
>>>
>>> loop.run_until_complete(coro)
Now making chamomile tea.
Now making herbal tea.
({Task(<coro>)<result='herbal tea'>, Task(<coro>)<result='chamomile tea'>}, set())

Note a couple of subtle differences here. First, unlike the gather method, the wait method returns a coroutine. This has its value; you can use it in a yield from statement, for example.

On the other hand, you are unable to attach callbacks directly to a coroutine returned from wait. If you want to do this, you must wrap it in a task using asyncio.async.

Also, the result is different. The asyncio.gather function aggregated the results in a list, and returned that. The result for asyncio.wait is a two-tuple containing the actual Future objects (which themselves contain their results). Additionally, the Future objects are reorganized. The asyncio.wait routine places them into two sets—one set for those that are done, and another set for those that are not. Because sets are themselves an unordered structure, that means you must rely on the Future objects to piece together which result corresponds to which task.

>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>>
>>> coro = asyncio.wait([asyncio.sleep(5), asyncio.sleep(1)], timeout=3)
>>> loop.run_until_complete(coro)
({Task(<sleep>)<result=None>}, {Task(<sleep>)<PENDING>})

>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>>
>>> coro = asyncio.wait([
...     asyncio.sleep(3),
...     asyncio.sleep(2),
...     asyncio.sleep(1),
... ], return_when=asyncio.FIRST_COMPLETED)
>>>
>>> loop.run_until_complete(coro)
({Task(<sleep>)<result=None>},
 {Task(<sleep>)<PENDING>, Task(<sleep>)<PENDING>})

>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>>
>>> @asyncio.coroutine
... def raise_ex_after(seconds):
...     yield from asyncio.sleep(seconds)
...     raise RuntimeError('Raising an exception.')
...
>>> coro = asyncio.wait([
...     asyncio.sleep(1),
...     raise_ex_after(2),
...     asyncio.sleep(3),
... ], return_when=asyncio.FIRST_EXCEPTION)
>>>
>>> loop.run_until_complete(coro)
({Task(<raise_ex_after>)<exception=RuntimeError('Raising an exception.',)>,
  Task(<sleep>)<result=None>},
 {Task(<sleep>)<PENDING>})

>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>>
>>> coro = asyncio.wait([
...     asyncio.sleep(1),
...     asyncio.sleep(2),
... ], return_when=asyncio.FIRST_EXCEPTION)
>>>
>>> loop.run_until_complete(coro)
({Task(<sleep>)<result=None>, Task(<sleep>)<result=None>}, set()) 

Maximum Size

It is also possible to give a Queue object a maximum size, by setting the maxsize keyword argument when creating the queue.

>>> import asyncio
>>> queue = asyncio.Queue(maxsize=5)

If you do this, the Queue will not allow any more than the maximum number of items onto the queue. A call to the put method will simply wait until a previous item is removed, and then (and only then) will it place the item on the queue. If you call put_nowait and the queue is full, it will raise QueueFull.