Mutually recursive coroutines with asyncio

Question

I had the hypothesis that if I wrote mutually recursive coroutines with asyncio, they would not hit the maximum recursion depth exception, since the event loop was calling them (and act like a trampoline). This, however, is not the case when I write them like this:

import asyncio

@asyncio.coroutine
def a(n):
    print("A: {}".format(n))
    if n > 1000: return n
    else: yield from b(n+1)

@asyncio.coroutine
def b(n):
    print("B: {}".format(n))
    yield from a(n+1)

loop = asyncio.get_event_loop()
loop.run_until_complete(a(0))

When this runs, I get RuntimeError: maximum recursion depth exceeded while calling a Python object.

Is there a way to keep the stack from growing in recursive coroutines with asyncio?

Answer 1

To keep the stack from growing, you have to allow each coroutine to actually exit after it schedules the next recursive call, which means you have to avoid using yield from. Instead, you use asyncio.async (or asyncio.ensure_future if using Python 3.4.4+) to schedule the next coroutine with the event loop, and use Future.add_done_callback to schedule a callback to run once the recursive call returns. Each coroutine then returns an asyncio.Future object, which has its result set inside the callback that's run when the recursive call it scheduled completes.

It's probably easiest to understand if you actually see the code:

import asyncio

@asyncio.coroutine
def a(n):
    fut = asyncio.Future()  # We're going to return this right away to our caller
    def set_result(out):  # This gets called when the next recursive call completes
        fut.set_result(out.result()) # Pull the result from the inner call and return it up the stack.
    print("A: {}".format(n))
    if n > 1000: 
        return n
    else: 
        in_fut = asyncio.async(b(n+1))  # This returns an asyncio.Task
        in_fut.add_done_callback(set_result) # schedule set_result when the Task is done.
    return fut

@asyncio.coroutine
def b(n):
    fut = asyncio.Future()
    def set_result(out):
        fut.set_result(out.result())
    print("B: {}".format(n))
    in_fut = asyncio.async(a(n+1))
    in_fut.add_done_callback(set_result)
    return fut

loop = asyncio.get_event_loop()
print("Out is {}".format(loop.run_until_complete(a(0))))


Output:
A: 0
B: 1
A: 2
B: 3
A: 4
B: 5
...
A: 994
B: 995
A: 996
B: 997
A: 998
B: 999
A: 1000
B: 1001
A: 1002
Out is 1002

Now, your example code doesn't actually return n all the way back up the stack, so you could make something functionally equivalent that's a bit simpler:

import asyncio

@asyncio.coroutine
def a(n):
    print("A: {}".format(n))
    if n > 1000: loop.stop(); return n
    else: asyncio.async(b(n+1))

@asyncio.coroutine
def b(n):
    print("B: {}".format(n))
    asyncio.async(a(n+1))

loop = asyncio.get_event_loop()
asyncio.async(a(0))
loop.run_forever()

But I suspect you really meant to return n all the way back up.

Answer 2

In Python 3.7, you can achieve the "trampoline" effect by using asyncio.create_task() instead of awaiting the coroutine directly.

import asyncio

async def a(n):
    print(f"A: {n}")
    if n > 1000: return n
    return await asyncio.create_task(b(n+1))

async def b(n):
    print(f"B: {n}")
    return await asyncio.create_task(a(n+1))

assert asyncio.run(a(0)) == 1002

However, this has the disadvantage that the event loop still needs to keep track of all the intermediate tasks, since each task is awaiting its successor. We can use a Future object to avoid this problem.

import asyncio

async def _a(n, f):
    print(f"A: {n}")
    if n > 1000:
        f.set_result(n)
        return
    asyncio.create_task(_b(n+1, f))

async def _b(n, f):
    print(f"B: {n}}")
    asyncio.create_task(_a(n+1, f))

async def a(n):
    f = asyncio.get_running_loop().create_future()
    asyncio.create_task(_a(0, f))
    return await f

assert asyncio.run(a(0)) == 1002