python - 如何将自定义异常抛出到正在运行的任务中

Question

我试图弄清楚是否有可能将自定义异常抛出到正在运行的 asyncio 任务中，类似于 Task.cancel(self) 所实现的其中安排了 CancelledError在底层协程中提出。

我遇到了 Task.get_coro().throw(exc) ，但调用它似乎打开了一大 jar 蠕虫，因为我们可能会使任务处于不良状态。特别是考虑到 task is throwing CancelledError into its coroutine 时发生的所有机器.

考虑以下示例:

import asyncio

class Reset(Exception):
    pass

async def infinite():
    while True:
        try:
            print('work')
            await asyncio.sleep(1)
            print('more work')
        except Reset:
            print('reset')
            continue
        except asyncio.CancelledError:
            print('cancel')
            break

async def main():
    infinite_task = asyncio.create_task(infinite())
    await asyncio.sleep(0)  # Allow infinite_task to enter its work loop.
    infinite_task.get_coro().throw(Reset())
    await infinite_task

asyncio.run(main())

## OUTPUT ##
# "work"
# "reset"
# "work"
# hangs forever ... bad :(

我尝试做的甚至可行吗？感觉好像我不应该像这样操纵底层协程。任何解决方法？

Answer 1

无法将自定义异常抛出到正在运行的任务中。你不应该惹.throw - 这是一个实现细节，改变它可能会破坏一些东西。

如果您想将信息(关于重置)传递到任务中，请通过参数进行。以下是它的实现方式:

import asyncio
from contextlib import suppress


async def infinite(need_reset):
    try:
        while True:
            inner_task = asyncio.create_task(inner_job())

            await asyncio.wait(
                [
                    need_reset.wait(),
                    inner_task
                ], 
                return_when=asyncio.FIRST_COMPLETED
            )

            if need_reset.is_set():
                print('reset')
                await cancel(inner_task)
                need_reset.clear()
    except asyncio.CancelledError:
        print('cancel')
        raise  # you should never suppress, see:
               # https://stackoverflow.com/a/33578893/1113207


async def inner_job():
    print('work')
    await asyncio.sleep(1)
    print('more work')


async def cancel(task):
    # more info: https://stackoverflow.com/a/43810272/1113207
    task.cancel()
    with suppress(asyncio.CancelledError):
        await task


async def main():
    need_reset = asyncio.Event()
    infinite_task = asyncio.create_task(infinite(need_reset))

    await asyncio.sleep(1.5)
    need_reset.set()

    await asyncio.sleep(1.5)
    await cancel(infinite_task)


asyncio.run(main())

输出:

work
more work
work
reset
work
more work
work
cancel