是否有一种非阻塞方式来检查 asyncio 子进程是否处于活动状态？

Posted 2023-03-30

【中文标题】是否有一种非阻塞方式来检查 asyncio 子进程是否处于活动状态？

【英文标题】：Is there a non-blocking way to check if an asyncio subprocess is alive?

【发布时间】：2020-01-15 15:03:46

【问题描述】：

使用asyncio.create_subprocess_exec 时，将返回asyncio.subprocess.process。 documentation 表示不存在 poll 或 is_alive 类型的方法。似乎wait 或communicate 提供了查看进程是否正在运行的唯一方法，但它们正在阻塞调用，并且通信的异步版本没有超时选项。

有没有一种很好的方法来检查 asyncio 子进程是否以 非阻塞 方式处于活动状态？

对于is_alive 样式函数，我能想到的最好的方法是：

import asyncio

async def is_alive(proc):
    try:
        await asyncio.wait_for(proc.wait(), 0.001)
    except asyncio.TimeoutError:
        return True
    else:
        return False

虚拟用例：

async def foo():
    proc = await asyncio.create_subprocess_exec('sleep', '5')
    i = 0
    res = True
    while res:
        res = await is_alive(proc)
        print(f"[i] is_alive: res")
        # ... do foo stuff while we wait ...
        await asyncio.sleep(1)
        i += 1

loop = asyncio.get_event_loop()
loop.run_until_complete(foo())

输出：

[0] is_alive: True
[1] is_alive: True
[2] is_alive: True
[3] is_alive: True
[4] is_alive: True
[5] is_alive: False

【参考方案1】：

您应该检查 .returncode 属性 (https://docs.python.org/3/library/asyncio-subprocess.html#asyncio.asyncio.subprocess.Process.returncode)，如果进程正在运行，该属性将为 None（请注意，0 表示它已退出，因此简单的真实性检查将不起作用）。

如果可以，您应该避免轮询，而是创建一个未来，启动一个等待.wait() 的任务，然后取消您的后台任务。

【讨论】：

看起来这就是我正在寻找的，在文档细则中遗漏了一点。