如何正确地将stdin重定向到按顺序创建的多个子进程？

Question

上下文

我正在试验一个类似于vegeta的ramp-requests.py的脚本。在这个脚本中，我使用subprocess.run()顺序运行多个子进程，并期望脚本的标准输入在这些子进程的整个生命周期内（每个5秒钟）重定向到这些子进程

#!/usr/bin/env python3

import json
import os
import subprocess
import sys
import time

rates = [1.0, 2.0, 3.0, 4.0]

# Run vegeta attack
for rate in rates:
    filename='results_%i.bin' % (1000*rate)
    if not os.path.exists(filename):
        cmd = 'vegeta attack -format=json -lazy --duration 5s -rate %i/1000s -output %s' % (1000*rate, filename)
        print(cmd, file=sys.stderr)
        subprocess.run(cmd, shell=True, encoding='utf-8')

我按如下方式调用脚本，通过管道将无限量的输入传输到脚本，每个输入用新行分隔vegeta连续读取此输入，直到--duration结束：

$ target-generator | ./ramp-requests.py

问题

第一个子进程（rate=1.0）似乎像我预期的那样接收stdin，并且每次命令都成功运行

但是，第二次迭代（rate=2.0）与所有后续迭代一起以静默方式失败。如果我使用vegeta report命令检查相应的报告文件（例如results_2000.bin），我会看到诸如parse error: syntax error near offset 0 of 'ource":["c...'之类的错误片段

我的直觉告诉我，第二个子过程开始消耗第一个离开的输入，在中间行，但是在{{CD2>}之后注入^ {< CD8>}并不能解决它。如果是这样的话，我如何才能干净地解决这个问题，并确保每个子流程从“良好”的位置开始读取

Answer 1

正如在@Barmar的评论中提到的，Python3以缓冲文本模式打开stdin，因此sys.stdin.read(1)和sys.stdin.readline()都会导致预读，并且不会将sys.stdin流重新定位到新行的开头

然而，正如Denilson Sá Maia在对Setting smaller buffer size for sys.stdin?的回答中指出的那样，有一种方法可以通过以二进制模式打开sys.stdin来禁用缓冲：

unbuffered_stdin = os.fdopen(sys.stdin.fileno(), 'rb', buffering=0)

通过这样做，可以在每个子流程返回后，从该无缓冲io对象读取截断的输入，直到行尾：

# Run vegeta attack
for rate in rates:
  # [...]

  cmd = 'vegeta attack [...]'
  subprocess.run(cmd, shell=True, encoding='utf-8')

  # Read potentially truncated input until the next '\n' byte
  # to reposition stdin to a location that is safe to consume.
  unbuffered_stdin.readline()

打印读取行显示与以下输出类似的内容：

b'a4b-b142-fabe0e96a6ca"],"Ce-Type":["perf.drill"],"Ce-Source":["load-test"]}}\n'

所有子流程现在都已成功执行：

$ for r in results_*.bin; do vegeta report "$r"; done
[...]
Success       [ratio]                           100.00%
Status Codes  [code:count]                      200:5
Error Set:
[...]
Success       [ratio]                           100.00%
Status Codes  [code:count]                      200:7
Error Set:
[...]
Success       [ratio]                           100.00%
Status Codes  [code:count]                      200:8
Error Set:
[...]

---

另见io - Raw I/O（Python 3文档）

Answer 2

从父进程中的stdin中读取许多行，并将其作为-its-stdin传递给子进程。根据需要重复。这样，您就不必担心子进程会把stdin弄得一团糟

请随意借用https://stromberg.dnsalias.org/~strombrg/mtee.html的想法

嗯