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

Question

问题应该很简单，但我找不到任何相关信息。

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

谢谢

Answer 1

您的要求是可能的，但并非微不足道。首先，请注意，您永远不能在每个 await 上暂停，而只能在导致协程暂停的那些上暂停，例如asyncio.sleep() 或没有的stream.read()准备返回的数据。等待一个协程立即开始执行它，如果协程可以立即返回，它不会掉到事件循环中。 await 仅在等待者（或 其 等待者等）请求它时才暂停到事件循环。这些问题的更多细节：[1]、[2]、[3]、[4]。

考虑到这一点，您可以使用 this answer 中的技术拦截协程的每次恢复，并使用额外的代码检查任务是否暂停，如果是，则在继续之前等待恢复事件。

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())