从子进程读取流

Question

我需要在子进程生成时读取它的输出 - 可能不会在每个进程上读取write,但是在进程完成之前.我在这里和这里尝试过Python3文档和SO问题的解决方案,但是在孩子终止之前我仍然没有得到任何结果.

该应用程序用于监控深度学习模型的培训.我需要获取测试输出(每次迭代大约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.
• export 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
)