是否可以在异步python中暂停和重新启动任务？

Question

这个问题应该很简单，但是我找不到任何关于它的东西。

我有一个异步python程序，其中包含一个运行时间相当长的任务，我希望能够在任意点挂起并重新启动（当然，任意意味着任何有wather关键字的地方）。我希望有一些类似于任务.挂起（）和任务.恢复（）的路线，但似乎没有。在任务或事件循环级别是否有任何用于此的API，或者我需要以某种方式自己执行此操作？我不想在每次等待之前放置事件.wait（） ...

谢谢

Answer 1

你所要求的是可能的，但不是微不足道的。首先，请注意，您永远不能在每个等待时都挂起，而只能在那些导致协程挂起的挂起上挂起，例如 asyncio.sleep（） 或没有准备返回数据的 stream.read（）。等待协程立即开始执行它，如果协程可以立即返回，则它不会下降到事件循环。仅当等待方（或其等待方等）请求事件循环时，await 才会挂起到事件循环。这些问题中的更多详细信息：[1]，[2]，[3]，[4]。

考虑到这一点，您可以使用此答案中的技术来拦截每个协程的恢复，并使用附加代码检查任务是否暂停，如果暂停，则在继续之前等待恢复事件。

import asyncio

class Suspendable:
    def __init__(self, target):
        self._target = target
        self._can_run = asyncio.Event()
        self._can_run.set()
        self._task = asyncio.ensure_future(self)

    def __await__(self):
        target_iter = self._target.__await__()
        iter_send, iter_throw = target_iter.send, target_iter.throw
        send, message = iter_send, None
        # This "while" emulates yield from.
        while True:
            # wait for can_run before resuming execution of self._target
            try:
                while not self._can_run.is_set():
                    yield from self._can_run.wait().__await__()
            except BaseException as err:
                send, message = iter_throw, err

            # continue with our regular program
            try:
                signal = send(message)
            except StopIteration as err:
                return err.value
            else:
                send = iter_send
            try:
                message = yield signal
            except BaseException as err:
                send, message = iter_throw, err

    def suspend(self):
        self._can_run.clear()

    def is_suspended(self):
        return not self._can_run.is_set()

    def resume(self):
        self._can_run.set()

    def get_task(self):
        return self._task

测试：

import time

async def heartbeat():
    while True:
        print(time.time())
        await asyncio.sleep(.2)

async def main():
    task = Suspendable(heartbeat())
    for i in range(5):
        print('suspending')
        task.suspend()
        await asyncio.sleep(1)
        print('resuming')
        task.resume()
        await asyncio.sleep(1)

asyncio.run(main())