python - 如果一个任务失败，如何取消收集中的所有剩余任务？

Question

万一gather的一项任务引发异常，其他人仍然可以继续。
嗯，这不完全是我需要的。我想区分致命的错误和需要取消所有剩余任务的错误，以及不是而是应该在允许其他任务继续的同时记录的错误。
这是我实现这一点的失败尝试:

from asyncio import gather, get_event_loop, sleep

class ErrorThatShouldCancelOtherTasks(Exception):
    pass

async def my_sleep(secs):
    await sleep(secs)
    if secs == 5:
        raise ErrorThatShouldCancelOtherTasks('5 is forbidden!')
    print(f'Slept for {secs}secs.')

async def main():
    try:
        sleepers = gather(*[my_sleep(secs) for secs in [2, 5, 7]])
        await sleepers
    except ErrorThatShouldCancelOtherTasks:
        print('Fatal error; cancelling')
        sleepers.cancel()
    finally:
        await sleep(5)

get_event_loop().run_until_complete(main())

(这里的 finally await sleep 是为了防止解释器立即关闭，这会自行取消所有任务)
奇怪的是，打电话 cancel在 gather实际上并没有取消它!

PS C:\Users\m> .\AppData\Local\Programs\Python\Python368\python.exe .\wtf.py
Slept for 2secs.
Fatal error; cancelling
Slept for 7secs.

我对这种行为感到非常惊讶，因为它似乎与 the documentation 矛盾。 ，其中指出:

asyncio.gather(*coros_or_futures, loop=None, return_exceptions=False)

Return a future aggregating results from the given coroutine objects or futures.

(...)

Cancellation: if the outer Future is cancelled, all children (that have not completed yet) are also cancelled. (...)

我在这里缺少什么？如何取消剩余的任务？

Answer 1

您的实现的问题在于它在 sleepers.cancel() 已经引发之后调用 sleepers 。从技术上讲，gather() 返回的 future 处于完成状态，因此它的取消必须是 no-op。
要更正代码，您只需要自己取消 child ，而不是相信 gather 的 future 会这样做。当然，协程本身不可取消，因此您需要先将它们转换为任务(gather 无论如何都会这样做，因此您无需做额外的工作)。例如:

async def main():
    tasks = [asyncio.ensure_future(my_sleep(secs))
             for secs in [2, 5, 7]]
    try:
        await asyncio.gather(*tasks)
    except ErrorThatShouldCancelOtherTasks:
        print('Fatal error; cancelling')
        for t in tasks:
            t.cancel()
    finally:
        await sleep(5)

I am very surprised by this behavior since it seems to be contradictory to the documentation[...]

gather 最初的绊脚石是它并没有真正运行任务，它只是一个等待它们完成的助手。出于这个原因， gather 如果其中一些任务因异常而失败，则不会取消剩余的任务——它只是放弃等待并传播异常，让剩余的任务在后台继续进行。这是 reported as a bug ，但不是为了向后兼容而修复的，因为行为被记录下来并且从一开始就没有改变。但这里我们还有另一个问题:文档明确 promise 能够取消返回的 future 。您的代码正是这样做的，但它不起作用，原因很明显(至少我花了一段时间才弄明白，并且需要阅读 source )。事实证明， Future 的契约实际上阻止了它的工作。当你调用 cancel() 时， gather 返回的 future 已经完成，取消一个完成的 future 是没有意义的，它只是 no-op。 (原因是一个完整的 future 有一个明确定义的结果，可以被外部代码观察到。取消它会改变它的结果，这是不允许的。)
换句话说，文档并没有错，因为如果您在 await sleepers 完成之前执行了取消操作，它就会起作用。但是，它具有误导性，因为它似乎允许在其可等待提升之一的重要用例中取消 gather()，但实际上不允许。
使用 gather 时会出现这样的问题，这就是为什么许多人急切地等待(没有双关语)三重奏风格的托儿所 in asyncio 的原因。