一覧OS-05-3 · プロセスとスレッド

パイプによるプロセス間通信

プロセス間でデータを送り、書込み端が残る場合と全部閉じた場合の違いを確かめます。

説明

パイプは一方向のバイト列を受け渡す仕組みです。書込み端へ送ったデータを読取り端から取り出します。シェルの縦棒も、前のコマンドの標準出力と次のコマンドの標準入力をこの仕組みで接続します。

Pythonから、受け取った行を大文字にして返す子プロセスを起動します。同じ子へ続けて送り、入力を閉じて終了を待ちます。その後、端点の複製を使って「空だが書込み可能なパイプ」と「書込み端がすべて閉じたパイプ」を比べます。

1変換する子プロセスを起動

Popenで子の標準入力と標準出力をパイプへ接続します。子は行を受け取るたびに大文字へ変換し、flushで出力バッファを送り出します。

import subprocess, sysプロセス操作; code = 'import sys\nfor line in sys.stdin:\n sys.stdout.write(line.upper()); sys.stdout.flush()'行単位で大文字へ変換; p = subprocess.Popen([sys.executable, '-c', code], stdin=subprocess.PIPE, stdout=subprocess.PIPE)子と二つのパイプを作る; print(p.poll() is None)実行中か

Trueは子がまだ終了していないことを示します。親から子への入力用と、子から親への出力用の二つのパイプがあります。一つのパイプが双方向通信をしているわけではありません。

21行を送って返答を読む

abと改行を送信し、親側もflushしてから子の出力を1行読みます。

p.stdin.write(b'ab\n')入力へ3バイト送る; p.stdin.flush()親のバッファを送出; print(p.stdout.readline())子の1行を受信

書込みは3バイト、返答はABと改行です。行の区切りは子のreadline相当の処理が解釈しています。パイプ自体は文字コードや行という構造を持たないバイトストリームです。

3同じ子へ続けて送信

次は二つの行を一回のwriteで送ります。受信は二回のreadlineに分けます。

p.stdin.write(b'cd\nef\n')二行をまとめて送る; p.stdin.flush()バッファを送出; print(p.stdout.readline(), p.stdout.readline())二行を順に受信

6バイトの一回の書込みに対して、読み手はCDとEFを別々の行として受け取っています。書込み回数と読取り回数は対応しません。複数の送信者が使う場合にはPIPE_BUF以下の書込みの原子性など、追加の条件も関係します。

4入力を閉じて子の終了を待つ

親が子の入力への書込み端を閉じると、子は残りを読み終えた後にEOFへ達します。子の終了によって返答側の書込み端も閉じます。

p.stdin.close()入力の書込み端を閉じる; print(p.stdout.read())返答をEOFまで読む; print(p.wait())終了を回収; p.stdout.close()読取り端を閉じる

返答は空のバイト列、終了コードは0です。ここでは未読の返答が残っていないため空になります。バッファにデータが残っていれば、そのデータを受け取ってからEOFへ到達します。

5書込み端の複製を作る

次は端点の状態に注目します。os.pipeで空のパイプを作り、書込み端をdupで複製します。読取りは非ブロッキングにして、待ちになる条件を例外として確認します。

import os端点の操作; r, w = os.pipe()読取り端と書込み端; extra = os.dup(w)書込み端の参照を追加; os.set_blocking(r, False)待たずに戻る設定; os.close(w)元の書込み記述子を閉じる; print('one write descriptor remains')残る参照を示す

元のwを閉じても、複製extraが同じ書込み端を参照しています。forkで引き継いだ端点を不要なプロセスで閉じ忘れた場合も、参照が残ってEOFに達しない原因になります。

6空でもEOFではない場合

データはありませんが書込み端が一つ残っています。読取りを試し、BlockingIOErrorを表示します。

try:読取りを試す

print(os.read(r, 8))最大8バイト読む

except BlockingIOError:待ちになる条件

print('would block: writer remains')書込み端が残る状態

would blockとなります。ブロッキング読取りなら、データ到着か最後の書込み端の閉鎖などを待つ状態です。現時点でデータが0バイトということと、将来も書き手がいないEOFは異なります。

7最後の書込み端を閉じる

残っていたextraを閉じ、同じ読取りを試します。

os.close(extra)最後の書込み端を閉じる; print(os.read(r, 8))同じ読取り; os.close(r)読取り端も解放

今度はb''です。書込み端への参照がなく、未読データもないためEOFになります。空のバイト列はこの条件での正常な読取り結果で、先ほどのBlockingIOErrorとは意味が違います。

8受信側が閉じた後の書込み

新しいパイプの読取り端を閉じ、その後の書込みを試します。Pythonは通常SIGPIPEを無視するため、失敗をBrokenPipeErrorとして扱えます。

r, w = os.pipe()新しいパイプ; os.close(r)受信端を閉じる

読取り端がなくなりました。次の書込みでは、送り先が存在しない場合のエラーを確認します。

9書込み失敗を処理して端点を解放

読取り端が一つもない状態で1バイトを書き込みます。最後に書込み端も閉じます。

try:書込みを試す

os.write(w, b'x')1バイト送る

except BrokenPipeError:受信端がないエラー

print('broken pipe: no readers')原因を表示

finally:成否にかかわらず

os.close(w)書込み端を解放

broken pipeとなります。一般のUnixプログラムではSIGPIPEの既定動作によって終了する場合があります。読取り側のEOFと書込み側のEPIPE・SIGPIPEは、どちらの端点がなくなったかによって起こる別の結果です。

まとめ

パイプはバイトストリームであり、writeとreadの回数や行の境界は一致するとは限りません。

EOFには未読データがなく、書込み端への参照がすべて閉じていることが必要です。

読取り端がすべて閉じた後の書込みはEPIPEやSIGPIPEの対象になります。

操作履歴と解説を振り返れます。

Python 3

import subprocess, sys; code = 'import sys\nfor line in sys.stdin:\n sys.stdout.write(line.upper()); sys.stdout.flush()'; p = subprocess.Popen([sys.executable, '-c', code], stdin=subprocess.PIPE, stdout=subprocess.PIPE); print(p.poll() is None)

True

p.stdin.write(b'ab\n'); p.stdin.flush(); print(p.stdout.readline())

3
b'AB\n'

p.stdin.write(b'cd\nef\n'); p.stdin.flush(); print(p.stdout.readline(), p.stdout.readline())

6
b'CD\n' b'EF\n'

p.stdin.close(); print(p.stdout.read()); print(p.wait()); p.stdout.close()

b''
0

import os; r, w = os.pipe(); extra = os.dup(w); os.set_blocking(r, False); os.close(w); print('one write descriptor remains')

one write descriptor remains

try:
 print(os.read(r, 8))
except BlockingIOError:
 print('would block: writer remains')

would block: writer remains

os.close(extra); print(os.read(r, 8)); os.close(r)

b''

r, w = os.pipe(); os.close(r)

try:
 os.write(w, b'x')
except BrokenPipeError:
 print('broken pipe: no readers')
finally:
 os.close(w)

broken pipe: no readers

演習完了