Python含无限循环子进程交互时communicate挂起问题求助

场景与代码 #

需求是实现父进程读取子进程输出、与其交互、查看结果后终止进程，以下是父、子进程代码：

child.py #

from random import randint

x = 1
while x < 9:
    x = randint(0, 10)
    print("weeeee got", x)


name = input("what's your name?\n")
print("hello", name)

x = 0
while True:
    x += 1
    print("bad", x)

parent.py #

import curio
from curio import subprocess


async def main():
    p = subprocess.Popen(
        ["python", "child.py"],
        stdout=subprocess.PIPE,
        stdin=subprocess.PIPE,
    )

    async for line in p.stdout:
        line = line.decode("ascii")
        print(p.pid, "got:", line, end="")
        if "what" in line and "name" in line:
            out, _ = await p.communicate(input=b"yaboi\n")
            print(p.pid, "got:", out.decode("ascii"), end="")
            # stuff to properly kill the process ...
            return


if __name__ == "__main__":
    curio.run(main)

问题现象 #

运行后父进程卡在out, _ = await p.communicate(input=b"yaboi\n")处；移除子进程中前一段循环则恢复正常；给子进程的print添加flush=True也无效。进程状态如下：

4219 13.2  1.5 493968 487936 pts/0   S+   14:08   0:01 python parent.py
4223  100  0.0  13764  9600 pts/0    R+   14:08   0:14 python child.py

原因分析 #

1. 子进程stdout缓冲区阻塞：子进程前一段循环会产生大量输出，默认情况下Python的stdout为行缓冲模式，当输出量过大时缓冲区会被填满，子进程会阻塞在print操作上，无法推进到input交互环节，父进程因此收不到"what's your name?"的提示，后续流程卡住。
2. communicate()方法特性导致永久阻塞：communicate()会在发送输入后，一直等待子进程终止并读取所有输出。但子进程收到输入后，执行完print("hello", name)会进入无限循环，永远不会终止，导致父进程卡在该调用上。

修复方案 #

1. 子进程端：强制刷新输出缓冲区 #

修改子进程所有print语句，添加flush=True，确保输出立即写入管道，避免缓冲区阻塞：

from random import randint

x = 1
while x < 9:
    x = randint(0, 10)
    print("weeeee got", x, flush=True)


name = input("what's your name?\n")
print("hello", name, flush=True)

x = 0
while True:
    x += 1
    print("bad", x, flush=True)

2. 父进程端：替换communicate()为直接读写+主动终止进程 #

放弃使用communicate()，改为直接向子进程stdin写入数据并刷新，继续读取stdout，获取到预期交互结果后主动终止子进程：

import curio
from curio import subprocess


async def main():
    p = subprocess.Popen(
        ["python", "child.py"],
        stdout=subprocess.PIPE,
        stdin=subprocess.PIPE,
        text=True,  # 启用text模式，无需手动decode
    )

    async for line in p.stdout:
        print(p.pid, "got:", line, end="")
        if "what" in line and "name" in line:
            # 写入输入并刷新
            await p.stdin.write("yaboi\n")
            await p.stdin.flush()
            # 读取后续输出
            async for line in p.stdout:
                print(p.pid, "got:", line, end="")
                if "hello" in line:
                    # 主动终止子进程
                    await p.terminate()
                    return


if __name__ == "__main__":
    curio.run(main)

补充说明 #

• 若子进程循环输出量极大，父进程逐行读取可能跟不上，可考虑使用更高效的读取方式，或限制子进程输出频率。
• terminate()发送SIGTERM信号终止进程，需强制终止时可使用kill()发送SIGKILL信号。

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