Python多进程程序使用多个os.pipe管道通信的异常排查

你的代码出问题核心是管道描述符泄漏、IO资源未正确释放、系统调用参数错误三个原因，对应你遇到的两个现象逐一说明：

问题根因 #

1. 加第三个管道后文件写入失效、进程卡死
   os.pipe创建的是半双工管道，内核会统计管道写端的引用计数：只要还有任意进程持有管道的写端文件描述符，读端调用read()就会一直阻塞等待数据，不会返回EOF。
   你的代码里有三个直接触发错误的点：
   • 第一次fork后的顶层父进程没有关闭自己用不到的r3、w3描述符，导致第三根管道的写端一直被顶层父进程持有，第二层子进程读r3时永久阻塞
   • 第二层父进程（pid2>0分支）里调用os.write(verarbeiten)时没有传入目标文件描述符，os.write要求第一个参数必须是可写的打开文件描述符，这行代码会直接抛出TypeError，进程异常退出时，之前写file.txt的用户态缓冲区没有被刷新到磁盘，表现为文件没写入内容
   • 所有进程fork后都没有及时关闭自己用不到的管道端，进一步加剧了引用计数错误，导致管道读写逻辑完全混乱
2. 文件内容重复写入两次
   你写file.txt时，打开文件后没有主动关闭、也没有刷新缓冲区，就直接调用了第二次fork。fork会把父进程地址空间里的文件缓冲区内容完整复制给子进程，后续两个进程（第二层父、子进程）退出时都会各自把缓冲区内容刷到磁盘，相当于同一段内容写了两次。

修复规则 #

写基于os.fork+os.pipe的多进程代码，必须严格遵守三个规则：

• 任何进程fork完成后，第一时间关闭自己永远不会用到的所有管道读/写端，不要留多余的文件描述符
• 不管是普通文件还是管道，用完立刻关闭，优先用with上下文管理器自动处理关闭、缓冲区刷新逻辑，避免fork时复制缓冲区导致重复写入
• 调用os.write时必须传入合法的文件描述符作为第一个参数，且写入内容必须是bytes类型

修复后可运行代码 #

import os
import random

if __name__ == "__main__":
    r1, w1 = os.pipe()
    r2, w2 = os.pipe()
    r3, w3 = os.pipe()

    pid = os.fork()

    # 顶层父进程：生成随机数，写入前两个管道
    if pid > 0:
        # 关闭所有用不到的管道端
        os.close(r1)
        os.close(r2)
        os.close(r3)
        os.close(w3)

        n = random.sample(range(1, 1000), 5)
        print(f"生成的随机数列表: {n}")
        zahlen_bytes = "".join(str(x) for x in n).encode("utf-8")
        print(f"序列化后传输内容: {zahlen_bytes.decode()}")

        os.write(w1, zahlen_bytes)
        os.write(w2, zahlen_bytes)
        # 写完立刻关闭写端，通知读端数据传输完成
        os.close(w1)
        os.close(w2)
        # 等待子进程退出，避免僵尸进程
        os.waitpid(pid, 0)

    # 第一层子进程：读第一根管道数据写文件，再fork第二层进程
    else:
        # 关闭不用的管道端
        os.close(w1)
        os.close(w3)

        # 读取第一根管道数据
        r1_fp = os.fdopen(r1, "r", encoding="utf-8")
        zahlen = r1_fp.read()
        r1_fp.close()
        print(f"\n第一层子进程收到数据: {zahlen}")

        # 写文件，with块退出时自动刷新缓冲区、关闭文件
        with open("file.txt", "w", encoding="utf-8") as f:
            f.write(zahlen)
        print("数据已成功写入file.txt")

        pid2 = os.fork()
        # 第二层父进程：读第二根管道，计算后写入第三根管道
        if pid2 > 0:
            os.close(r3)
            r2_fp = os.fdopen(r2, "r", encoding="utf-8")
            verarbeiten = r2_fp.read()
            r2_fp.close()
            calc_res = int(verarbeiten) - 1000
            print(f"第二层父进程计算结果({verarbeiten}-1000): {calc_res}")

            os.write(w3, str(calc_res).encode("utf-8"))
            os.close(w2)
            os.close(w3)
            os.waitpid(pid2, 0)

        # 第二层子进程：读第三根管道的计算结果
        else:
            os.close(w2)
            os.close(w3)
            print("第二层子进程启动，等待读取计算结果")
            r3_fp = os.fdopen(r3, "r", encoding="utf-8")
            final_res = r3_fp.read()
            r3_fp.close()
            print(f"第二层子进程收到计算结果: {final_res}")

运行后可以验证：

• file.txt正常生成，内容为拼接后的随机数字符串，无重复
• 三个管道读写均无阻塞，所有进程正常退出
• 第二层子进程可以正常收到第二层父进程计算的减1000结果

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