Q&A
前提
Pythonでマルチスレッド、マルチプロセスの処理を勉強しようと思い、concurrent.futuresを触っています。
concurrent.futuresでは原則マルチスレッドとマルチプロセスは同じ書き方でできるようですが、
ThreadPoolExecutorとProcessPoolExecutorの違いについて疑問があります。
また、ThreadPoolExecutorではうまくいってもProcessPoolExecutorではうまくいかないので何が原因なのかを知りたいです。
Windows11のsurface laptop4で
CPUは
11th Gen Intel(R) Core(TM) i7-1185G7 @ 3.00GHz 3.00 GHz
メモリは32GBのsurfaceです。
Python3.9.12です。Anacondaで実装しました。
実現したいこと
手始めに時間がかかる計算を行う関数を作りました。
Python
1def sample_func(num, result): 2 print("start:", str(num)) 3 the_list = [] 4 list1 = np.arange(0,num,1) 5 list2 = np.arange(0,num,1) 6 for i in list1: 7 for j in list2: 8 the_list = np.append(the_list, j*i) 9 10 answer = np.log10(sum(the_list)) 11 r = result 12 r.append(answer) 13 print("end:", str(num)) 14 return num
単純に1からnumまでの数字のリストを2つ作り、すべてのペアを掛け合わせて足すだけです。
num=400くらいだと10秒前後かかります。
これを
Python
1numbers_list = [397, 398, 399, 400] 2results = [] 3# 並列化 4with timer(): 5 with futures.ProcessPoolExecutor(max_workers=2) as executor: 6 task_list = [] 7 for i in numbers_list: 8 # 関数の実行部 9 task_list.append(executor.submit(sample_func, num = i, result=results)) 10 print("submit end") 11 true_results = [] 12 for future in task_list: 13 true_results.append(future.result()) 14 15 16# 結果表示 17print(results) #処理終わった順 18print(true_results) #入力した順 19print('completed')
で実行するとエラーが返ってきます。
Python
1submit end 2--------------------------------------------------------------------------- 3BrokenProcessPool Traceback (most recent call last) 4Input In [35], in <cell line: 5>() 5 12 true_results = [] 6 13 for future in task_list: 7---> 14 true_results.append(future.result()) 8 17 # 結果表示 9 18 print(results) #処理終わった順 10 11File ~\anaconda3\lib\concurrent\futures\_base.py:446, in Future.result(self, timeout) 12 444 raise CancelledError() 13 445 elif self._state == FINISHED: 14--> 446 return self.__get_result() 15 447 else: 16 448 raise TimeoutError() 17 18File ~\anaconda3\lib\concurrent\futures\_base.py:391, in Future.__get_result(self) 19 389 if self._exception: 20 390 try: 21--> 391 raise self._exception 22 392 finally: 23 393 # Break a reference cycle with the exception in self._exception 24 394 self = None 25 26BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.
ProcessPoolExecutorではなくThreadPoolExecutorで行うと問題なく実行されます。
ProcessPoolExecutorの部分をThreadPoolExecutorに置き換えただけで他は何も変えていません。
ここで疑問が2つあります。
①
ThreadPoolExecutorで行ったときでも、通常のforループ処理で計算したときより早くなります(42.6589秒→29.5869秒)
マルチスレッドで高速化できるのはI/O処理といったCPUではない部分が律速しているときだと認識していたのですが、
なぜ42秒から29秒まで速くなったのでしょうか？
上記の関数(400＊400個程度の数字のペアの計算)はCPUバインドな計算ではないということでしょうか？
②
ThreadPoolExecutorをProcessPoolExecutorに置き換えただけではエラーが出てくるのは何が理由でしょうか？
BrokenProcessPoolというエラーは何かが無理やりプロセスを止めているような感じだと思っているのですが、
コードを修正すればいいのか、環境を修正すればいいのか、さっぱりわかりません。
普段はJupyter Notebookですべて行っているのですが、futureは対話型ではうまくいかないと書いてあったので
.pyファイルにしてコマンドプロンプトから実行してみましたが、それでも同じエラーが返ってきました。
どなたか助けていただけないでしょうか。
初心者ですので、質問に不備があれば追記いたします。
よろしくお願いいたします。
回答2件
2
ベストアンサー
マルチスレッドで高速化できるのはI/O処理といったCPUではない部分が律速しているときだと認識していたのですが、なぜ42秒から29秒まで速くなったのでしょうか？
Why are numpy calculations not affected by the global interpreter lock?
に書かれてるように、numpyの計算の多くの部分はgilの制約を受けないので、マルチスレッドで高速化できます
「def sample_func(num, result):」内の計算に、そういうものがあるのでしょう
ThreadPoolExecutorをProcessPoolExecutorに置き換えただけではエラーが出てくるのは何が理由でしょうか？
concurrent.futures -- 並列タスク実行
の「ProcessPoolExecutor」のコード例のように、「def main():」の中に入れて、
python
1if __name__ == '__main__':
の中で「main()」を実行してみてください
投稿2022/12/05 12:47
編集2022/12/05 12:56
総合スコア7638
1
futureは対話型ではうまくいかない
はい。 なので、メインの実行部分を
python
1if __name__ == '__main__':
の後に入れてみてください。
timer()がどのような関数なのかわからかったので、それを省いたコードでこちらで動きました。
ちなみに、windowsはfork()がちゃんと実装されていないので、マルチプロセスによる並列処理がうまくうごかないことがあります。ちゃんとマルチプロセスで動かしたいのであれば、linuxなどに移行することをお勧めします。
投稿2022/12/05 12:16
総合スコア12285
あなたの回答
tips
プレビュー
下記のような回答は推奨されていません。
このような回答には修正を依頼しましょう。
また依頼した内容が修正された場合は、修正依頼を取り消すようにしましょう。
2022/12/05 13:24
2022/12/05 23:58
2022/12/07 11:38