python - Linux FIFO 在我期望的时候不返回 EOF

Question

让我们考虑以下 Python 代码，由 cpython 在 Linux 系统上执行(警告:它将尝试在 /tmp/first、/tmp/second 和 /tmp/third)。

#!/usr/bin/env python2
# -*- coding: utf-8 -*-

import subprocess
import os
import sys
import threading

class ThreadizedPopen(threading.Thread):

    def __init__(self, command, stdin_name, stdout_name):
        super(ThreadizedPopen, self).__init__()
        self.command = command
        self.stdin_name = stdin_name
        self.stdout_name = stdout_name
        self.returncode = None

    def run(self):
        with open(self.stdin_name, 'rb') as fin:
            with open(self.stdout_name, 'wb') as fout:
                popen = subprocess.Popen(self.command, stdin=fin, stdout=fout, stderr=None)
                popen.communicate()
                self.returncode = popen.returncode

def main():
    os.system('mkfifo /tmp/first')
    os.system('mkfifo /tmp/second')
    os.system('mkfifo /tmp/third')

    popen1 = ThreadizedPopen(['cat'], '/tmp/first', '/tmp/second')
    popen2 = ThreadizedPopen(['cat'], '/tmp/second', '/tmp/third')
    popen1.start()
    popen2.start()
    with open('/tmp/third') as fin:
        print fin.read()
    popen1.join()
    popen2.join()

if __name__ == '__main__':
    main()

然后我执行它，在另一个 shell 上，我在 /tmp/first 中写了一些东西(比如 echo test >/tmp/first)。我希望 Python 程序能够快速退出并打印我提供给第一个 FIFO 的相同内容。

理论上，我在 /tmp/first 中写入的字符串应该会被我的程序生成的两个 cat 进程复制到另外两个 FIFO，并且然后由主要 Python 程序获取并写入其标准输出。每一个cat进程一结束，就应该关闭自己的写FIFO端，使相应的读端返回EOF，触发后面的cat进程终止。使用 strace 查看程序会发现测试字符串已通过所有三个 FIFO 正确复制并由主 Python 程序读取。第一个 FIFO 也正确关闭(并且第一个 cat 进程与其管理器 Python 线程一起退出)。然而，第二个 cat 进程卡在 read() 调用中，需要从其读取 FIFO 中获取数据。

我不明白为什么会这样。来自pipe(t) man page (据我所知，它也涵盖了这种 FIFO)似乎一旦写入结束(及其所有副本)关闭，对 FIFO 的读取就会返回 EOF。根据 strace 这似乎是跟踪(特别是 cat 进程已死，因此它的所有文件描述符都已关闭；它的管理线程也关闭了它的描述符，我可以在 strace 输出中看到它)。

你能告诉我为什么会这样吗？如果有用，我可以发布 strace 输出。

Answer 1

我找到了 this question并简单地将 close_fds=True 添加到您的 subprocess 调用中。您的代码现在显示为:

#!/usr/bin/env python2
# -*- coding: utf-8 -*-

import subprocess
import os
import sys
import threading

class ThreadizedPopen(threading.Thread):

    def __init__(self, command, stdin_name, stdout_name):
        super(ThreadizedPopen, self).__init__()
        self.command = command
        self.stdin_name = stdin_name
        self.stdout_name = stdout_name
        self.returncode = None

    def run(self):
        with open(self.stdin_name, 'rb') as fin:
            with open(self.stdout_name, 'wb') as fout:
                popen = subprocess.Popen(self.command, stdin=fin, stdout=fout, stderr=None, close_fds=True)
                popen.communicate()
                self.returncode = popen.returncode

def main():
    os.system('mkfifo /tmp/first')
    os.system('mkfifo /tmp/second')
    os.system('mkfifo /tmp/third')

    popen1 = ThreadizedPopen(['cat'], '/tmp/first', '/tmp/second')
    popen2 = ThreadizedPopen(['cat'], '/tmp/second', '/tmp/third')
    popen1.start()
    popen2.start()
    with open('/tmp/third') as fin:
        print fin.read()
    popen1.join()
    popen2.join()

if __name__ == '__main__':
    main()

我将您的代码放在名为 fifo_issue.py 的脚本中，并在终端中运行它。脚本如您所料处于空闲状态(忽略 mkfifo: cannot create fifo):

$ python fifo_issue.py 
mkfifo: cannot create fifo ‘/tmp/first’: File exists
mkfifo: cannot create fifo ‘/tmp/second’: File exists
mkfifo: cannot create fifo ‘/tmp/third’: File exists

然后，在第二个终端中，我输入:

$ echo "I was echoed to /tmp/first!" > /tmp/first

回到仍在运行空闲线程的第一个终端:

$ python fifo_issue.py 
mkfifo: cannot create fifo ‘/tmp/first’: File exists
mkfifo: cannot create fifo ‘/tmp/second’: File exists
mkfifo: cannot create fifo ‘/tmp/third’: File exists
I was echoed to /tmp/first!

之后python正确退出