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

Question

上下文

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

#!/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)与所有后续迭代一起无声地失败了。如果我使用results_2000.bin命令检查相应的报表文件(例如vegeta report )，就会看到一些错误片段，比如parse error: syntax error near offset 0 of 'ource":["c...'。

我的直觉告诉我，第二个子进程开始消耗第一个子进程放在行中间的输入，但是在subprocess.run()之后注入一个subprocess.run()并不能解决这个问题。如果是这样的话，我如何才能干净利落地解决这个问题，并确保每个子进程开始从一个“好”的位置开始阅读？

Answer 1

正如@Barmar的注释中提到的，Python3在缓冲文本模式下打开stdin，因此sys.stdin.read(1)和sys.stdin.readline()都会提前读取，并且不会将sys.stdin流重新定位到新行的开头。

然而，有一种方法可以通过在二进制模式下打开来禁用缓冲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:
[...]

另见原始I/O (Python3docs)

Answer 2

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

可以从https://stromberg.dnsalias.org/~strombrg/mtee.html那里借来一些想法

HTH