Pythonパイプラインに接続しようとしたときの `ValueError：閉じたファイルの入出力操作`

Question

次のコードを使用して、Python Popenプロセスのパイプ処理を簡素化しようとしました。この考え方は、Processは引数を持つプロセスを表していますが（stdoutまたはstdinは含まれていません）、次に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() 

はしかし、次のように：私はprevious.stdout.close()を省略した場合

ValueError: I/O operation on closed file 

注目すべきは、このエラーが消える：

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

が、私はエラーを取得します。しかし、 subprocess docsは、SIGPIPEを動作させたい場合には、これを強くお勧めします。

私は間違っていますか？

Answer 1

すぐにstdoutを閉じるのではなく、最後に閉じてください。一度に

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がまだアクティブであるので、あなたのコマンドは、完了していません。