Q&A
以下のドキュメントが参考になるかもしれません。(もしくは ProcessPoolExecutor)
ThreadPoolExecutor の例
https://docs.python.org/ja/3/library/concurrent.futures.html#threadpoolexecutor-example
実現したいこと
外部のexeファイルと文字列の入出力のやり取りをするpythonのコードにconcurrent.futuresを使えるようにして関数をマルチスレッド化したい。スレッドの最大数を制限し、終了したスレッドが発生次第スレッドをまた動かすことを繰り返したいです。
前提
subprocess.Popenを用いて、外部のexeファイルと文字列の入出力のやり取りをするプログラムを書いています。外部のexeは詰将棋を解答するプログラム（脊尾詰）で、こちらのURLでダウンロードしたものをそのまま使っています。
http://panashogi.web.fc2.com/seotsume.html
私が書いているexeファイルは、usiプロトコルに従って文字列の入出力を行います。
http://shogidokoro.starfree.jp/usi.html#ProblemExample
参考URLの要点だけを説明すると、私のプログラムで行うexeへの入出力とその説明は以下の通りです。
usi\n：詰将棋エンジン起動。これをexeへ入力すると、usiokを含む文字列がexeから出力される
setoption name USI_Hash value 512\n：詰将棋エンジンに関する設定を変更
setoption name Do_YoTsume_Search value false\n：詰将棋エンジンに関する設定を変更
isready\n：詰将棋解答開始前の準備。これをexeへ入力すると、readyokを含む文字列がexeから出力される
usinewgame\n：これをexeへ入力することで、exeは詰将棋盤面の文字列の入力待ち状態に移行する。
position sfen…\n：詰将棋の盤面をexeへ入力。
go mate infinite\n：これをexeへ入力することで、exeは入力された詰将棋を解答し始める。結果が出ると'checkmate'を含む文字列を出力する。終了すると盤面の入力待ち状態に戻り、再び"position sfen..."コマンドを待つ状態になる。
私のコードでは、1つ目の関数(funcX)でusinewgame\nまでexeに入力してexeを盤面の入力待ち状態に移行させた後、2つ目の関数(funcY)で詰将棋の盤面をexeに入力して解答結果を得る、という構成になっていて、1回だけfuncXを実行してその後はfuncYを繰り返し実行しています。スレッドの最大数を制限しつつ、funcXとfuncYをマルチスレッド化(orマルチプロセス化)し、複数の詰将棋を複数の詰将棋解答exeに入力して、解答が終了したものから解答結果を得ていくようにしたいです。あるスレッドで関数Yが終了したら、スレッドをまた動かして別の盤面を指定したfuncYを実行して…ということをやりたいです。
Python
1import subprocess 2import keyboard 3import concurrent.futures 4 5def main(): 6 #list1は、詰将棋の盤面を表現した文字列を集めてリストにしたもの。非本質なので気にしないでください 7 list1=['position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b R2G2Nr2b2g4s3l15p 1\n', 8 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b RGS2NPr2b3g3s3l14p 1\n', 9 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b RSNLr2b4g3sn2l15p 1\n', 10 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b RSNPr2b4g3sn3l14p 1\n', 11 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b RSN2Pr2b4g3sn3l13p 1\n', 12 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b RBSNrb4g3sn3l15p 1\n', 13 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b RBSNLrb4g3sn2l15p 1\n', 14 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b RBGNrb3g4sn3l15p 1\n', 15 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b R2BNr4g4sn3l15p 1\n', 16 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b B2GS2rb2g3s2n3l15p 1\n', 17 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b BG2SL2rb3g2s2n2l15p 1\n', 18 'position sfen 7nl/7k1/9/5Np1p/9/5+P3/9/9/9 b 2B3S2L2r4gs2nl15p 1\n', 19 'position sfen 2n3p2/5+r1p1/7+pk/3p5/3+p5/9/9/6L2/9 b R4GNP2b4s2n3l12p 1\n'] 20 21 file_to_path1="SeoTsume_1.exe"#exeファイルのパス 22 file_to_path2="SeoTsume_2.exe"#exeファイルのパス2 23 proc1=subprocess.Popen(file_to_path1,stdin=subprocess.PIPE,stdout=subprocess.PIPE)#サブプロセスを起動しておく。exeが起動される 24 proc2=subprocess.Popen(file_to_path2,stdin=subprocess.PIPE,stdout=subprocess.PIPE)#マルチスレッド用にexeをコピーした別のexeを用意して起動。 25 26 funcX(proc1)#funcXは最初に一度だけやっておかなければならない入出力のやり取りの関数 27 funcX(proc2) 28 num=0 29 while(num<len(list1)-1): 30 with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:#マルチスレッド化したい 31 executor.submit(funcY,proc1,list1[num]) #funcYの引数として、funcXを介して盤面の入力待ち状態になったPopenオブジェクトと、盤面の文字列list1[num]を指定する 32 executor.submit(funcY,proc2,list1[num+1]) 33 num+=2 34 35def funcX(proc):#exeファイルを盤面の入力待ち状態に移行させるために最初に一度だけ実行しておく関数 36 proc.stdin.write(('usi\n').encode())#pyからexeファイルへ文字列"usi\n"を入力 37 proc.stdin.flush()#flushは必須 38 39 while True: 40 line = proc.stdout.readline().rstrip().decode('utf-8')#exeからpyへ文字列出力を1行ずつ受け取る 41 if line == 'usiok':#exeからpyへ"usiok"という文字列が出力された場合、出力の受け取りを中断する。 42 print(line) 43 break 44 45 proc.stdin.write(('setoption name USI_Hash value 512\n').encode())#pyからexeファイルへ文字列を入力。 46 proc.stdin.flush() 47 48 proc.stdin.write(('setoption name Do_YoTsume_Search value false\n').encode())#pyからexeファイルへ文字列を入力。 49 proc.stdin.flush() 50 51 proc.stdin.write(('isready\n').encode())#pyからexeファイルへ文字列を入力 52 proc.stdin.flush() 53 54 while True: 55 line = proc.stdout.readline().rstrip().decode('utf-8')#exeからpyへ文字列出力を1行ずつ受け取る 56 if line == 'readyok':#exeからpyへ"readyok"という文字列が出力された場合、出力の受け取りを中断する 57 print(line) 58 break 59 60 proc.stdin.write(('usinewgame\n').encode())#pyからexeファイルへ文字列を入力。ここで、exeファイルが詰将棋の盤面入力待ち状態に移行する。 61 proc.stdin.flush() 62 63 64def funcY(proc,input_cmd):#funcXによって盤面の入力待ち状態に移行したexeファイルに対して、実際に盤面を入力し、exeから解答結果を得る 65 print(input_cmd) 66 proc.stdin.write(input_cmd.encode())#pyからexeファイルへ文字列を入力。input_cmdは詰将棋の盤面。これをexeに投げる。 67 proc.stdin.flush() 68 69 proc.stdin.write(('go mate infinite\n').encode())#pyからexeファイルへ文字列を入力。これによりexeが詰将棋を解き始める。 70 proc.stdin.flush() 71 72 while True:#exeが詰将棋解答中。 73 line = proc.stdout.readline().rstrip().decode('utf-8')#exeからpyへ文字列出力を1行ずつ取る 74 print(line) 75 if line[:9]=='checkmate':#exeからpyへ"checkmate"を含む文字列が出力された=詰将棋の解答結果が出たという意味。出力の受け取りを中断し、解答結果を得る 76 result=line 77 # print(line) 78 break 79 if keyboard.is_pressed("escape"):#キーボード操作でいつでも中断できる 80 break 81 82 return result#詰将棋を解答した結果を返す 83 84if __name__ == "__main__": 85 main()
発生している問題・エラーメッセージ
上に示すコードはfuncXは普通に実行し、funcYだけマルチスレッド化したコードです。funcYの内部にprint文を入れて出力を観察しましたが、funcXが終わった後はfuncY内部のprint文は一切反応せず、少し間があいてプログラムが終了します。
concurrent.futuresを使用しない普通のシングルスレッドのコードなら普通に動作することから、executor.submit()が動作してなさそうです。submit()の引数としてPopenオブジェクトを指定できないのではないかと考えています。
usiok readyok usiok readyok #本来はここからfuncYでの実行結果が表示されるはずだが…
あり得る疑問点
A.funcXとfuncYを分離させている意味、funcXを1回だけ実行してfuncYを繰り返す処理を行う意味
Q.
funcXは、exeをただの起動状態から盤面の入力待ち状態にする関数。funcYは、盤面の入力待機状態のexeに盤面を入力して解答させ、exeからの結果を待つ関数。
なのでfuncYはfuncXの後でないと動かない。また、上記した通り、funcYを1度実行した後はexeは盤面の入力待機状態に戻るので、funcYを実行した後はfuncXを実行し直さなくてよい。
無視できない程度に時間がかかり、実行回数は最低限にしたいfuncXとfuncYを纏めて繰り返すことはしたくないから、funcXとfuncYを分離させ、funcYだけ繰り返すようにした。
A.funcYの引数として、proc=Popenオブジェクトを指定する理由
Q.
funcXを介した後のPopenオブジェクトの状態（exeの盤面入力待ち状態）を維持したままfuncYに移りたいので、Popenオブジェクトを指定している。
例えば、funcYの引数としてexeのファイルパスを渡し、funcY内でsubprocess.Popen(<path to exe>,stdin=subprocess.PIPE,stdout=subprocess.PIPE)としてexeを起動する方法もあるが、それだとfuncY内で起動したexeをただの起動状態から入力待ち状態に移行させる必要性が出てくる。
それはfuncXとfuncYを纏めて繰り返す処理と同じだが、前述の理由でそれはしたくないです。
補足情報（FW/ツールのバージョンなど）
Windows11環境です。
ありがとうございます。concurrent.futuresを使った方が単純になのですが、並行処理をしたい関数の引数としてproc=subprocess.Popen()が入っていて、executor.submit()がprocを指定できないので困っています。リストにprocを入れても同様です。そこで仕方なくthreadingモジュールを使用している感じですね…
そういう経緯があるのであれば、質問に明記したほうが回答が付きやすいと思います。
「引数としてproc=subprocess.Popen()が入っていて」というのはちょっとあまり見ないやりかただと思いますが、サブプロセスの起動は起動先のスレッド/プロセスで行なってはだめなのでしょうか？
説明不足で申し訳ありません。その点を先に質問するべきでした。
procを使って外部のexeファイルと入出力のやり取りをする関数が2つありまして、それぞれfuncX、funcYとして説明申し上げます。2つとも現状proc=subprocess.Popen(<path to exe>,stdin=subprocess.PIPE,stdout=subprocess.PIPE)を引数に指定しています。
具体的には、
proc.stdin.write(('cmd1').encode())
proc.stdin.flush()
の2行でpythonからcmd1という文字列を外部のexeファイルに入力し、
response1=proc.stdout.readline().rstrip().decode('utf-8')
で外部のexeファイルからpythonへ文字列response1を出力しています。
funcXは最初に1度だけ実行する関数であり、funcYは1回だけ行うfuncXの後でないと動かない且つfor文で何度も実行する関数です。
プログラムの流れは、大まかにfuncX→funcY→funcY→funcY→…です。
funcXやfuncYの引数としては、procではなく、<path to exe>を指定することで、funcXやfuncY内部でsubprocess.Popenを呼んでサブプロセスを起動しようとしましたが、funcYで起動したときにfuncXで起動して入出力をやり取りしたprocの状態がリセットされてしまい、動いてくれないのです…funcXは実行に少し時間がかかるので、「funcYで入出力のやり取りをする度に毎回その前にfuncXで入出力のやり取りをやっておく」ようなことはしたくなく、funcXの実行は最初の1回だけにしたい。
そこで、関数の外でsubprocess.Popenを呼んで予めサブプロセスを起動しておいて、procをfuncXに投げ、続いてfuncXでやり取りをした状態のprocをfuncYに投げることで無理矢理動かすようにしたという感じです。このあたり、上手く実装する方法があると思います。
長々とすみません。
重要な話だし長いので、質問を編集して追記したほうがいいですね。ここは質問へのコメントの欄なので。
状況からすると、対象となるプロセスを起動して、 funcX→funcY→funcY→funcY→ は1つの流れで順に処理する必要があるように理解したので、これを1つずづ別のプロセスにしたい理由がわかりません。
もしかして、 funcY の出力結果を処理が終る度に得たいというのが理由ですか？
上のコメント読んだ限り、ThreadPoolExecutor を使って、submit に proc を渡せばいいように思うのですが、それがダメな理由がイマイチわかりにくいです。
>TakaiYさん
そうですね。質問の内容を変更して立て直そうと思います。
>bsdfanさん
説明不足ですみません。procはPopenオブジェクトです。実行したい関数funcYの引数にprocが入っている場合、executor.submit(funcY,proc)のような書き方をすると思うのですが、どうやらsubmitメソッドはPopenオブジェクトを認識してくれないようで、funcYの処理が全く行われません。[proc]としてリストにして、funcY内でprocを取り出してもダメでした。しかし、一方でthreadingモジュールを用いた場合、threading.Thread(target=funcY,args=(proc,))としてPopenオブジェクトを認識したため、threadingを使うことを考えていました。
「並行処理をしたい関数の引数としてproc=subprocess.Popen()が入っている」で、なぜそうしたいのか? が
「どうやらsubmitメソッドはPopenオブジェクトを認識してくれないようで、funcYの処理が全く行われません」で、どう書いてどうなったのか?
が他人に伝わってないと思いました。
そのせいで「threading.Thread(target=funcY,args=(proc,))としてPopenオブジェクトを認識したため、threadingを使う」という判断になぜ? となっている感じかと。
「立て直そうと思います」とありますが、別の質問を作るのではなく、この質問の内容を編集したほうがいいと思います。 これまでの経緯も残りますので。
>quickquipさん
質問の内容を編集して、コードを提示しながら詳しく説明しようかと思います。
>TakaiYさん
確かにそうですね。ただ、大幅に書き直しても大丈夫でしょうかね…teratail初心者なこともあり、その点が不安です。
コメント欄でのやり取りを見るに、
(1)Threadingではなくconcurrent.futuresを使った方が所望のコード（所望の内容は、現在の私の質問の内容）が実現しやすくなる
(2)ただし、マルチスレッド化したい関数の引数としてPopenオブジェクトを使っているため、concurrent.futuresを使うことができない
(3)(2)ゆえマルチスレッド化したい関数の引数としてPopenオブジェクトを使わないで済むようにしたいが、その方法が自分では分からない。したがってconcurrent.futuresを使うことができないでいる。
(1)～(3)についてより詳しく実際に書いたコードを提示しながら説明しつつ、そのコードをconcurrent.futuresを使えるように書き替えられる方法を質問の内容の趣旨にしようと考えているのですが、ここまで大きく変更してしまって大丈夫かなと。
質問をするのは問題を解決するためですが、XY問題といって、根本の問題を解決するのではない方法を質問してしまうといことはよくあることで、本当に問題になっているのは何なのかが明確になったのであれば、直すのもありだと思います。
また、この質問そのものは、記録として取っておきたいということであれば、解決済みとして閉じて、別のを立ててもいいかもしません。
ちなみに、僕はまだ、この問題をマルチスレッド/プロセスで解決する必要があるかどうかに疑問を持っています。
funcX→funcY→funcY→funcY ... のように順に処理したいということだと思うのですが、これをそれぞれのfuncX,Y を別のスレッド/プロセスで実行したい理由は何ですか？
>TakaiYさん
本質問を大きく修正しようかと思います。
マルチスレッド化したい理由は、funcXは無視できない程度に時間がかかる処理で、funcYは時間がかかる処理だからです。このあたり、プログラムの仕様を実際に書くべきだと感じましたので、修正後の質問に説明を加えます。色々と申し訳ございません。
編集後の質問を見ました。
結局、1つの詰将棋エンジンは、初期化->入力待ち->詰将棋問題入力->回答出力->入力待ち...、という流れでしか動作できないのですから、(FuncX実行)->(FuncY実行を繰替えし) というプログラムを複数スレッド/プロセスで実行すればいいだけではないかと思うのですが、どうでしょう？ であれば、起動したプロセスをスレッド/プロセスに渡す必要は無いのではないですかね？
分かりにくい質問ですみません。教えていただきありがとうございます。
FuncXとwhile文入りのFuncYを纏めて一つの関数にしてしまうということでしょうか？
そうです。
で、必要であればそれを並列で複数動かすことができるし FuncYが終るたびに結果を取り出す方法もいろいろあります。
気になる点としては、
・ProcessPoolExecutor では、ThreadPoolExecutor と違って、pickle 化できるオブジェクトしかやりとりできないはずなので、Popenオブジェクトは渡せないと思います。
・while ループの中で ProcessPoolExecutor を作っているので、2つを平行で走らせて両方が終わったら、新しいPoolを作って次の2つを実行、という繰り返しになっています。たぶんやりたいことは、Pool をひとつ作ってからループでsubmitなのでは？
・(上記が解決できたとしても)今の方法だと、同じPopenオブジェクトを2つのスレッド(プロセス)から同時に使ってしまうケースが発生して、おかしくなります。スレッド(プロセス)毎にPopenオブジェクトが固定されるような設計にしたほうがすっきりすると思います。
→ 結論としては、上でTakaiYさんが書かれているやり方がいいと思います。
>TakaiYさん
>bsdfanさん
詳しく教えていただきありがとうございます。
ThreadPoolExecutorだとPopenオブジェクトは渡せるには渡せるんですが、お書きした通り全く機能しないようです。ProcessPoolExecutorを一応試しましたが、こちらは明確なエラーが出ました。
>2つを平行で走らせて両方が終わったら、新しいPoolを作って次の2つを実行、という繰り返しになっています。
はい、現在そういう設計になっています。やりたいことは、終了したものには次の盤面を与えて走らせるような処理で、並行で走らせる数を（この場合だと）常に2にしておくようなイメージです。
>Pool をひとつ作ってからループでsubmitなのでは？
すみません、どういう意味か教えていただけませんか。
実は、個人的にclassを使う方法を思いつき、一応マルチスレッドで動きはしました。
コンストラクタのdef __init__の引数としてselfとexeのパスを指定しておき、そこでパスを使ってPopenオブジェクトを作成し、funcXとfuncYもclassの中に格納しておきます(funcXの引数はself、funcYの引数はselfとinput_cmd)
そして、インスタンスを立ち上げて、インスタンス.funcXや、インスタンス.funcYをsubmit()の引数に指定しました。これなら、Popenオブジェクトの代わりにexeのパスを引数に指定しなくて良さそうです。
ただ、気になる点としては、classのメソッドもマルチスレッド化orマルチプロセスするにあたって通常の関数と同様に扱ってよいのでしょうか？
方向性が違うようなので、僕のコメントはこれで終りにします。
クラスのメソッドも関数も機能は同じです。なので、関数でできないことはメソッドでもできませんし、メソッドでできることは関数でもできます。
> すみません、どういう意味か教えていただけませんか。
「スレッドの最大数を制限」にはなっていますが、「終了したスレッドが発生次第スレッドをまた動かすことを繰り返したい」にはなっていないという意味です。
また、手元の適当なexeで試してみましたが、ThreadPoolExecutor の submit に、Popen のオブジェクトを渡しても、ちゃんと動きました。
なので、質問者さんのやり方になにか問題がある可能性もありそうです。
>TakaiYさん
わかりにくい質問にも関わらず、ここまで丁寧にご対応してくださり感謝申し上げます。
今後の質問の作り方についてもいただいたアドバイス通り改善します。
>bsdfanさん
>ThreadPoolExecutor の submit に、Popen のオブジェクトを渡しても、ちゃんと動きました。
コードを書き直したところ、ちゃんと動きました。ThreadPoolExecutorとProcessPoolExecutorを混同してしまったようで、お騒がせしてすみません…
回答1件
0
ベストアンサー
自分だったらこう書くかなというイメージ(参考例)です。
ざっくりいろいろ省略しているので、このままでは動きませんので、適宜肉付けしてください。
タイトルの「スレッド数の上限の範囲内で関数を並行処理する」だけだと、ThreadPoolExecutor を使うのが良いと思ったのですが、質問の内容をみると今回のケースでは使いにくいと思います。
実行したサブプロセスを submit で渡して処理するやり方だと、排他処理をちゃんとしないといけなくなり面倒なので、スレッド毎にサブプロセスを固定してしまうのが良いと思います。
そのためには、スレッドの中でサブプロセスを起動するのでも良さそうで、そうなると ThreadPoolExecutor ではなく、通常の Thread のほうが書きやすいと思います。
方法としては、実行したい内容をキューに入れておいて、初期化が終わったスレッド、処理が終わったスレッドが、キューから順に取り出して処理させていくのでどうでしょうか。
python
1from threading import Thread 2from queue import Queue 3 4def func(exec_path, in_q, out_q): 5 proc = subprocess.Popen(exec_path) 6 7 # 初期化 8 funcX(proc) 9 10 # 本体 11 while True: 12 cmd = in_q.get() 13 result = funcY(proc, cmd) 14 out_q.put(result) 15 16def main(): 17 input_q = Queue() 18 result_q = Queue() 19 20 for x in list1: 21 input_q.put(x) 22 23 th1 = Thread(target=func, args=(file_path1, input_q, result_q)) 24 th2 = Thread(target=func, args=(file_path2, input_q, result_q)) 25 26 th1.run() 27 th2.run() 28 29 while True: 30 result = result_q.get() 31 print(result) 32 33main()
投稿2023/02/21 13:56
総合スコア4899
関数を一纏めにする解決方法をお聞きした時は、「それで解決はするけどどうなんだろう」と少しモヤモヤしたのですが、提示していただいたコードを見ると腑に落ちました。
それでスレッドの中でサブプロセスを起動し、入力したい内容のqueueと結果を受け取るqueueと2つ用意しておけばよかったのですね。
わかりにくい質問にも関わらず、丁寧に対応してくださってありがとうございました。
時間が経った後の質問で申し訳ないのですが、threadが終了するたびにfunc内の本体の実行結果を得たい＋実行結果からlist1の要素が一部削除されていく（つまりinput_q自体が変わっていく）ような場合なら、どういう実装にするべきなのでしょうか？
イメージはこんな感じです。
funcXを実行しておいて初期化を行っておく。list1=[a,b,c,d,e,f,g,h,i]、スレッド数は2とする。
thread1がlist1[0]=aを使用してfunc内の本体の部分を実行
↓
thread2がlist1[1]=bを使用してfunc内の本体の部分を実行
↓
thread1が終了し、すぐ結果を得る。その結果からlist1がc,eを削除→list1=[a,b,d,f,g,h,i]となる
↓
thread1がlist1[2]=dを使用してfunc内の本体の部分を実行
↓
thread1が終了し、すぐ結果を得る。その結果からlist1がgを削除→list1=[a,b,d,f,h,i]となる
↓
thread2がlist1[3]=fを使用してfunc内の本体の部分を実行。
↓
以下、threadがlist1の中で最後の要素に辿りつくまで繰り返す。
ちょっとこれだけだとわからないです。thread1とthread2の終わる順番でも変わってきそうですが・・・
ここに書かれている通りにやりたいなら、list1を排他制御しておけばできるのではないでしょうか。
(もう少し具体的にして、別質問に上げたほうがいいと思います)
返信ありがとうございます。申し訳ないのですが、イメージはこんな感じです以下で微妙に誤っている箇所がありますね。。。
条件として、(1)同時に複数スレッドを動かすが、同時に動かせるスレッド数の上限は決まっている（これは元々の質問の通りです）。(2)終わったスレッドがあったらそこから結果を得てlist1を更新し、またそのスレッドを動かす。(3)スレッドが終わる順番はランダムに近く、thread1とthread2との間に特に決まった順番は無い。といった感じです。
あなたの回答
tips
プレビュー
