reveal.js


from prezutils import setup


setup(
    name="Awaits, how do they work?",
    author="Yann Kaiser",
    author_extra=[
        Twitter("@YannKsr"),
        Employer("Criteo", is_hiring=True, at="Palo Alto"),
        PyPI("clize", purpose="Turn functions into CLIs"),
    ],
)

How they want us to use it


import asyncio

import aiohttp


async def get_things_from_internet(url):
    async with aiohttp.get(url) as resp:
        print(await resp.text())


loop = asyncio.get_event_loop()

all_requests = asyncio.gather([
    get_things_from_internet('https://www.google.com/')
    get_things_from_internet('https://www.python.org/')
])

result = loop.run_until_complete(all_requests)
print(result)

2011

Node JS

CGP Grey

How we really want to use it


async def fib(n):
    if n < 2:
        return n
    return (await fib(n-1)) + (await fib(n-2))

print(something_we_will_build_here(fib(100)))


def recur_fib(n):
    if n < 2:
        return n
    return recur_fib(n-1) + recur_fib(n-2)


def iter_fib(n):
    n_2 = 0
    n_1 = 1
    for _ in range(n - 1):
        n_2, n_1 = n_1, n_2+n_1
    return n_1

For those of you who have scrubbed it from your memory, Fibonacci numbers are this sequence of numbers you can compute by making a sum of the two previous numbers.
It's usually spelled out like above, in doubly-recursive form. You see it commonly in programming curriculums, as it brings up the concept of recursion.
In Python, you might be told about its iterative form, which is nice for learning how to deal with recursion in Python (and more specifically how to avoid it).
If you have time, run python -i 1-fibs.py from the examples folder and try then with n=10, 100 and 1000.


import dis

@dis.dis
async def myfunc():
    return await "some object"


  3           0 LOAD_CONST               1 ('some object')
              2 GET_AWAITABLE
              4 LOAD_CONST               0 (None)
              6 YIELD_FROM
              8 RETURN_VALUE


  3           0 LOAD_CONST               1 ('some object')
              2 GET_AWAITABLE
              4 LOAD_CONST               0 (None)
              6 YIELD_FROM
              8 RETURN_VALUE


'some object'

Python runs by following bytecode instructions (what your Python code is compiled to) in order and managing a part of memory designated for each function call (the stack)

First we load a constant, the string 'some object' and put it on the stack.


              GET_AWAITABLE

≈


stack[0] = stack[0].__await__()

GET_AWAITABLE is still not too complex. Most of the time it is equivalent to the above: replace the value on the top of the stack with the result of calling its __await__ method.


  3           0 LOAD_CONST               1 ('some object')
              2 GET_AWAITABLE
              4 LOAD_CONST               0 (None)
              6 YIELD_FROM
              8 RETURN_VALUE


<coroutine object>


  3           0 LOAD_CONST               1 ('some object')
              2 GET_AWAITABLE
              4 LOAD_CONST               0 (None)
              6 YIELD_FROM
              8 RETURN_VALUE


<coroutine object>
None

YIELD_FROM requires a second value on the stack, so it adds one.

(From what I gather, that value is set to whatever must be passed into the next generator when resuming this generator).


  3           0 LOAD_CONST               1 ('some object')
              2 GET_AWAITABLE
              4 LOAD_CONST               0 (None)
              6 YIELD_FROM
              8 RETURN_VALUE


<return value of the coro>

YIELD_FROM is a bit more tricky, so we'll make abstraction of it for now.

Let's just say that it replaces the two topmost items on the stack with a result.


  3           0 LOAD_CONST               1 ('some object')
              2 GET_AWAITABLE
              4 LOAD_CONST               0 (None)
              6 YIELD_FROM
              8 RETURN_VALUE

Finally, RETURN_VALUE takes a one value on the stack and makes it the return value of the function, destroying the part of the stack dedicated to this function call in the process.

One thing might jump at you: what is YIELD_FROM doing here?


import dis

@dis.dis
def x():
    yield from "some object"


  3           0 LOAD_CONST               1 ('some object')
              2 GET_YIELD_FROM_ITER
              4 LOAD_CONST               0 (None)
              6 YIELD_FROM
              8 RETURN_VALUE

YIELD_FROM is indeed also present when you use yield from in a generator. This is because they are implemented the same way.

(The only difference is a flag on the function that says whether the object must be used with for loops,

yield
              from

or next(), and the other one with await or send.)


  3           0 LOAD_CONST               1 ('some object')
              2 GET_AWAITABLE
              4 LOAD_CONST               0 (None)
              6 YIELD_FROM
              8 RETURN_VALUE

Now obviously, strings don't have an __await__ method, so this will fail when we run it. So let's replace 'some object' with something that has this method.


class AdditionAwaitable:
    def __init__(self, *args):
        self.args = args

    def __await__(self):
        result = yield ('add', *self.args)
        return result


async def add(*args):
    return await AdditionAwaitable(*args)

Remember how await used same same mechanisms as yield from. It's within __await__ that the veil of async/await is removed to reveal a generator!

In it, we just return a tuple with the string 'add' and whichever values were passed when creating the object

And since we're here to use await, let's await this object.

(This example is available as 3-await_add.py)


async def add(*args):
    return await AdditionAwaitable(*args)


              g = add(1, 2)


>>> g
<coroutine object add at 0x7f7f4a0ace60>
>>> dir(g)
['__await__', '__class__', '__del__', '__delattr__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__name__', '__ne__', '__new__', '__qualname__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', 'close', 'cr_await', 'cr_code', 'cr_frame', 'cr_running', 'send', 'throw']
>>> [attr for attr in dir(g) if not attr.startswith('_')]
['close', 'cr_await', 'cr_code', 'cr_frame', 'cr_running', 'send', 'throw']

Let's see if we can use it!

Calling add gives us a coroutine object. Since we don't know what that is, let's try dir() on it to check its attributes.

There's a bit too many, but if we filter out those that start with _, we find a few attributes of interest:
A bunch of cr_XXX methods, plus send, throw and close.


>>> g.send("spam")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: can''t send non-None value to a just-started coroutine


>>> g.send(None)
('add', 1, 2)


>>> g.send(None)
('add', 1, 2)


class AdditionAwaitable:
    def __init__(self, *args):
        self.args = args

    def __await__(self):
        result = yield ('add', *self.args)
        return result


async def add(*args):
    return await AdditionAwaitable(*args)

When we send something into it, it runs all the way until into the __await__ method, where it encounters a yield. And the value that was yielded is the value that we get!


>>> g.send(None)
('add', 1, 2)


>>> g.cr_frame
<frame object at 0x7f8c31d21dc8>
>>> inspect.getframeinfo(g.cr_frame)
Traceback(filename='add.py', lineno=10, function='add', code_context=['    result = await AdditionAwaitable(*args)\n'], index=0)
>>> inspect.getframeinfo(g.cr_await.gi_frame)
Traceback(filename='add.py', lineno=6, function='__await__', code_context=["        return (yield ('add', *self.args))\n"], index=0)

Let's have a look at those internal attributes then. If we use the inspect modue to look at cr_frame, we get an object that shows us where Python left off in

add

.

Digging a little deeper, cr_await gives us the object being awaited, and cr_await.gi_frame gives us where we are paused in the __await__ generator.


>>> g.send(None)
('add', 1, 2)
>>> g.send(3)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration: 3

Anyhow, that generator is asking us something, so let's be nice and give it the answer!

The generators and coroutines resume and eventually throw a StopIteration at us, with the result as the exception's value. Now they don't mean to be rude, it's just what generators do when they've finished.

Recap:

await is based on yield from
async functions pause before running
coro.send(...) resumes such a function
That function will progress until it hits a yield
coro will store where it is paused

Let's automate this!

#4: It will do so going through awaits and yield froms. It's only when it finds a yield that it will pause.

But we're here to make programs, so let's automate this!


def run_task(coro):
    msg = coro.send(None)

Any good program has functions, so here's a function that uses send


def run_task(coro):
    msg = coro.send(None)
    action, *args = msg
    if action == 'add':
        result = sum(args)


def run_task(coro):
    while True:
        msg = coro.send(None)
        action, *args = msg
        if action == 'add':
            result = sum(args)


def run_task(coro):
    result = None
    while True:
        msg = coro.send(result)
        action, *args = msg
        if action == 'add':
            result = sum(args)


def run_task(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)

We need to catch the final StopIteration and return its value.


def run_task(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)
        else:
            coro.throw(ValueError(action))


def run_task(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)
        else:
            coro.throw(ValueError(action))

print(run_task(add(1, 2)))

So let's run it. (Use 4-await_run.py.)

It works! (It usually does in slides.)


async def fib(n):
    if n < 2:
        return n
    result = await AdditionAwaitable(await fib(n - 2),
                                     await fib(n - 1))
    return result


cache = {}


async def fib(n):
    if n in cache:
        return cache[n]
    if n < 2:
        return n
    result = await AdditionAwaitable(await fib(n - 2),
                                     await fib(n - 1))
    cache[n] = result
    return result


run_task(fib(100))

But let's cheat a bit and add some caching so we don't recompute the same values twice.

If we run it (5-await_fib.py), it works well for small numbers, but if we use a bigger number (try 10000), we will exceed the stack limit. Oops!


def run_task(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)
        else:
            coro.throw(ValueError(action))


def run_task(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)
        elif action == 'sub':
            a, b = args
            result = a - b
        else:
            coro.throw(ValueError(action))


def run_task(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)
        elif action == 'sub':
            a, b = args
            result = a - b
        elif action == 'mul':
            a, b = args
            result = a * b
        elif action == 'div':
            a, b = args
            result = a / b
        else:
            coro.throw(ValueError(action))


def run_task(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)
        elif action == 'sub':
            a, b = args
            result = a - b
        elif action == 'mul':
            a, b = args
            result = a * b
        elif action == 'div':
            a, b = args
            result = a / b
        elif action == 'mod':
            a, b = args
            result = a % b
        elif action == 'divmod':
            a, b = args
            result = a / b, a % b
        elif action == 'getattr':
            obj, attr = args
            result = getattr(obj, attr)
        elif action == 'setattr':
            obj, attr, val = args
            result = setattr(obj, attr, val)
        else:
            coro.throw(ValueError(action))

Deferred

Future

Promise

Callbacks

a Promise


class Promise:
    def __init__(self):
        self.success = None
        self.result = None

    def set_success(self, value):
        self.success = True
        self.result = value
        return self

    def set_exception(self, exc):
        self.success = False
        self.result = exc
        return self

    def __await__(self):
        return (yield self)

So what does this look like in Python?
We can have it represented as an object with two properties. One that indicates the status of the promise (waiting / success / failure), and one with the result value or exception, depending on status. (See table below)

We can add __await__ so we can use await on Promises if we want to yield them to the task loop.


class Promise:
    def __init__(self):
        self.success = None
        self.result = None

	`p.success`	`p.result`
🥚	`None`
🐣	`True`	`12345`
🍳	`False`	`ValueError('oops')`


def run(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)
        else:
            coro.throw(ValueError(action))


def run(coro):
    result = None
    while True:
        try:
            msg = coro.send(result)
        except StopIteration as e:
            return e.value
        action, *args = msg
        if action == 'add':
            result = sum(args)
        else:
            coro.throw(ValueError(action))


def run(coro):
    result = None
    while True:
        try:
            promise = coro.send(result)
        except StopIteration as e:
            return e.value


def run(coro):
    promise = Promise().set_success(None)
    while True:
        try:
            promise = coro.send(promise.result)
        except StopIteration as e:
            return e.value


def run(coro):
    promise = Promise().set_success(None)
    while True:
        try:
            if promise.success:
                promise = coro.send(promise.result)
            else:
                promise = coro.throw(promise.result)
        except StopIteration as e:
            return e.value


def run(coro):
    promise = Promise().set_success(None)
    while True:
        if promise.success is None:
            ... # wait ??
        try:
            if promise.success:
                promise = coro.send(promise.result)
            else:
                promise = coro.throw(promise.result)
        except StopIteration as e:
            return e.value


coros = []


def add_coro(coro):
    kickoff_promise = Promise().set_success(None)
    result_promise = Promise()
    coros.append((coro, kickoff_promise, result_promise))
    return result_promise

To do this we need to keep state about all the coroutines we'll run. For each of them we'll store: What promise it is currently waiting on (initially resolved to None), and a promise that will hold the result of the coroutine. We return that promise so it can be awaited.


def run(initial_coro):
    final_result = add_coro(initial_coro)
    while final_result.success is None:
        ... # run coroutines
    if final_result.success:
        return final_result.result
    else:
        raise final_result.result

select a coro whose promise has materialized


        for i, (coro, awaited_promise, result_promise) in enumerate(coros):
            if awaited_promise.success is not None:
                coros.pop(i)
                break
        else:
            raise RuntimeException("Deadlock!")

run the coroutine


        # coro, awaited_promise, result_promise
        try:
            if awaited_promise.success:
                new_promise = coro.send(awaited_promise.result)
            else:
                new_promise = coro.throw(awaited_promise.result)
        except StopIteration as e:
            result_promise.set_success(e.value)
        except BaseException as e:
            result_promise.set_exception(e)
        else:
            coros.append((coro, new_promise, result_promise))

Then we run the coroutine we picked.
We put it back in coros when it yields a new promise. We put its reseult in the promise when it completes (StopIteration), and we do the same when any other exception arises.


coros = []


def add_coro(coro):
    kickoff_promise = Promise().set_success(None)
    result_promise = Promise()
    coros.append((coro, kickoff_promise, result_promise))
    return result_promise


def run(initial_coro):
    final_result = add_coro(initial_coro)
    while final_result.success is None:
        for i, (coro, awaited_promise, result_promise) in enumerate(coros):
            if awaited_promise.success is not None:
                coros.pop(i)
                break
        else:
            raise RuntimeException("Deadlock!")
        try:
            if awaited_promise.success:
                new_promise = coro.send(awaited_promise.result)
            else:
                new_promise = coro.throw(awaited_promise.result)
        except StopIteration as e:
            result_promise.set_success(e.value)
        except BaseException as e:
            result_promise.set_exception(e)
        else:
            coros.append((coro, new_promise, result_promise))
    if final_result.success:
        return final_result.result
    else:
        raise final_result.result


cache = {}


async def fib(n):
    if n in cache:
        return cache[n]
    if n < 2:
        return n
    result = await fib(n-2) + await fib(n-1)
    cache[n] = result
    return result

We can run our existing fib function as-is...


cache = {}


async def fib(n):
    if n in cache:
        return cache[n]
    if n < 2:
        return n
    result = await add_coro(fib(n-2)) + await add_coro(fib(n-1))
    cache[n] = result
    return result


run(fib(10000))

... but let's use our new multitasking ability.

If we run it (6-promise_fib), we can ask for numbers above 10000 without issue (other than becoming slow at higher values).

Recap:

Promises simplify our task loop
We made our task loop switch between coroutines
Creating new tasks help us cheat the stack limit

I was dreaming of using awaits to compute Fibonacci numbers. How weird...


async def echo_server(conn):
    text = await read(conn)
    while text.strip() != b'bye':
        print(f"Received: {text}")
        await write(conn, b'echo: ' + text)
        text = await read(conn)
    conn.close()

select

epoll

IOCP

…


def run(initial_coro):
    final_result = add_coro(initial_coro)
    while final_result.success is None:
        for i, (coro, awaited_promise, result_promise) in enumerate(coros):
            if awaited_promise.success is not None:
                coros.pop(i)
                break
        else:
            raise RuntimeException("Deadlock!")
        try:
            if awaited_promise.success:
                new_promise = coro.send(awaited_promise.result)
            else:
                new_promise = coro.throw(awaited_promise.result)
        except StopIteration as e:
            result_promise.set_success(e.value)
        except BaseException as e:
            result_promise.set_exception(e)
        else:
            coros.append((coro, new_promise, result_promise))
    if final_result.success:
        return final_result.result
    else:
        raise final_result.result


        for i, (coro, awaited_promise, result_promise) in enumerate(coros):
            if awaited_promise.success is not None:
                coros.pop(i)
                break
        else:
            if reads or writes:
                check_socks(timeout=None)
                continue
            else:
                raise RuntimeException("Deadlock!")


reads = collections.defaultdict(list)
writes = collections.defaultdict(list)


def readability(sock):
    p = Promise()
    reads[sock].append(p)
    return p

def writability(sock):
    p = Promise()
    writes[sock].append(p)
    return p


async def read(sock):
    await readability(sock)
    return f.recv(4096)


async def write(sock, to_send):
    await writability(sock)
    return f.send(to_send)

We can make it nicer by implementing a read and write coroutine that await readability/writability and do the actual read/write operation.


def check_socks(timeout):
    rlist, wlist, _ = \
        select.select(list(reads), list(writes), [], timeout)

    for fd in rlist:
        for p in reads.pop(fd):
            p.set_success(None)

    for fd in wlist:
        for p in writes.pop(fd):
            p.set_success(None)


async def echo_server(conn):
    text = await read(conn)
    while text.strip() != b'bye':
        print(f"Received: {text}")
        await write(conn, b'echo: ' + text)
        text = await read(conn)
    conn.close()


async def launch_server(func, port):
    sock = socket.socket(
        type=socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
    sock.bind(('127.0.0.1', port))
    sock.listen()
    print(f"Listening on port {port}")
    while True:
        await readability(sock)
        conn, addr = sock.accept()
        print(f'Connected to {addr}')
        add_coro(func(conn))


run(launch_server(echo_server, 4321))

Great, we just need to launch our server.
To do that, we'll have to deal with the socket module.
To put it shortly, we listen on the network for any connecting client. When the socket we use to listen becomes readable, this means a client is attempting to connect. We accept it and get a socket for the resulting connection, which we pass to the function we're using as server (that would be echo_server).


run(launch_server(echo_server, 4321))

We can run it (7-promise_netw.py), and if we use netcat (nc 127.0.0.1 4321), we can write one-line messages that will be echoed back until we say bye.
You can even connect multiple instances of netcat to it and it will continue to serve both.

Recap:

select & co help us not make blocking calls
We can upgrade our task loop to an event loop
We can build friendlier tools using select and Promises

Thank you!

@YannKsr

Examples:
https://github.com/epsy/ahdtw

Slides:
https://epsy.github.io/ahdtw

MORE!

curio

PEP 525

ceval.c

Callback hell

dir()