python-3.x - 从子进程中流式读取

Question

我需要在子进程生成时读取输出——也许不是在每次写入时读取，而是在进程完成之前读取。我已经尝试过Python3文档和SO问题的解决方案here和 here ，但在子进程终止之前我仍然一无所获。

该应用程序用于监控深度学习模型的训练。我需要获取测试输出(每次迭代大约 250 个字节，大约每隔 1 分钟一次)并观察统计失败。

• 我无法更改训练引擎；例如，我无法在子进程代码中插入 stdout.flush() 。
• 我可以合理地等待十几行输出的积累；我希望缓冲区填充能解决我的问题。

代码:变体已被注释掉。

父级

cmd = ["/usr/bin/python3", "zzz.py"]
# test_proc = subprocess.Popen(
test_proc = subprocess.run(
    cmd,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT
    )

out_data = ""
print(time.time(), "START")
while not "QUIT" in str(out_data):
    out_data = test_proc.stdout
    # out_data, err_data = test_proc.communicate()
    print(time.time(), "MAIN received", out_data)

子级 (zzz.py)

from time import sleep
import sys

for _ in range(5):
    print(_, "sleeping", "."*1000)
    # sys.stdout.flush()
    sleep(1)

print("QUIT this exercise")

尽管发送了 1000 多个字节的行，但缓冲区(在其他地方测试为 2kb；在这里，我已经高达 50kb)填充不会导致父级“看到”新文本。

为了让它发挥作用，我缺少什么？

<小时/>

更新有关链接、评论和 iBug 发布的答案:

• Popen 而不是 run 修复了阻塞问题。不知何故，我在文档和我对两者的实验中错过了这一点。
• universal_newline=True 巧妙地将字节返回为字符串:尽管有交错的空行(易于检测和丢弃)，但在接收端更易于处理。
• 将 bufsize 设置为很小的值(例如 1)不会产生任何影响；父级仍然需要等待子级填充 stdout 缓冲区，在我的例子中为 8k。
• 在执行之前导出 PYTHONUNBUFFERED=1 确实修复了缓冲问题。感谢 wim 提供链接。

除非有人想出一个规范的、漂亮的解决方案来使这些过时，否则我明天会接受 iBug 的答案。

Answer 1

subprocess.run 始终生成子进程，并阻塞线程直至其退出。

唯一的选择是使用 p = subprocess.Popen(...) 并使用 s = p.stdout.readline() 或 p.stdout.__iter__()(见下文)。

如果子进程在打印一行后刷新标准输出(请参阅下面的扩展注释)，则此代码对我有用。

cmd = ["/usr/bin/python3", "zzz.py"]
test_proc = subprocess.Popen(
    cmd,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT
)

out_data = ""
print(time.time(), "START")
while not "QUIT" in str(out_data):
    out_data = test_proc.stdout.readline()
    print(time.time(), "MAIN received", out_data)
test_proc.communicate()  # shut it down

查看我的终端日志(从 zzz.py 中删除点):

ibug@ubuntu:~/t $ python3 p.py
1546450821.9174328 START
1546450821.9793346 MAIN received b'0 sleeping \n'
1546450822.987753 MAIN received b'1 sleeping \n'
1546450823.993136 MAIN received b'2 sleeping \n'
1546450824.997726 MAIN received b'3 sleeping \n'
1546450825.9975247 MAIN received b'4 sleeping \n'
1546450827.0094354 MAIN received b'QUIT this exercise\n'

您还可以使用 for 循环来完成此操作:

for out_data in test_proc.stdout:
    if "QUIT" in str(out_data):
        break
    print(time.time(), "MAIN received", out_data)

<小时/>

如果您无法修改子进程，unbuffer(来自包 expect - 使用 APT 或 YUM 安装)可能会有所帮助。这是我的工作父代码没有更改子代码。

test_proc = subprocess.Popen(
    ["unbuffer"] + cmd,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT
)