Pythonでmultiprocessingのpoolを使うときのselfの有無で結果が大きく変わる原因がわからない

Question

pythonでマルチタスクの並列化について勉強していました。
複数のタスクを同時に動かす場合、poolを使うと良いと知ったので
例文を動かしてみました。

from multiprocessing import Pool
import os
import time
import random


def long_time_task(name):
    print('Run task {} ({})...'.format(name, os.getpid()))
    start = time.time()
    time.sleep(3)
    end = time.time()
    print('Task {} runs {} seconds.'.format(name, (end - start)))


def main():
	print('Parent process {}.'.format(os.getpid()))
	p = Pool(8)  # 同時に最大4個の子プロセス
	for i in range(5):
		p.apply_async(long_time_task, args=(i+1,))
	# 非同期処理のため、親プロセスは子プロセスの処理を待たずに、
	# 次のprintをする
	print('Waiting for all subprocesses done...')
	p.close()
	p.join()

	
	print('All subprocesses done.')

if __name__ == '__main__':
	main()

これだときれいに結果が出ました。
以下が結果です。

Parent process 37632.
Waiting for all subprocesses done...
Run task 1 (28292)...
Run task 2 (1516)...
Run task 3 (26880)...
Run task 4 (27584)...
Run task 5 (38716)...
Task 1 runs 3.0020313262939453 seconds.
Task 5 runs 3.0018770694732666 seconds.
Task 4 runs 3.005866050720215 seconds.
Task 2 runs 3.013845443725586 seconds.
Task 3 runs 3.0108532905578613 seconds.
All subprocesses done.

しかし、実際のコードにはよく関数にはよく定義する際にselfが使われているなあと思い、
def long_time_task(name):
のコードを
def long_time_task(self,name):
に書き換えて実行したところ,

Parent process 11200.
Waiting for all subprocesses done...
All subprocesses done.

実行結果に定義した関数が素通りされて出力されました。
上手くいかないとエラーが出るものだと思っていたので, なぜスルーされるのかわからないので教えていただきたいです. また関数にselfを使う際にはどうすればスルーされずに実行できるのか教えてください。

追記
さらに
p.apply_async(long_time_task, args=(i+1,))
を
p.apply_async(self.long_time_task, args=(i+1,))
に書き直すと

p.apply_async(self.long_time_task, args=(i+1,))
NameError: name 'self' is not defined

というエラーが出ました。

Answer 1

質問内容１：

def long_time_task(name):
print('Run task {} ({})...'.format(name, os.getpid()))
start = time.time()
time.sleep(3)
end = time.time()
print('Task {} runs {} seconds.'.format(name, (end - start)))

を
```diff

+def long_time_task(self, name):
．．．
...

とすると、なぜスルーされるのか？

回答：
１．引数不一致のTypeError
これは、まず 呼び出し側である、main()内の long_time_task()関数側を見てみると

    for i in range(5):
        p.apply_async(long_time_task, args=(i+1,))

となっていることから、long_time_task関数に、「(i+1,)」という1つの要素を持ったタプルを引数として渡していることがわかります。

一方スルーされる結果となった後半のコードでは

def long_time_task(self, name):

のように、2つのパラメータで受け取るようにしています。

　このように、呼び出される側（引数を受け取る側）は受け取り口を2つ用意しているのに、呼び出し側は1つしか引数を与えていません。

これは引数の数が不一致のためTypeErrorという例外を発生させる事象となります。

２．multiprocessingの子プロセスの実行時例外
multiprocessingにおいて、Poolで生成した子プロセスの実行時に例外が発生しても、その例外は通常、出力されません（親側プロセスも子の例外を検出して止まったりしません）

　上記で言えば、後半のコードの場合は、子プロセスであるlong_time_taskは、１.で言及した引数不一致のTypeErrorが発生してすぐ終了してしまっています。

しかしその例外が発生したことは一般的に見えている画面のどこにも出力されません。

これが何もエラーっぽいこともなく終わったように見えた原因です。

３．例外を捕捉するには
では、子の中でどんな例外がおきたか、どうやって捕捉すればいいのでしょうか。

Pool.apply_async() が返す結果のクラスである、AsyncResultにget()という関数があり、これを利用することができます。

略
+results = []

def main():
    print('Parent process {}.'.format(os.getpid()))

    for i in range(5):
      res =   p.apply_async(long_time_task, args=(i+1,))
+     results.append(res)

    print('Waiting for all subprocesses done...')
    p.close()
    p.join()

    print('All subprocesses done.')

+   for res in results:
+       res.get()

略

　4行追加しました（読みやすいようコメントは削除しています）
これは、apply_asyncの結果をresultsというリストに追加し、
全部子プロセスが終了した後、resultsの中の結果のget()を呼び出しています。

　通常、このget()は、それぞれの子プロセスが走らせた関数の結果を返すために使うものですが、
子プロセス実行中に例外が発生していた場合は、get()の段階でその例外が送出され、親プロセスも止まります。

上記後段のコードにあてはめると、

Parent process ****.
Waiting for all subprocesses done...
All subprocesses done.
multiprocessing.pool.RemoteTraceback: 
"""
Traceback (most recent call last):
  File "C:\Python\Python3\lib\multiprocessing\pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
TypeError: long_time_task() missing 1 required positional argument: 'name'
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File ".\test357269.py", line 34, in <module>
    main()
  File ".\test357269.py", line 31, in main
    print(res.get())
  File "C:\Python\Python3\lib\multiprocessing\pool.py", line 771, in get
    raise self._value
TypeError: long_time_task() missing 1 required positional argument: 'name'

という出力が出てすぐ止まります。

出力内容を見ると、子プロセス実行中の

TypeError: long_time_task() missing 1 required positional argument: 'name'

が原因で止まっていることが分かります。

---

質問内容２：

また関数にselfを使う際にはどうすればスルーされずに実行できるのか教えてください。

selfを使う、というのは、多分クラスのところでご覧になったのだと思いますが、今回の件は、クラスで使うselfとは無関係です。
単純に「1つだけ書くべきパラメータを2つにしてしまった」ことが原因で発生した事象に過ぎません。
selfではなく、例えば別の名前の変数を使っても、数が多すぎたり少なすぎたりするだけで、同じような事象になります。

def long_time_task(foo, name):  # selfではない別の名前、数が多い（2個）

def long_time_task(baz, yoo, name):  # 数がさらに多い（3個）

def long_time_task():  # 数が少なすぎる（0個）

　

逆に１．冒頭の後半のパターンで無理やり例外を生じさせないようにするなら、下記のようなコードになるでしょう。

def long_time_task(self, name):　# selfの代わりにfooなど別の名前を使っても同じ。
    print('Run task {} ({})...'.format(name, os.getpid()))
    start = time.time()
    time.sleep(3)
    end = time.time()
    print('Task {} runs {} seconds.'.format(name, (end - start)))

results = []

def main():
    print('Parent process {}.'.format(os.getpid()))

    for i in range(5):
-     res =   p.apply_async(long_time_task, args=(i+1,))
+     res =   p.apply_async(long_time_task, args=(None,i+1))

    print('Waiting for all subprocesses done...')
    p.close()
    p.join()

    print('All subprocesses done.')

if __name__ == '__main__':
    main()

変えたのは、

+     res =   p.apply_async(long_time_task, args=(None,i+1))

の部分です。

受け取り側がself, name の2つなのですから、呼び出し側を1つの要素から2つの要素をもったタブルに変え、さらにnameは2番目なので、2番目のパラメータにi+1を渡すようにします。

Comments

[_toy]

なるほど、パラメータの数があっていなかったからスルーされたり、エラーが出ていたのですね。
逆に言えばそこさえ合わせれば何とでもなると。
全然わからなかった部分がすべてわかりました。
丁寧に説明していただき感謝します。
ありがとうございました。