`ValueError: I/O operation on closed file` 尝试连接 Python 管道时

Question

我尝试使用以下代码简化 Python Popen 进程之间的管道连接。这个想法是 Process 表示一个带有参数的进程（但没有标准输出或标准输入），然后 pipe 函数将它们连接起来。

def Process(parameters, *args, **kwargs):
    """
    Represents a process that can be piped into another
    """
    parameters = [str(p) for p in parameters]

    # Partially apply the constructor, so we can handle the piping later
    return functools.partial(subprocess.Popen, parameters, *args, **kwargs)


def pipe(commands, stdin=None):
    """
    Pipes a series of commands into each other
    :param commands: An array of commands, each of which is an instance of Process
    :param stdin: stdin to the first command
    :param kwargs: Any extra arguments to pass to subprocess.Popen
    :return:
    """
    # Keep track of previous processes
    processes = []

    # Each process's stdin is the stdout of the previous stage
    for i, cmd in enumerate(commands):
        if i == 0:
            process = cmd(stdin=subprocess.PIPE, stdout=subprocess.PIPE)
        else:
            previous = processes[-1]
            process = cmd(stdin=previous.stdout, stdout=subprocess.PIPE)

            # Close stdout of previous command so we get SIGPIPE
            previous.stdout.close()

        processes.append(process)

    first = processes[0]
    final = processes[-1]

    if first == final:
        # If we only have one process, return its output
        return first.communicate(stdin)
    else:
        # Pipe input into first process
        first.communicate(stdin)

        # Return Final process
        return final.communicate()

但是，如果我按如下方式运行管道函数：

stdout, stderr = pipe([
    Process(['tr', 'n', '\\n']),
    Process(['rev']),
    Process(['wc', '-l']),
], text)

我得到错误：

ValueError: I/O operation on closed file

值得注意的是，如果我省略 previous.stdout.close()，这个错误就会消失。但如果我想让 SIGPIPE 工作，subprocess docs 强烈建议不要这样做。

我做错了什么？

Answer 1

您不应该立即关闭标准输出，而应在最后关闭它们。

if first == final:
    # If we only have one process, return its output
    result = first.communicate(stdin)
    first.stdout.close()
    return first.communicate(stdin)
else:
    # Pipe input into first process
    first.communicate(stdin)

    # Return Final process
    result = final.communicate()
    for process in processes:
        process.stdout.close()
    return result

因为Popen 立即返回，你的命令还没有完成，所以 stdout 仍然处于活动状态。