subprocessで生成したプロセスの実行結果が上手く出力されません。

Question

###環境
python3.8.1
OS : Windows10

###実現しようとしたこと
pythonの標準ライブラリであるsubprocessモジュールを使用し、プロセスを生成したのちに
値を渡し、渡された値に応じてプロセスが適切な結果を返すことを確認しようとしておりました。
プロセスを生成する側のコード(test.py)と実行されるプロセスのコード(process_router.py)
は以下の通りです。

#test.py
class CallProcessRouter():

    def __init__(self):
        self.BASEDIR = os.path.abspath( os.path.dirname( __file__ ) )
        #サブプロセスを開く
        cmd = ['python', self.BASEDIR+'/../src/process_router.py']
        self.proc = subprocess.Popen(cmd, 
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE
                        )
    
    def drive_process_router(self, arg):
        self.proc.stdin.write(arg.encode())
        self.proc.stdin.write('\n'.encode())
        self.proc.stdin.flush()
        output = self.proc.stdout.readline().strip()
        stderr = self.proc.stderr.readline().strip()

if __name__=="__main__":
    a = CallProcessRouter()
    path = "~\xxx.json" #json file path
    a.drive_process_router(path)

#process_router.py
while True:
    try:
        param = sys.stdin.readline().strip()
        if param == '':
            break
        #中略
        #処理した結果の出力
        output = {"result" : True}
        output_str = json.dumps(output)
        print(output_str)
    except:
        #When error occured, process_router.py send error message as json normal output.
        output = {"result":False,
                    "data":{
                        "error_message":traceback.format_exc(),
                    }
                }
        print(json.dumps(output))

###問題点
test.pyを実行したところ、process_router.pyには繋がるのですが、
print文で出力した結果がtest.pyのstdout.readline().strip()で
受け取れていないようなのです。
process_router.pyは親プロセスによってkillされない限り動き続けるため、
値を受け取れないとそこで処理がブロックされ、いつまでも処理が終わらない状態と
なります。

###試したこと

• readline()による入力は\nで区切られるため、print出力を行うoutput_strに

直接\nを付加したのですが結果は変わりませんでした。

• subprocess.Popenの出力がバイナリになっている可能性を踏まえ

output = self.proc.stdout.readline().decode().strip()
と変更したのですが上手く値を受け取ることが出来ませんでした。

• process生成時のコマンドを

cmd = 'python '+self.BASEDIR+'/../src/process_router.py'
に変更したところ、何故か次のエラーを吐きました。原因はわかっておりません。

File "test.py", line 56, in drive_process_router
    self.proc.stdin.flush()
OSError: [Errno 22] Invalid argument

• Linux(Amazon Linux)環境下でも実行しましたが、Windowsと同じ状況でした。

どのような意見でもよいので、ご教授よろしくお願いします。

Answer 1

コードの全貌が解らないので、他にも原因があるかもしれませんが
適切なタイミングで flush でどうでしょう。win10/py3.8.0で動作確認済。

# process_router.pyのJSON出力後

print(output_str)

sys.stdout.flush()
sys.stderr.flush()

stderr も必要な理由は、読み込み側が必要としているからです。

output = self.proc.stdout.readline().strip()
stderr = self.proc.stderr.readline().strip()

JSONのメッセージが1行に収まらない・複数行となってくると
読込<=>書出のタイミングで同期をとる為に、何らかの工夫をする必要があります。

---

他の方法: subprocessベースのもの。

非同期に扱う場合は、asyncio.subprocess 等
追加で少し設定が必要ですけど、win10でも動作確認。

subprocess 経由のRPCでしたらこんなのもあります。(同期/非同期対応)
https://pypi.org/project/jsonrpyc/

Comments

[teamikl]

(後から気がついた)otn さんの回答とは、コードが違うだけで同じ内容でした。
問題の現象を確認の後、stdout/stderrのflushで修正を確認できたので、
一応、動作報告まで。

[onomu_]

ご回答ありがとうございます。

>JSONのメッセージが1行に収まらない・複数行となってくると
>読込<=>書出のタイミングで同期をとる為に、何らかの工夫をする必要があります。
JSONの出力結果は、12000字程度あるため、1回の送信でパイプバッファがオーバーしてしまっている可能性は十分に考えられます。
flush()ではうまくいかなかったため、ほかの方法を検討する必要はあると感じました。

他の方法も含めて情報を提供してくださり感謝しています。

Answer 2

process_router.py側の出力を、下記の様にしてはどうでしょうか。except句も同様。

        print(output_str,flush=True)
        print("",file=sys.stderr,flush=True)

Comments

[onomu_]

教えてくださりありがとうございます。
修正してもうまく動作しなかったのですが、printの強制出力は知らなかったため
参考になりました。

[otn]

readlineが完了しないと言う事は、まだprintが実行されてないからですね。

Answer 3

現状での解決策を記述します。
subprocessモジュールの公式ドキュメントには次のような記述があります。

Warning Use communicate() rather than .stdin.write, .stdout.read or .stderr.read to avoid deadlocks due to any of the other OS pipe buffers filling up and blocking the child process.

~~Windowsでは、常に何かしらのプロセスがバックグラウンドで稼働しているため、そのどれかとデットロック
を起こしてしまっている可能性が高いです。
~~
出力されるjson文字列が膨大であるために、パイプバッファが一杯になってしまった可能性が高いです。

ドキュメントにも書いてある通り、communicate()が使用できる状況下ではstdin.writeではなくてcommunicate()を使いましょう。
ただし、communicate()は子プロセスが終了するまで結果を返しません。
上記のドキュメントのようにwhile Trueで常に稼働させているプロセスの場合、communicate()は
使用できないのでご注意ください。

上記のプログラムのように、プロセス起動時のオーバーヘッドをなくすためにプロセスを常駐させる必要が
ある場合、現状ではpythonではなく他の言語で記述した方がいいのかもしれません。
私の場合、Node.jsのPython Shellを用いることでうまく接続することが出来ました。

ただし、何となくPythonでも解決策を見いだせないと気持ち悪い気もするので、
もっとよい方法があるとか、言ってることが違うと思う方はどしどし意見お願いいたします。
以上です。

Comments

[otn]

> Windowsでは、常に何かしらのプロセスがバックグラウンドで稼働しているため、そのどれかとデットロックを起こしてしまっている可能性が高いです。

は誤読ですね。他のプロセスは関係ないです。

Popen.wait のところにも同じような注釈がありますが、子プロセス起動後waitでその終了を待つと、子プロセス側で大量にパイプバッファに書いている場合writeがブロックされて、子は親がそのパイプから読まないと終われませんが、親がパイプを読まずにwaitしちゃうとデッドロックになるということを言ってると思います。

あるいは、往復2本のパイプで通信し合う2つのプロセスが、双方ともパイプを読まずに書き込みだけ続ければ、パイプバッファが一杯になって両プロセスともそれ以上の書き込みがブロックされることを言ってるのかも。

常駐プロセス同士で適宜通信するには、ノンブロッッキングの読み書きを行うか、selectで今読み書き可能な状態かどうかを調べた上で読み書きすれば、デッドロックは起こりませんが、どちらもWindowsだと出来ないかも知れません。

[onomu_]

subprocessのパイプバッファについて勘違いしておりました。
ご指摘ありがとうございます。

subprocessベースのものがいくつかあるようなので、時間があれば試してみようと思っております。

Answer 4

そもそもreadline()ではなにが取れてるのでしょう。それを確認することですね

Comments

[onomu_]

test.pyのreadline()で値が受け取れていないのです。process_router.pyのアウトプットまでは正常に動作しているのですが、print文出力~stdout.readline().strip()までの経路のどこかで処理がブロックされてしまっている状態です。

[y_waiwai]

> #process_router.py
while True:
try:
param = sys.stdin.readline().strip()

このreadline() にはなにが帰ってくるつもりですか？実際にはなにが帰ってきてますか？

[onomu_]

ファイルのパスが返ってきます。(C:\Users\~\xxx.json)
その後のパスに応じてファイルを読み込む処理は上手くいっているので、ここは問題ではないのだろうと思っております。