使用asyncio.ensure_future启动的子任务如何关闭

Question

如果我有以下内容：

async def foo():
    for i in range(10):
        print "1"
        await asyncio.sleep(1)

async def bar()
    t = asyncio.ensure_future(foo())
    while not t.done():
        print("waiting")
        await asyncio.sleep(0.5)

async def baz():
    t = asyncio.ensure_future(bar())
    await asyncio.sleep(2.5)
    t.cancel()

当在一个较大的事件循环中调用baz()时，如何确保取消foo()?

- Eric

Answer 1

使用 try/finally 块可以捕获 CancelledError 异常：

task = asyncio.ensure_future(foo())
try:
    [...]
finally:
    task.cancel()  # return False if task is already finished

一些任务函数已经处理了子任务的取消：

result = await asyncio.wait_for(foo(), 60.)  
result, = await asyncio.gather(foo())

在这两个示例中，foo() 被包装在一个任务中，如果父任务被取消，它将自动被取消。

---

编辑

以下是协程控制另一个协程执行的简单示例：

async def control(coro, timeout=1.):
    task = asyncio.ensure_future(coro)
    while not task.done():
        try:
            print("The task is not finished yet")
            await asyncio.wait([task, asyncio.sleep(timeout)], "FIRST_COMPLETED")
        except asyncio.CancelledError:
            task.cancel()
    return task.result()  # Raise task.exception() if any

现在，await control(foo(), 1.)应该与await foo()完全相同（关于取消、异常和结果），只是它每秒打印The task is not finished yet，直到foo()完成。

- Vincent

---

因为我想观看其他内容并提前取消任务。 - Eric

@Eric 任务完成后，使用await task或task.result()将立即引发task.exception()（如果有的话）。这样行得通吗？ - Vincent

但是这样做总会至少引发一个 CancelledError，对吗？我能安全地忽略它吗，还是这会使得忽略任务有时无法取消？ - Eric

@Eric 我认为你说得有道理；我改变了我的示例，这样即使子任务没有完成，任务也不会停止。await task 和 await foo() 的工作方式也是如此。 - Vincent

@Eric 其实是有效的，可以查看 Task.cancel() 中的文档字符串和注释：[...] 它可能是一个捕获并忽略取消的任务，因此我们可能需要稍后再次取消它。 - Vincent