子プロセスと標準入出力を介してやりとりする件

現代の将棋AIでは、GUI側と思考エンジン側とはUSIプロトコルというプロトコルに基づいてやりとりをするのが普通である。やねうら王ももちろん、このUSIプロトコルに対応している。

USIプロトコルとは
http://shogidokoro.starfree.jp/usi.html

子プロセスと標準入出力を介してやりとりする時の罠

USIプロトコルは、子プロセスを起動して、その子プロセスと標準入出力を介してやりとりを行う。

この方法は、わりと汎用性があってJenkinsのようなCIツールから呼び出されるスクリプトを書く時にも活用できる。

起動される側は、普通に標準入出力に対して(Pythonなら) readline() / print() するだけでやりとりができ、起動される側を書く人に特殊な通信の知識を要求しない。プログラミングの教科書の最初の方に出てくる標準入力からの入力と、標準出力への出力だけでやりとりができるので非常に読みやすいソースコードとなる。

ところが、一つだけ厄介な問題があって、print()する時にflushをしないと送信がなされないことがあるのだ。

今回はこのことについて詳しく書く。

Popenの仕様について

まず、状況としては、ホスト側(GUI側)は、PythonのPopenでスクリプト(思考エンジン)を起動しているものとする。この時に、標準入出力をリダイレクト(?)することによって、ここを介してやりとりしようとしている。

script = subprocess.Popen(script_path, # type:ignore
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE,
                        encoding="UTF-8",
                        # スクリプトの存在するフォルダをworking dirにしておかないと
                        # ファイル読み込みなどで変なところを参照してしまう。
                        cwd=script_dir)

script = subprocess.Popen(script_path, # type:ignore

stdin=subprocess.PIPE,

stdout=subprocess.PIPE,

encoding="UTF-8",

# スクリプトの存在するフォルダをworking dirにしておかないと

# ファイル読み込みなどで変なところを参照してしまう。

cwd=script_dir)

それで、こう書くと、ある程度熟練したプログラマーは、「標準出力って、(言語ランタイムが)バッファリングしてるからでしょ。標準出力遅いのが普通だから、どの言語でもバッファリングしてるのが普通じゃん。」みたいなことを言ってくる。

まあ、確かにファイルの読み書きみたいなのは、ランタイムがバッファリングしているのが普通だし、それゆえ、flushを呼び出さないとファイルに書き出されないというのも間違ってはいない。

ところが、今回の挙動は、どうもそうではなさそうなのだ。

例えば、次のようなスクリプトを書いた場合、Windows 11では、1秒ごとにprintしたもの(0,1,2,…,9)が出力される。

import time

print("Hello Script World!!")

for i in range(10):
    print(i)
    time.sleep(1)

import time

print("Hello Script World!!")

for i in range(10):

print(i)

time.sleep(1)

ところがWindows 10だと、このスクリプトが終了するタイミングで一気に出力されるのだ。

つまりこれは言語ランタイムによるバッファリングではなく、OSによるバッファリングなのであろう。とは言え、Windows 10の場合であっても、バッファは有限であろうから、flushを呼び出さなくともどこかの時点でflushはなされるはずである。

Windows 11の場合もflushを呼び出さなくていいと言うよりは、OSがおそらく一定時間ごとにflushみたいなことをしているだけで、即時flushが保証されているかは怪しい。

いずれにせよ、スクリプト側は(ホスト側に即時で受け取って欲しいならば)printごとにflushを呼び出すべきである。

flushの呼び出しについて

それで、このflushをどうやって行うのが良いかと言う話になる。

Python 3.3以降では、

print(i, flush=True)

1	print(i, flush=True)

のように書くことができる。

いまPythonは最新が3.11.4なので3.3以前のバージョンなんて使ってる人いないだろうけど、Python 2系は使っている人がいるかも知れないので、その場合どうすれば良いのかも書いておくと、

import sys

print ""
sys.stdout.flush()

import sys

print ""

sys.stdout.flush()

のようにsys.stdout.flush()を呼び出すとよろしい。

しかし、printのごとにflush=Trueなんて書くのは嫌だから、普通は↓のように何らかの関数を用意して、これを呼び出す

def send_to_host(s:string):
    print(s, flush=True)

1 2	def send_to_host(s:string): print(s, flush=True)

でもこれだけだと、

print(1,2)

1	print(1,2)

のように複数の引数を取ったりできない。不便である。

そこで、functools.partialを使うというテクニックがある。

from functools import partial

print_flushed = partial(print, flush=True)
print_flushed("Hello world!")

from functools import partial

print_flushed = partial(print, flush=True)

print_flushed("Hello world!")

これでもいいのだが、あんまりPythonicなコードではない気もするし、functoolsなんか持ち出さなくとも、decorator(decorateした関数を返す関数)パターンで十分だと言う話もある。

def print_decorator(func):
  printer = func
  def wrapped(*args):
    printer(*args, flush=True)
  return wrapped

print = print_decorator(print)

print(1, 2)

def print_decorator(func):

printer = func

def wrapped(*args):

printer(*args, flush=True)

return wrapped

print = print_decorator(print)

print(1, 2)

これで(さきほどはできなかった) print(1,2) のようなことができる。

■　ホスト側とのパラメーターのやりとりについて

USIプロトコルでは、ホスト側からエンジンオプションは、
setoption …
のようなコマンドを送ることになっているのだが、もし新たにこういうプロトコルを作るとしたら、この仕様はあまりいけてるとは言い難い。

例えば、Pythonであれば、json.dumps()で、json形式にできる。これを送るのがお手軽である。json.dumps()で出力した文字列には改行は含まれないので、これは1行に収まるし、json.load()で復元できる。

やってみよう。

def send_to_script(message:str):
  '''スクリプトにメッセージを送信する。'''
  # 改行文字列を付与しないと改行が送れない。
  script.stdin.write(message + '\n') # type:ignore
  # flushしないと届かないので注意。
  script.stdin.flush() # type:ignore

# json.dumps()は1行にserializeされるので都合がいい。
# これならスクリプト側でinput()で一発で読み込める。

send_to_script(json.dumps(json_obj))

def send_to_script(message:str):

'''スクリプトにメッセージを送信する。'''

# 改行文字列を付与しないと改行が送れない。

script.stdin.write(message + '\n') # type:ignore

# flushしないと届かないので注意。

script.stdin.flush() # type:ignore

# json.dumps()は1行にserializeされるので都合がいい。

# これならスクリプト側でinput()で一発で読み込める。

send_to_script(json.dumps(json_obj))

これに対して、スクリプト側からは、以下のように書くだけでこのパラメーターの受け渡しが完了する。

import json
# ホスト側から渡されたパラメーターのdeserialize
params = json.loads(input())
# 渡されたパラメーター
path,line = params&#91;'path'],params&#91;'line']

import json

# ホスト側から渡されたパラメーターのdeserialize

params = json.loads(input())

# 渡されたパラメーター

path,line = params['path'],params['line']

非常に簡単に実装できた。

まとめ

今回は将棋AIで標準的に採用されているUSIプロトコルの、「子プロセスと標準入出力を介してやりとりする」という部分に触れ、子プロセスとして起動されるプログラム側ではprint()ごとのflushが必要であることと、そのflushをどうやって書けばいいかについて詳細に解説した。

「子プロセスと標準入出力を介してやりとりする」ことで、スクリプト側をホスト側とは別の言語で開発できて大変便利なので覚えておくと将棋AI開発以外でも役に立つシーンも多いと思う。