为什么没有取消父协程？答案

【问题标题】：Why parent coroutine is not canceled?为什么没有取消父协程？
【发布时间】：2017-03-15 01:31:08
【问题描述】：

我正在使用 Python asyncio。我的程序只有三个协程。其中两个是我直接安排的，而第三个是从前者之一安排的。当用户按下Ctrl+C时，我想正确地完成我的程序：

import asyncio

async def coro1():
    try:
        print('coro1')
        await asyncio.sleep(1000)
    except Exception as e:
        print('coro1 exc %s' % repr(e))
        raise

async def coro2():
    try:
        print('coro2')
        await asyncio.ensure_future(coro3())
        await asyncio.sleep(1000)
    except Exception as e:
        print('coro2 exc %s' % repr(e))
        raise

async def coro3():
    try:
        print('coro3')
        await asyncio.sleep(1000)
    except Exception as e:
        print('coro3 exc %s' % repr(e))
        raise

loop = asyncio.get_event_loop()    
try:
    f1 = asyncio.ensure_future(coro1())
    f2 = asyncio.ensure_future(coro2())    
    loop.run_forever()
except KeyboardInterrupt:
    print('Exiting... Cancelling all tasks')

    f2.cancel()
    f1.cancel()

    # This code gives the same result:
    # for task in asyncio.tasks.Task.all_tasks(loop):
    #    task.cancel()

    print('Cancellation is done!')

    loop.stop()
    loop.run_forever()
finally:
    loop.close()

这段代码产生下一个输出：

coro1
coro2
coro3
^CExiting... Cancelling all tasks
Cancellation is done!
coro3 exc CancelledError()
coro1 exc CancelledError()
Task was destroyed but it is pending!
task: <Task pending coro=<coro2() running at test.py:15> wait_for=<Task cancelled coro=<coro3() done, defined at test.py:23>>>

所以我想知道，为什么coro2 没有被取消而coro3 实际上被取消了？

【问题讨论】：

协程 3 返回后协程 2 再休眠 1000 秒。尝试更短的睡眠时间（或等待 1000 秒！）
@shongololo，恐怕即使我完全删除await sleep(1000)，观察到的行为也不会改变。
啊，我没有看到您的代码示例的最低行（被滚动隐藏）。可能是协程 2 在循环关闭时没有机会返回异常并关闭。您可以在关闭循环之前等待检查所有任务是否已取消，但是您将看不到正在运行的异常...
@shongololo 看起来我找到了答案。

标签： python python-asyncio coroutine event-loop

【解决方案1】：

知道了！问题出在except 块中的这两行：

# ...
loop.stop()
loop.run_forever()

由于loop.stop()，预期的取消传播不起作用。如果将代码更改为以下内容：

# ...

try:
    f1 = asyncio.ensure_future(coro1())
    f2 = asyncio.ensure_future(coro2())    
    loop.run_forever()
except KeyboardInterrupt:
    print('Exiting... Cancelling all tasks')
    f2.cancel()
    f1.cancel()
    print('Cancellation is done!')    
    try:
        loop.run_forever()
        # Wait a very short time for f2 cancelation and press Ctrl+C again.
    except KeyboardInterrupt:
        loop.stop()
        loop.run_forever()
finally:
    loop.close()

Task was destroyed but it is pending! 的消息会消失。

更好的方法是使用loop.run_until_complete() 方法：

f1 = asyncio.ensure_future(coro1())
f2 = asyncio.ensure_future(coro2())
tasks = asyncio.gather(f1, f2)
try:
    loop.run_until_complete(tasks)
except KeyboardInterrupt:
    print('Exiting... Cancelling all tasks')
    tasks.cancel()  # or f1.cancel(); f2.cancel()
    print('Cancellation is done!')

    loop.run_forever()
    tasks.exception()  # To skip '_GatheringFuture exception was never retrieved' warning
finally:
    loop.close()

run_until_complete 添加内部回调，在完成（或取消）所有任务后停止循环。

【讨论】：