如何将sys.stdout.buffer.write输出作为subprocess的输入？

问题描述 #

我有一个类通过runGenerator方法持续向sys.stdout.buffer写入无符号32位整数，功能正常。现在需要创建多个调用dieharder模块的子进程，要让这些子进程复用该生成器的输出作为stdin，避免重复生成数据（数据实时生成且耗时）。目前只能为每个子进程单独生成数据，效率极低，求可行方案。

现有代码 #

生成器实现 #

def runGenerator(self):
    while self.runEvent.is_set():
        sys.stdout.buffer.write(struct.pack('I', self.next()))

子进程调用逻辑 #

def runSub(self):
    args = [DIEHARDER, GENERATOR_NUMBER, '-d0']
    dieharderTestProc = subprocess.Popen(
        args, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    print(dieharderTestProc.stdout.readlines())
    self.runEvent.clear()

当前低效实现 #

while dieharderTestProc.returncode is None:
    dieharderTestProc.stdin.write(struct.pack('>I', struct.pack('I', self.next())))
    dieharderTestProc.poll()

可行解决方案思路 #

1. 多进程队列分发（跨平台推荐） #

单独启动一个生成器进程，将生成的字节数据放入multiprocessing.Queue，多个dieharder子进程从队列中取数据并写入自身stdin。队列会自动处理多进程间的数据同步，避免重复生成。

示例实现代码：

from multiprocessing import Process, Event, Queue
import subprocess
import struct
import time

# 生成器工作进程
def generator_task(run_event, data_queue):
    # 替换成你的生成逻辑，这里模拟生成无符号32位整数
    def next_value():
        import random
        return random.getrandbits(32)
    
    while run_event.is_set():
        packed_data = struct.pack('I', next_value())
        data_queue.put(packed_data)
        # 可选：根据生成速度加小延迟，防止队列占用过多内存
        time.sleep(0.0001)

# Dieharder子进程工作逻辑
def dieharder_task(data_queue, run_event, worker_id):
    args = ['dieharder', '-g', '200', '-d0']  # 替换为你的实际参数
    proc = subprocess.Popen(
        args,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=False
    )

    while run_event.is_set() or not data_queue.empty():
        try:
            # 超时等待队列数据，避免无限阻塞
            data = data_queue.get(timeout=1)
            proc.stdin.write(data)
            proc.stdin.flush()  # 强制将数据写入子进程
        except Queue.empty:
            continue

    # 关闭输入，等待子进程完成
    proc.stdin.close()
    output = proc.communicate()[0]
    print(f"Worker {worker_id} output:\n{output.decode()}")

if __name__ == "__main__":
    run_event = Event()
    run_event.set()
    # 设置队列最大容量，防止内存溢出
    data_queue = Queue(maxsize=2000)

    # 启动生成器进程
    gen_proc = Process(target=generator_task, args=(run_event, data_queue))
    gen_proc.start()

    # 启动多个Dieharder子进程
    worker_count = 3
    worker_procs = []
    for i in range(worker_count):
        proc = Process(target=dieharder_task, args=(data_queue, run_event, i+1))
        proc.start()
        worker_procs.append(proc)

    # 等待所有子进程完成
    for proc in worker_procs:
        proc.join()

    # 停止生成器进程
    run_event.clear()
    gen_proc.join()

2. 命名管道（FIFO，类Unix系统适用） #

创建系统级命名管道，生成器进程向管道写入数据，多个dieharder子进程直接从管道读取数据作为输入。这种方式无需额外进程同步逻辑，依赖操作系统管道机制。

步骤：

1. 创建命名管道（终端执行）：

mkfifo /tmp/random_gen_fifo

2. 修改生成器代码写入管道：

def runGenerator(self):
    with open('/tmp/random_gen_fifo', 'wb') as fifo:
        while self.runEvent.is_set():
            fifo.write(struct.pack('I', self.next()))
            fifo.flush()

3. 修改子进程调用逻辑，直接读取管道：

def runSub(self, worker_id):
    args = [DIEHARDER, GENERATOR_NUMBER, '-d0', '-f', '/tmp/random_gen_fifo']
    dieharderTestProc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    output = dieharderTestProc.communicate()[0]
    print(f"Worker {worker_id} output:\n{output.decode()}")
    self.runEvent.clear()

注意：需确保生成器进程先启动并打开管道写入端，否则子进程会阻塞等待数据。

3. Unix域套接字（类Unix系统适用） #

创建Unix域套接字作为数据传输通道，生成器作为服务器端发送数据，每个dieharder子进程作为客户端连接套接字读取数据。适合需要更灵活控制连接的场景。

内容的提问来源于stack exchange，提问作者civlon