处理异步死锁

Question

此示例代码无限期挂起：

import asyncio


async def main():
    async def f():
        await g_task

    async def g():
        await f_task

    f_task = asyncio.create_task(f())
    g_task = asyncio.create_task(g())
    await f_task


asyncio.run(main())

我正在寻找一种自动检测和处理死锁的方法，就像 GoLang 所做的那样。

到目前为止，我想出了asyncio.wait_for() 的变体：

[编辑] 大修设计

https://gist.github.com/gimperiale/549cbad04c24d870145d3f38fbb8e6f0

原代码1行改动：

await wait_check_deadlock(f_task)

它有效，但有两个主要问题：

1. 它依赖于 asyncio.Task._fut_waiter，这是 CPython 的一个实现细节
2. 死锁的任务将永远保留在 RAM 中。 aw.cancel() 似乎什么都不做。如果我捕获了我的辅助函数引发的 RecursionError，asyncio.run() 在尝试取消所有任务时会引发另一个 RecursionError。

这个问题有更强大的解决方案吗？

Answer 1

已经对避免死锁进行了很多研究，存在一些实用的解决方案，但一般情况下，问题是不可判定的（我认为可以归结为停机问题）。

为了说明实用性，考虑一下：

await asyncio.sleep(2 ** (1 / random.random()))

根据你的运气，它要么很快回来，要么“几乎永远不会”。

这个技巧可以用来证明基于回调的程序是不可能预测的：

f = asyncio.Future()

async foo():
    await asyncio.sleep(2 ** (1 / random.random()))
    f.set_result(None)

async bar():
    await f

await asyncio.gather(foo(), bar())

同样，它可以应用于您的“纯”异步/等待程序：

async def f():
    await g_task

async def g():
    await asyncio.wait(f_task,
                       asyncio.sleep(2 ** (1 / random.random())),
                       return_when=asyncio.FIRST_COMPLETED)

f_task = asyncio.create_task(f())
g_task = asyncio.create_task(g())
await f_task

同时，不完美但实用的死锁检测器可能非常有用，请考虑将您的代码发布到核心异步开发和/或独立库。

当前的做法是使用 PYTHONASYNCIODEBUG=1 运行测试，这会显示未等待的任务（在读取结果/异常之前已销毁）。

您的库可能会更好，例如，它可以报告某些任务花费的时间超过 X，或者当依赖于给定任务的 DAG 任务增长过大时。