如何在 Python 的 `asyncio.gather` 中正确处理取消的任务

Question

既然 3.8 已经发布，我将再次尝试 asyncio 模块。但是，在尝试优雅地关闭事件循环时，我得到了意想不到的结果。具体来说，我正在监听SIGINT，取消正在运行的Tasks，收集那些Tasks，然后.stop()ing 事件循环。我知道Tasks 会在它们被取消时引发CancelledError，这将向上传播并结束我对asyncio.gather 的调用，除非根据documentation，我将return_exceptions=True 传递给asyncio.gather，这应该会导致gather 等待所有Tasks 取消并返回CancelledErrors 数组。但是，如果我尝试 gather 取消 Tasks，return_exceptions=True 似乎仍会导致我的 gather 呼叫立即中断。

这里是重现效果的代码。我正在运行 python 3.8.0：

# demo.py

import asyncio
import random
import signal


async def worker():
    sleep_time = random.random() * 3
    await asyncio.sleep(sleep_time)
    print(f"Slept for {sleep_time} seconds")

async def dispatcher(queue):
    while True:
        await queue.get()
        asyncio.create_task(worker())
        tasks = asyncio.all_tasks()
        print(f"Running Tasks: {len(tasks)}")

async def shutdown(loop):
    tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    for task in tasks:
        task.cancel()
    print(f"Cancelling {len(tasks)} outstanding tasks")
    results = await asyncio.gather(*tasks, return_exceptions=True)
    print(f"results: {results}")
    loop.stop()

async def main():
    loop = asyncio.get_event_loop()
    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(shutdown(loop)))
    queue = asyncio.Queue()
    asyncio.create_task(dispatcher(queue))

    while True:
        await queue.put('tick')
        await asyncio.sleep(1)


asyncio.run(main())

输出：

>> python demo.py 
Running Tasks: 3
Slept for 0.3071352174511871 seconds
Running Tasks: 3
Running Tasks: 4
Slept for 0.4152310498820644 seconds
Running Tasks: 4
^CCancelling 4 outstanding tasks
Traceback (most recent call last):
  File "demo.py", line 38, in <module>
    asyncio.run(main())
  File "/Users/max.taggart/.pyenv/versions/3.8.0/lib/python3.8/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/Users/max.taggart/.pyenv/versions/3.8.0/lib/python3.8/asyncio/base_events.py", line 608, in run_until_complete
    return future.result()
asyncio.exceptions.CancelledError

我猜测事件循环仍有一些我不理解的地方，但我希望所有 CancelledErrors 都作为存储在 results 中的对象数组返回，然后能够继续而不是立即看到错误。

Answer 1

导致错误的原因是什么？

使用asyncio.all_tasks() 的问题是它会返回所有任务，即使是您没有直接创建的任务。按照以下方式更改您的代码以查看您取消的内容：

for task in tasks:
    print(task)
    task.cancel()

您不仅会看到worker 相关的任务，还会看到：

<Task pending coro=<main() running at ...>

取消main 会导致asyncio.run(main()) 内部混乱，您会收到错误消息。让我们进行快速/脏修改以从取消中排除此任务：

tasks = [
    t 
    for t 
    in asyncio.all_tasks() 
    if (
        t is not asyncio.current_task()
        and t._coro.__name__ != 'main'
    )
]

for task in tasks:
    print(task)
    task.cancel()

现在你会看到你的results。

loop.stop() 导致错误

当您实现results 时，您将收到另一个错误Event loop stopped before Future completed。这是因为asyncio.run(main()) 想要运行到main() 完成。

你必须重组你的代码以允许你传递给asyncio.run的协程完成而不是停止事件循环，或者，例如，使用loop.run_forever()而不是asyncio.run。

这是我的意思的快速/肮脏的演示：

async def shutdown(loop):
    # ...

    global _stopping
    _stopping = True
    # loop.stop()

_stopping = False

async def main():
    # ...

    while not _stopping:
        await queue.put('tick')
        await asyncio.sleep(1)

现在您的代码可以正常工作了。不要在实践中使用上面的代码，这只是一个示例。试着按照我上面提到的那样重构你的代码。

如何正确处理任务

不要使用asyncio.all_tasks()。

如果您创建了一些将来想取消的任务，请将其存储并仅取消已存储的任务。伪代码：

i_created = []

# ...

task = asyncio.create_task(worker())
i_created.append(task)

# ...

for task in i_created:
    task.cancel()

这可能看起来不方便，但这是一种确保您不会取消不想被取消的事情的方法。

还有一件事

还请注意，asyncio.run() 执行 much more 不仅仅是启动事件循环。特别是it cancels所有挂起的任务才完成。在某些情况下它可能很有用，但我建议改为手动处理所有取消。