检查异步屏蔽协程是否运行

Question

如果我有以下代码示例

async def coro():
    # Cancelled error could be raised here
    await asyncio.sleep(1)
    # Or here
    await asyncio.shield(
        another_coro()
    )
    # Or here

async def wait_on_it(loop):
    f = loop.create_task(coro())
    # Pretend f may or may not happen, I just sleep in this example
    await asyncio.sleep(1)
    if not f.done():
        f.cancel() # Will raise CancelledError when some await finishes in coro()

如何确定屏蔽任务是否实际运行？如果屏蔽任务确实运行，我有必须运行的重要逻辑。也许屏蔽那个功能不是正确的方法？

Answer 1

coro() 可以通过修改从调用者那里接收到的可变对象将信息传递给调用者：

class Ref:
    def __init__(self, **kwargs):
        self.__dict__.update(**kwargs)

async def coro(run_ref):
    await asyncio.sleep(1)
    run_ref.ran_another_coro = True
    await asyncio.shield(another_coro())

async def wait_on_it(loop):
    run_ref = Ref(ran_another_coro=False)
    f = loop.create_task(coro(run_ref))
    await asyncio.sleep(1)
    if not f.done():
        f.cancel()
    if run_ref.ran_another_coro:
         # ... another_coro() was started

由于asyncio.shield 无法挂起，如果wait_on_it 观察到run_ref.ran_another_coro 的真值，则保证another_coro() 已启动。