中断当前正在执行的所有 asyncio.sleep答案

【问题标题】：Interrupt all asyncio.sleep currently executing中断当前正在执行的所有 asyncio.sleep
【发布时间】：2016-09-09 15:25:14
【问题描述】：

在哪里

这是在 Linux，Python 3.5.1 上。

什么

我正在与asyncio 一起开发一个监控进程，它在await 的各个地方执行各种持续时间的asyncio.sleep 呼叫。

有时我希望能够中断所有所说的asyncio.sleep 呼叫并让所有任务正常进行，但我找不到如何做到这一点。一个示例是正常关闭监视进程。

如何（假设失败）

我认为我可以发送一个 ALRM 信号来达到这种效果，但进程终止了。我尝试通过以下方式捕获 ALRM 信号：

def sigalrm_sent(signum, frame):
    tse.logger.info("got SIGALRM")

signal.signal(signal.SIGALRM, sigalrm_sent)

然后我得到关于捕获 SIGALRM 的日志行，但 asyncio.sleep 调用没有中断。

如何（杂牌）

此时，我将所有 asyncio.sleep 调用替换为对该协程的调用：

async def interruptible_sleep(seconds):
    while seconds > 0 and not tse.stop_requested:
        duration = min(seconds, tse.TIME_QUANTUM)
        await asyncio.sleep(duration)
        seconds -= duration

所以我只需要选择一个不太小也不太大的TIME_QUANTUM。

但是

有没有办法中断所有正在运行的asyncio.sleep 呼叫而我错过了它？

【问题讨论】：

标签： python linux async-await python-asyncio

【解决方案1】：

中断所有正在运行的asyncio.sleep 调用似乎有点危险，因为它可以用于代码的其他部分，用于其他目的。相反，我会创建一个专用的sleep 协程来跟踪它的运行调用。然后可以通过取消相应的任务来中断它们：

def make_sleep():
    async def sleep(delay, result=None, *, loop=None):
        coro = asyncio.sleep(delay, result=result, loop=loop)
        task = asyncio.ensure_future(coro)
        sleep.tasks.add(task)
        try:
            return await task
        except asyncio.CancelledError:
            return result
        finally:
            sleep.tasks.remove(task)

    sleep.tasks = set()
    sleep.cancel_all = lambda: sum(task.cancel() for task in sleep.tasks)
    return sleep

例子：

async def main(sleep, loop):
    for i in range(10):
        loop.create_task(sleep(i))
    await sleep(3)
    nb_cancelled = sleep.cancel_all()
    await asyncio.wait(sleep.tasks)
    return nb_cancelled

sleep = make_sleep()
loop = asyncio.get_event_loop()
result = loop.run_until_complete(main(sleep, loop)) 
print(result)  # Print '6'

出于调试目的，loop.time = lambda: float('inf') 也可以。

【讨论】：

这似乎是一个合适的解决方案（尽管我会选择一个类或一个不需要make_sleep 的公共函数）并且它只需要（就我而言）一个额外的@987654327 @ 以适应 SIGTERM 的情况。非常感谢。

【解决方案2】：

根据文森特的回答，我使用了以下类（该类的每个实例都可以取消其所有正在运行的.sleep 任务，从而实现更好的划分）：

class Sleeper:
    "Group sleep calls allowing instant cancellation of all"

    def __init__(self, loop):
        self.loop = loop
        self.tasks = set()

    async def sleep(self, delay, result=None):
        coro = aio.sleep(delay, result=result, loop=self.loop)
        task = aio.ensure_future(coro)
        self.tasks.add(task)
        try:
            return await task
        except aio.CancelledError:
            return result
        finally:
            self.tasks.remove(task)

    def cancel_all_helper(self):
        "Cancel all pending sleep tasks"
        cancelled = set()
        for task in self.tasks:
            if task.cancel():
                cancelled.add(task)
        return cancelled

    async def cancel_all(self):
        "Coroutine cancelling tasks"
        cancelled = self.cancel_all_helper()
        await aio.wait(self.tasks)
        self.tasks -= cancelled
        return len(cancelled)

【讨论】：