python - 只要有待处理的取消屏蔽任务，但不再存在，我如何运行 asyncio 循环？

Question

我正在尝试向我现有的 asyncio 循环添加一些代码，以在 Ctrl-C 上提​​供干净的关闭。下面是它正在做的事情的抽象。

import asyncio, signal

async def task1():
    print("Starting simulated task1")
    await asyncio.sleep(5)
    print("Finished simulated task1")

async def task2():
    print("Starting simulated task2")
    await asyncio.sleep(5)
    print("Finished simulated task2")

async def tasks():
    await task1()
    await task2()

async def task_loop():
    try:
        while True:
            await asyncio.shield(tasks())
            await asyncio.sleep(60)
    except asyncio.CancelledError:
        print("Shutting down task loop")
        raise

async def aiomain():
    loop = asyncio.get_running_loop()
    task = asyncio.Task(task_loop())
    loop.add_signal_handler(signal.SIGINT, task.cancel)
    await task

def main():
    try:
        asyncio.run(aiomain())
    except asyncio.CancelledError:
        pass

#def main():
#    try:
#        loop = asyncio.get_event_loop()
#        loop.create_task(aiomain())
#        loop.run_forever()
#    except asyncio.CancelledError:
#        pass

if __name__ == '__main__':
    main()

在这个例子中，假设 task1 的序列和 task2需要在启动后完成，否则一些工件将处于不一致的状态。 (因此 asyncio.shield 包装器围绕着调用 tasks 。)

使用上面的代码，如果我在脚本启动后很快中断它并且它只是打印 Starting simulated task1然后循环停止和 task2永远不会开始。如果我尝试切换到 main 的版本这被注释掉了，即使循环被正确取消并且至少在几分钟内没有进一步发生，它也永远不会退出。它确实有一些进步，因为它至少完成了任何正在进行的 task1 序列。和 task2 .

头脑 Storm 中的一些可能的解决方案，尽管我仍然觉得一定有一些更简单的东西我错过了:

在 asyncio.shield 周围创建一个包装器它增加了一个由 asyncio.Condition 同步的变量对象，运行屏蔽函数，然后递减变量。然后，在 aiomain在 CancelledError处理程序，在重新引发异常之前等待变量达到零。 (在一个实现中，我可能会使用 __aexit__ 将它的所有部分组合到一个类中，在 CancelledError 逻辑上实现等待零。)

跳过使用 asyncio的取消机制，而是使用 asyncio.Event或类似的允许中断点或可中断的 sleep 。尽管这看起来似乎更具侵入性，需要我指定哪些点被认为是可中断的，而不是声明哪些序列需要避免取消。

Answer 1

这个问题问得好。我在制定答案时学到了一些东西，所以我希望你仍然关注这个线程。

首先要调查的是，shield() 方法是如何工作的？在这一点上，至少可以说文档令人困惑。直到我阅读了 test_tasks.py 中的标准库测试代码，我才弄明白。以下是我的理解:

考虑这个代码片段:

async def coro_a():
    await asyncio.sheild(task_b())
    ...
task_a = asyncio.create_task(coro_a())
task_a.cancel()

当task_a.cancel()语句被执行时，task_a确实被取消了。 await 语句立即抛出一个 CancelledError，无需等待 task_b 完成。但 task_b 继续运行。外部任务 (a) 停止但内部任务 (b) 没有。

这是您的程序的修改版本，说明了这一点。主要的变化是在你的 CancelledError 异常处理程序中插入一个等待，让你的程序多存活几秒钟。我在 Windows 上运行，这就是为什么我也稍微改变了你的信号处理程序，但这是一个小问题。我还在打印语句中添加了时间戳。

import asyncio
import signal
import time

async def task1():
    print("Starting simulated task1", time.time())
    await asyncio.sleep(5)
    print("Finished simulated task1", time.time())

async def task2():
    print("Starting simulated task2", time.time())
    await asyncio.sleep(5)
    print("Finished simulated task2", time.time())

async def tasks():
    await task1()
    await task2()

async def task_loop():
    try:
        while True:
            await asyncio.shield(tasks())
            await asyncio.sleep(60)
    except asyncio.CancelledError:
        print("Shutting down task loop", time.time())
        raise

async def aiomain():
    task = asyncio.create_task(task_loop())
    KillNicely(task)
    try:
        await task
    except asyncio.CancelledError:
        print("Caught CancelledError", time.time())
        await asyncio.sleep(5.0)
        raise

class KillNicely:
    def __init__(self, cancel_me):
        self.cancel_me = cancel_me
        self.old_sigint = signal.signal(signal.SIGINT,
                                        self.trap_control_c)

    def trap_control_c(self, signum, stack):
        if signum != signal.SIGINT:
            self.old_sigint(signum, stack)
        else:
            print("Got Control-C", time.time())
            print(self.cancel_me.cancel())

def main():
    try:
        asyncio.run(aiomain())
    except asyncio.CancelledError:
        print("Program exit, cancelled", time.time())

# Output when ctrlC is struck during task1
# 
# Starting simulated task1 1590871747.8977509
# Got Control-C 1590871750.8385916
# True
# Shutting down task loop 1590871750.8425908
# Caught CancelledError 1590871750.8435903
# Finished simulated task1 1590871752.908434
# Starting simulated task2 1590871752.908434
# Program exit, cancelled 1590871755.8488846        

if __name__ == '__main__':
    main()

你可以看到你的程序没有运行，因为它在 task_loop 被取消后就退出了，在 task1 和 task2 有机会完成之前。他们一直都在那里(或者更确切地说，如果程序继续运行，他们会一直在那里)。

这说明了 shield() 和 cancel() 如何交互，但它实际上并没有解决您陈述的问题。为此，我认为，您需要有一个可等待的对象，您可以使用它来使程序保持事件状态，直到完成重要任务。这个对象需要在顶层创建，并沿着堆栈向下传递到执行重要任务的地方。这是一个与您类似的程序，但可以按照您想要的方式执行。

我做了三个运行:(1)task1 期间的 control-C，(2)task2 期间的 control-C，(3)两个任务完成后的 control-C。在前两种情况下，程序一直持续到 task2 完成。在第三种情况下，它立即结束。

import asyncio
import signal
import time

async def task1():
    print("Starting simulated task1", time.time())
    await asyncio.sleep(5)
    print("Finished simulated task1", time.time())

async def task2():
    print("Starting simulated task2", time.time())
    await asyncio.sleep(5)
    print("Finished simulated task2", time.time())

async def tasks(kwrap):
    fut = asyncio.get_running_loop().create_future()
    kwrap.awaitable = fut
    await task1()
    await task2()
    fut.set_result(1)

async def task_loop(kwrap):
    try:
        while True:
            await asyncio.shield(tasks(kwrap))
            await asyncio.sleep(60)
    except asyncio.CancelledError:
        print("Shutting down task loop", time.time())
        raise

async def aiomain():
    kwrap = KillWrapper()
    task = asyncio.create_task(task_loop(kwrap))
    KillNicely(task)
    try:
        await task
    except asyncio.CancelledError:
        print("Caught CancelledError", time.time())
        await kwrap.awaitable
        raise

class KillNicely:
    def __init__(self, cancel_me):
        self.cancel_me = cancel_me
        self.old_sigint = signal.signal(signal.SIGINT,
                                        self.trap_control_c)

    def trap_control_c(self, signum, stack):
        if signum != signal.SIGINT:
            self.old_sigint(signum, stack)
        else:
            print("Got Control-C", time.time())
            print(self.cancel_me.cancel())

class KillWrapper:
    def __init__(self):
        self.awaitable = asyncio.get_running_loop().create_future()
        self.awaitable.set_result(0)

def main():
    try:
        asyncio.run(aiomain())
    except asyncio.CancelledError:
        print("Program exit, cancelled", time.time())

# Run 1 Control-C during task1
# Starting simulated task1 1590872408.6737766
# Got Control-C 1590872410.7344952
# True
# Shutting down task loop 1590872410.7354996
# Caught CancelledError 1590872410.7354996
# Finished simulated task1 1590872413.6747622
# Starting simulated task2 1590872413.6747622
# Finished simulated task2 1590872418.6750958
# Program exit, cancelled 1590872418.6750958
#
# Run 1 Control-C during task2
# Starting simulated task1 1590872492.927735
# Finished simulated task1 1590872497.9280624
# Starting simulated task2 1590872497.9280624
# Got Control-C 1590872499.5973852
# True
# Shutting down task loop 1590872499.5983844
# Caught CancelledError 1590872499.5983844
# Finished simulated task2 1590872502.9274273
# Program exit, cancelled 1590872502.9287038
#
# Run 1 Control-C after task2 -> immediate exit
# Starting simulated task1 1590873694.2925708
# Finished simulated task1 1590873699.2928336
# Starting simulated task2 1590873699.2928336
# Finished simulated task2 1590873704.2938952
# Got Control-C 1590873706.0790765
# True
# Shutting down task loop 1590873706.0804725
# Caught CancelledError 1590873706.0804725
# Program exit, cancelled 1590873706.0814824