asynkit: a toolkit for asyncio

This module provides some handy tools for those wishing to have better control over the way Python's asyncio module does things

Installation

$ pip install asynkit

Coroutine Tools

eager() - lower latency IO

Did you ever wish that your coroutines started right away, and only returned control to the caller once they become blocked? Like the way the async and await keywords work in the C# language?

Now they can. Just decorate or convert them with acyncio.eager:

@asynkit.eager
async def get_slow_remote_data():
    result = await execute_remote_request()
    return result.important_data

async def my_complex_thing():
    # kick off the request as soon as possible
    future = get_slow_remote_data()
    # The remote execution may now already be in flight. Do some work taking time
    intermediate_result = await some_local_computation()
    # wait for the result of the request
    return compute_result(intermediate_result, await future)

By decorating your function with asynkit.eager, the coroutine will start executing right away and control will return to the calling function as soon as it blocks, or returns a result or raises an exception. In case it blocks, a Task is created and returned.

What's more, if the called async function blocks, control is returned directly back to the calling function maintaining synchronous execution. In effect, conventional code calling order is maintained as much as possible. We call this depth-first-execution.

This allows you to prepare and dispatch long running operations as soon as possible while still being able to asynchronously wait for the result.

asynckit.eager can also be used directly on the returned coroutine:

log = []
async def test():
    log.append(1)
    await asyncio.sleep(0.2) # some long IO
    log.append(2)

async def caller(convert):
    del log[:]
    log.append("a")
    future = convert(test())
    log.append("b")
    await asyncio.sleep(0.1) # some other IO
    log.append("c")
    await future

# do nothing
asyncio.run(caller(lambda c:c))
assert log == ["a", "b", "c", 1, 2]

# Create a Task
asyncio.run(caller(asyncio.create_task))
assert log == ["a", "b", 1, "c", 2]

# eager
asyncio.run(caller(asynkit.eager))
assert log == ["a", 1, "b", "c", 2]

coro() is actually a convenience function, invoking either coro_eager() or async_eager() (see below) depending on context.

coro_eager(), async_eager()

coro_eager() is the magic coroutine wrapper providing the eager behaviour:

1. It runs coro_start() on the coroutine.
2. If coro_is_blocked() returns False, it returns coro_as_future()
3. Otherwise, it creates a Task and invokes coro_contine() in the task.

The result is an awaitable, either a Future or a Task.

async_eager() is a decorator which automatically applies coro_eager() to the coroutine returned by an async function.

coro_start(), coro_is_blocked(), 1coro_as_future(), coro_continue()

These methods are helpers to perform coroutine execution and are what what power the coro_eager() function.

• coro_start() runs the coroutine until it either blocks, returns, or raises an exception. It returns a special tuple reflecting the coroutine's state.
• coro_is_blocked() returns true if the coroutine is in a blocked state
• coro_as_future() creates a future with the coroutine's result in case it didn't block
• coro_continue() is an async function which continues the execution of the coroutine from the initial state.

Event loop tools

Also provided is a mixin for the built-in event loop implementations in python, providing some primitives for advanced scheduling of tasks.

SchedulingMixin mixin class

This class adds some handy scheduling functions to the event loop. They primarily work with the ready queue, a queue of callbacks representing tasks ready to be executed.

• ready_len(self) - returns the length of the ready queue
• ready_pop(self, pos=-1) - pops an entry off the queue
• ready_insert(self, pos, element) - inserts a previously popped element into the queue
• ready_rotate(self, n) - rotates the queue
• call_insert(self, pos, ...) - schedules a callback at position pos in the queue

Concrete event loop classes

Concrete subclasses of Python's built-in event loop classes are provided.

• SchedulingSelectorEventLoop is a subclass of asyncio.SelectorEventLoop with the SchedulingMixin
• SchedulingProactorEventLoop is a subclass of asyncio.ProactorEventLoop with the SchedulingMixin on those platforms that support it.

Event Loop Policy

A policy class is provided to automatically create the appropriate event loops.

• SchedulingEventLoopPolicy is a subclass of asyncio.DefaultEventLoopPolicy which instantiates either of the above event loop classes as appropriate.

Use this either directly:

asyncio.set_event_loop_policy(asynkit.SchedulingEventLoopPolicy)
asyncio.run(myprogram())

or with a context manager:

with asynkit.event_loop_policy():
    asyncio.run(myprogram())

Scheduling functions

A couple of functions are provided making use of these scheduling features. They require a SchedulingMixin event loop to be current.

sleep_insert(pos)

Similar to asyncio.sleep() but sleeps only for pos places in the runnable queue. Whereas asyncio.sleep(0) will place the executing task at the end of the queue, which is appropriate for fair scheduling, in some advanced cases you want to wake sooner than that, perhaps after a specific task.

task_reinsert(pos)

Takes a task which has just been created (with asyncio.create_task() or similar) and reinserts it at a given position in the queue. It assumes the task is already at the end of the queue. Similarly as for sleep_insert() this can be useful to achieve certain goals.

create_task_descend(coro)

Implements depth-first task scheduling.

Similar to asyncio.create_task() this creates a task but starts it running right away, and positions the caller to be woken up right after it blcks. The effect is similar to using asynkit.eager() but it achieves its goals solely by modifying the runnable queue. A Task is always created, unlike eager, which only creates a task if the target blocks.

Runnable task helpers

A few functions are added to help working with tasks. They require a SchedulingMixin event loop to be current.

The following identity applies:

asyncio.all_tasks(loop) = asynkit.runnable_tasks(loop) | asynkit.blocked_tasks(loop) | {asyncio.current_task(loop)}

runnable_tasks(loop=None)

Returns a set of the tasks that are currently runnable in the given loop

blocked_tasks(loop=None)

Returns a set of the tasks that are currently blocked on some future in the given loop