使用 asyncio 的相互递归协程

Question

我假设如果我用 asyncio 编写相互递归协程，它们将不会遇到最大递归深度异常，因为事件循环正在调用它们（并且就像蹦床一样）。然而，当我这样写它们时情况并非如此：

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))

运行时，我得到 RuntimeError: maximum recursion depth exceeded while calling a Python object。

有没有办法防止堆栈在使用 asyncio 的递归协程中增长？

Answer 1

为了防止堆栈增长，您必须允许每个协程在安排下一个递归调用后实际退出，这意味着您必须避免使用 yield from。相反，您使用 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 在递归调用 return 后安排对运行的回调。每个协程然后 return 一个 asyncio.Future 对象，当它计划的递归调用完成时，它的结果集在运行的回调中。

亲眼看到代码可能最容易理解：

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

现在，您的示例代码实际上并没有 return n 一直备份堆栈，因此您可以做一些功能上等效但更简单的东西：

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()

但我怀疑你真的打算 return n 一直往回走。

Answer 2

我把代码改成了async，await然后测了时间。我真的很喜欢它的可读性。

未来：

import asyncio

@asyncio.coroutine
def a(n):
    fut = asyncio.Future()
    def set_result(out):
        fut.set_result(out.result())
    if n > 1000: 
        return n
    else: 
        in_fut = asyncio.async(b(n+1))
        in_fut.add_done_callback(set_result)
    return fut

@asyncio.coroutine
def b(n):
    fut = asyncio.Future()
    def set_result(out):
        fut.set_result(out.result())
    in_fut = asyncio.async(a(n+1))
    in_fut.add_done_callback(set_result)
    return fut

import timeit
print(min(timeit.repeat("""
loop = asyncio.get_event_loop()
loop.run_until_complete(a(0))
""", "from __main__ import a, b, asyncio", number=10)))

结果：

% time python stack_ori.py 
0.6602963969999109
python stack_ori.py  2,06s user 0,01s system 99% cpu 2,071 total

异步，等待：

import asyncio

async def a(n):
    if n > 1000:
        return n
    else:
        ret = await asyncio.ensure_future(b(n + 1))
    return ret

async def b(n):
    ret = await asyncio.ensure_future(a(n + 1))
    return ret

import timeit
print(min(timeit.repeat("""
loop = asyncio.get_event_loop()
loop.run_until_complete(a(0))
""", "from __main__ import a, b, asyncio", number=10)))

结果：

% time  python stack.py
0.45157229300002655
python stack.py  1,42s user 0,02s system 99% cpu 1,451 total

Answer 3

在Python3.7中，可以通过asyncio.create_task()实现"trampoline"的效果，而不是直接等待协程

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

然而，这有一个缺点，即事件循环仍然需要跟踪所有中间任务，因为每个任务都在等待其后继任务。我们可以使用 Future 对象来避免这个问题。

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

使用 asyncio 的相互递归协程

Mutually recursive coroutines with asyncio

python

recursion

python-asyncio