シェルスクリプトにおいて、2つ以上のアプリケーションでデータをやり取りさせたいときは、パイプ | を用います。他方、Pythonのsubprocessにおいてこうしたパイプを利用するには、Popenを直接使うと良いという知見を得たので、パイプって何？というところも含めてざっくりと自分なりに紹介します。

検証環境

uname -srvmpio
# Linux 5.15.133.1-microsoft-standard-WSL2 #1 SMP Thu Oct 5 21:02:42 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux
python3 --version
# Python 3.10.12

subprocessモジュールは便利

Pythonのsubprocessモジュールは便利なモジュールです。とくに、run() メソッドは優秀で、簡単に子プロセスを立ち上げ、その実行結果を取得できます。たとえば、以下のコマンドは、dd にブロックデバイスの/dev/zero から10バイトを1回分コピーして来い！という命令です。実際、バイト数を数えることができる関数である、len() で確かめると、確かに10バイトのデータがコピーされていることが分かります。

>>> import subprocess
>>> process = subprocess.run( ['dd', 'if=/dev/zero', 'bs=10', 'count=1'], capture_output=True, )
>>> process
CompletedProcess(args=['dd', 'if=/dev/zero', 'bs=10', 'count=1'], returncode=0, stdout=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', stderr=b'1+0 records in\n1+0 records out\n10 bytes copied, 6.6194e-05 s, 151 kB/s\n')
>>> len(process.stdout)
10

確かに、これは便利そうです。このようにして、run()メソッドを使用して、一コマンドの出力および実行結果を取得できました。

run()の使い道は限られる

実際には、2つ以上のコマンドを組み合わせてなにか有意な結果を得たいということもままあるでしょう。しかし、run()の融通が利かないところは、capture_output=True として、標準出力・標準エラー出力を捕捉したときに、それらをすべてメモリ上にため込んでしまうところです。

たとえば、あるブロックデバイスのフルバックアップを取るために、dd コマンドを叩いて、その出力を zstd に渡して圧縮したい、というケースを考えてみましょう。こうしたケースでは、dd の出力はわざわざメモリに保持する必要はなく、zstd に直接出力を渡したのち、圧縮してディスクに保存してもらえば事足ります。こういった場合、「パイプ」を使って、dd と ztd を直接つないでしまいましょう。

パイプは、2つの異なるプロセスがデータをやり取りするときに用いるデータの通り道であり、プロセス間通信（IPC）の一つです。パイプは、FIFO（First in First Out）と呼ばれるデータ構造に基づいていて、一種のキューです。Linuxを含め各種UNIX系のOSにおいては、無名パイプ（Anonymous Pipe）と名前付きパイプ（Named Pipe）というものがあり、前者は、シェルの | 、後者は、mkfifo コマンドによって作成できます。無名パイプについて、たとえば、wc に dd が出力した1KBのデータをパイプを通じて送信したいときは、以下のようにします。

dd if=/dev/urandom bs=1k count=1 | wc
# 1+0 records in
# 1+0 records out
# 1024 bytes (1.0 kB, 1.0 KiB) copied, 6.6427e-05 s, 15.4 MB/s
#       4      19    1024

前述のとおり、dd と zstd をつなぐとき、シェルでは、| という記号で表現されるパイプを使います。一方、Pythonの subprocess において、パイプは subprocess.PIPE によって表現されます。それでは、subprocessを使って実際にプロセス間でデータをやり取りさせてみましょう。

Popenを使ってプロセス間でデータをやり取りさせる

プロセス間でデータを直接やり取りさせるには、既存のプログラムに対する多少の工夫が必要になります。まずは、Popenインターフェースを直接使用するようにしましょう。そして、2つのプロセスそれぞれに引数を設定した後、subprocess.PIPEを用いて、一つ目のプロセスである process_one の標準出力を、二つ目のプロセスである process_two の標準入力に接続します。

ここで、subprocess.PIPEは、プロセスの標準出力を捕捉、つまり、プログラム内で利用可能にするために使用されます。これが設定されていない場合は、プロセスの出力はPythonを通じて利用することはできません。

>>> from subprocess import Popen, PIPE
>>> process_one = Popen(['dd', 'if=/dev/zero', 'bs=1', 'count=1'], stdout=PIPE)
>>> 1+0 records in
1+0 records out
1 byte copied, 3.3262e-05 s, 30.1 kB/s

>>> process_two = Popen(['wc'], stdin=process_one.stdout, stdout=PIPE, stderr=PIPE)
>>> outs, errs = process_two.communicate(timeout=15)
>>> outs.decode()
'      0       0       1\n'
>>> errs.decode()
''

あとは、process_twoに生えているcommunicate()を呼び出すことで、プロセス間の通信を開始します。process_twoの communicate() は、標準入力から、process_one の標準出力（と標準エラー出力）を、End-of-fileに達するまで読み出したあと、process_one が終了し、returncode属性を付けるのを確認するまで待ってから実行結果のタプルを返します。なお、タイムアウトが発生した場合、TimeoutExpired例外を送出するということで、例外が発生した場合の対処については、色々と気を付けなければならない点があるようなので、しっかりとリファレンスを読むようにしてください（免責）。

こうすることで、プロセス間で直接データをやり取りさせることができました。