子进程在终止前获取结果

Posted 2023-03-30

【中文标题】子进程在终止前获取结果

【英文标题】：subprocess get result before terminate

【发布时间】：2021-02-18 18:19:44

【问题描述】：

实时获取子流程的结果

我想在子进程终止之前实时获取每个结果 (sys.stdout)。 假设我们有以下 file.py。

import time,sys
sys.stdout.write('something')
while True:
    sys.stdout.write('something else')
    time.sleep(4)

好吧，我对子进程、异步和线程模块进行了一些尝试，尽管所有方法在进程完成时都会给我结果。理想情况下，我想自己终止进程并实时获取每个结果（stdout、stderr），而不是在进程完成时。

import subprocess
proc = sp.Popen([sys.executable, "/Users/../../file.py"], stdout = subprocess.PIPE, stderr= subproces.STDOUT)
proc.communicate() #This one received the result after finish

我也尝试使用 threading 模块和 asyncio 在另一个线程中使用 readline proc.stdout.readline()，但它也会等待进程完成。

我发现唯一有用的是psutil.Popen(*args, **kwargs) 的使用，我可以随时终止该进程并获取一些统计信息。但主要问题仍然是在每次打印时实时（异步）获取 file.py 的每个 sys.stdout 或 print。

*python3.6的首选解决方案

【问题讨论】：

你有没有试过在`sys.stdout.write()`之后调用sys.stdout.flush()。您可能还需要在传递给后者的字符串末尾添加换行符。

Thanx4respose。我已经尝试过了，但没有结果。试想一下，即使是第一个 sys.stdout 这也无法读取它并等到终止。

对于（近乎）实时的进程间通信，您可能需要使用multiprocessing.Manager。另一种可能性是mmap。

我更喜欢在不使用网络的情况下这样做。我有多种通过网络协议的选择，例如twisted、zmq等。核心是捕获sys.stdout每次被其他进程使用。

我的建议都不需要使用网络——它们提供了在同一台计算机上运行的进程之间进行通信的方法。刚想到第三个：multiprocessing.shared_memory.

【参考方案1】：

正如 cmets 中所述，首要的事情是确保您的 file.py 程序实际上按照您认为的方式写入数据。

例如，您显示的程序将在大约 40 分钟内不写入任何内容，因为这是每隔 4 秒发出的 14 字节打印填满 8 KB IO 缓冲区所需的时间。更令人困惑的是，如果您在 TTY 上测试某些程序（即仅运行它们），某些程序会出现写入数据，但当您将它们作为子进程启动时则不会。这是因为在 TTY 标准输出上是行缓冲的，而在管道上它是完全缓冲的。当输出没有刷新时，其他程序根本无法检测到输出，因为它被困在子进程的缓冲区中，它从不费心与任何人共享。

换句话说，别忘了冲洗：

while True:
    # or just print('something else', flush=True)
    sys.stdout.write('something else')
    sys.stdout.flush()
    time.sleep(4)

解决了这个问题，让我们来看看如何读取该输出。 Asyncio 为子进程提供了一个很好的基于流的接口，该接口非常有能力在任意输出到达时访问它。例如：

import asyncio

async def main():
    loop = asyncio.get_event_loop()
    proc = await asyncio.create_subprocess_exec(
        "python", "file.py",
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )
    # loop.create_task() rather than asyncio.create_task() because Python 3.6
    loop.create_task(display_as_arrives(proc.stdout, 'stdout'))
    loop.create_task(display_as_arrives(proc.stderr, 'stderr'))
    await proc.wait()

async def display_as_arrives(stream, where):
    while True:
        # 1024 chosen arbitrarily - StreamReader.read will happily return
        # shorter chunks - this allows reading in real-time.
        output = await stream.read(1024)
        if output == b'':
            break
        print('got', where, ':', output)

# run_until_complete() rather than asyncio.run() because Python 3.6
asyncio.get_event_loop().run_until_complete(main())

【讨论】：

@szZzr 好点；我测试它的程序从未退出，所以我没有注意到这个问题。我现在编辑了答案以包括（略有不同的）EOF 检测。

首先非常感谢您的解决方案...您的代码正在对 python 3.6 进行一些更改，并且在您的函数 display_as_arrives 中，您可以修改以下 while 语句 while not stream.at_eof() 以接收file.py 的数据直到它在这种情况下的 EOF 永远不会结束。

@szZzr 注意：我拒绝了建议的编辑，因为 Python 3.6 支持 async/await（它是在 3.5 中引入的），所以 yield from 和 @asyncio.coroutine 不应该被需要。此外，我实际上使用 Python 3.6 测试了答案中的代码，并且它有效。如果您对此有任何疑问，请在评论中告诉我，我们可以努力解决。

我很抱歉你完全正确！您可以只保留 EOF 和 readiline() 以避免出现 read() 的问题。非常感谢。 :)

@szZzr 请注意，readline() 将坚持阅读完整的行，因此它不会是“实时的”。答案中的代码甚至会拾取不包含换行符的打印件（只要子进程确保它们被实际打印，即刷新它们），正如您问题中的代码 sn-p 发出的那样。