python - 从 subprocess.Popen 进行流式传输并使用两个子命令时出现死锁

Question

在我的 wsgi 应用程序中，我必须调用命令行工具来生成数据，有时还必须调用另一个命令行工具来转换该数据。这一切都通过标准输出/标准输入进行。我正在使用子进程，并且以前使用过通信，它工作正常，但速度较慢，因为它不是流式传输，并且尝试将其转换为从标准输出增量式流式传输是导致我出现问题的原因。

旧代码(可以工作但不能流式传输):

generator_process = subprocess.Popen(generator_command, stdout=subprocess.PIPE)
if convert_command:
    convert_process = subprocess.Popen(convert_command, stdout=subprocess.PIPE, stdin=subprocess.PIPE)

output, err = generator_process.communicate()

if convert_command:
    output, err = convert_process.communicate(output)

yield output

当前代码 - 当不需要进行转换时对我来说工作得很好。但否则会卡在 subprocess.stdout.read() 上(使用 readline 并不会以某种方式产生影响):

generator_process = subprocess.Popen(generator_command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL, bufsize=1)
convert_process = None
if convert_command:
    convert_process = subprocess.Popen(convert_command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, stdin=subprocess.PIPE, bufsize=1)

while True:
    chunk = generator_process.stdout.read(chunk_size)
    if convert_process:
        if chunk:
            convert_process.stdin.write(chunk)
        chunk = convert_process.stdout.read(chunk_size)
        if chunk: yield chunk
        else: break

    else:
        if chunk: yield chunk
        else: break

请注意，由于这是一个 wsgi 应用程序，因此 asyncio/coroutines 将无法工作(这是我在深入 asyncio 兔子洞后意识到的)。

目前的Python有没有办法从子进程流到客户端而不可能出现死锁？

Answer 1

如果没有任何子进程尝试从 stdin 读取数据，那么我在代码中看到的死锁的唯一原因是 .write(chunk) , .read(chunk_size)如果 convert_process 可能会不同步不逐字节返回(如果 .flush() 之后 .write(chunk) 没有帮助)。

模拟generator | convert Python 中的命令:

#!/usr/bin/env python3
from functools import partial
from subprocess import Popen, PIPE, DEVNULL

def get_chunks(generator_command, convert_command, chunk_size=1024):
    with Popen(generator_command, stdin=DEVNULL,
               stdout=PIPE, stderr=DEVNULL) as generator_process, \
         Popen(convert_command, stdin=generator_process.stdout, 
               stdout=PIPE, stderr=DEVNULL) as convert_process:
        yield from iter(partial(convert_process.stdout.read, chunk_size), b'')
    return generator_process.returncode, convert_process.returncode