Python：types.ModuleTypeで生成したDynamicModuleを並列処理させようとするとModuleNotFoundErrorになる

Question

前提・実現したいこと

生成したDynamicModuleを並列処理させようとすると、以下のエラーメッセージが発生しました。
このエラーを解消したいと考えております。

発生している問題・エラーメッセージ

Process SpawnPoolWorker-1:
Traceback (most recent call last):
  File "C:\AppData\Local\Programs\Python\Python37\lib\multiprocessing\process.py", line 297, in _bootstrap
    self.run()
  File "C:\AppData\Local\Programs\Python\Python37\lib\multiprocessing\process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "C:\AppData\Local\Programs\Python\Python37\lib\multiprocessing\pool.py", line 110, in worker
    task = get()
  File "C:\AppData\Local\Programs\Python\Python37\lib\multiprocessing\queues.py", line 354, in get
    return _ForkingPickler.loads(res)
**ModuleNotFoundError: No module named 'dynMod'**

・
・

該当のソースコード

    # DynamicModule生成（モジュール名はdynMod）
    DynamicModule()

　　# DynamicModuleインポート
    import dynMod

    # 並列処理数
    pool = Pool(4)

    # 並列処理
    list1 = pool.starmap(dynMod.ParaProc, multi_args)

■再現コード
import types
import sys


def dynamicModule(Param):

    spam_module = types.ModuleType('spam', 'dynamic generated module')
    spam_class = """
def calculation(Num1, Num2):

    # 計算式
    result = """ + Param + """

    return result

if __name__ == "__main__":

    calculation(1, 2)    

"""
    exec(spam_class, spam_module.__dict__)
    sys.modules['spam'] = spam_module

    return spam_module


if __name__ == "__main__":

    from multiprocessing import Pool

    Param = 'Num1 * Num2'

    # DynamicModule生成
    dynamicModule(Param)

    # DynamicModuleインポート
    import spam

    # 並列処理数
    pool = Pool(3)

    # インプットパラメータ
    multi_args = ([2, 2], [2, 3], [2, 4], [2, 5], [2, 6], [2, 7])

    # 並列処理
    list_result = pool.starmap(spam.calculation, multi_args)

試したこと

・並列処理ではなく、実行すれば正常終了
・import dynModをDynamicModule()で生成するロジックに差し込んだが、同様のエラーが発生
・DynamicModule()で生成するロジックでrunpyをインポートし、再度dynModモジュールを読み込ませようとしたが、同様のエラーが発生

補足情報（FW/ツールのバージョンなど）

Python3.7
Windows10

Answer 1

エラーの原因については、子プロセスから実行される位置にモジュールが無い為です。

プロセスベースの並列処理

> このパッケージに含まれる機能を使用するためには、
> 子プロセスから __main__ モジュールをインポートできる必要があります。

---

解決策1: ThreadPool を使う

from multiprocessing.pool import ThreadPool as Pool

multiprocessing.pool.Pool の代わりに、
multiprocessing.pool.ThreadPool なら、現状のコードを動かす事も可能です。

スレッドの場合は、プロセスの時のような制限はありません。

---

解決策2: モジュール定義を if __name__ == "__main__": の外に出す。

一般的なケースでは、子プロセスで使われるモジュールや関数は、
if __name__ == "__main__": の外に出す必要があります。

但し、期待された挙動かどうかはわかりません。
dynamicModuleが合計４回呼ばれます。

import types
import sys

def dynamicModule(Param):
    spam_module = types.ModuleType('spam', 'dynamic generated module')
    spam_class = """
def calculation(Num1, Num2):
    result = """ + Param + """
    return result
"""
    exec(spam_class, spam_module.__dict__)
    sys.modules['spam'] = spam_module

    print("DynamicModule生成") #### XXX: <----- 確認用。計4回呼ばれます。

    return spam_module

Param = 'Num1 * Num2'
# DynamicModule生成
dynamicModule(Param)
# DynamicModuleインポート
import spam

if __name__ == "__main__":
    from multiprocessing import Pool

    # インプットパラメータ
    multi_args = ([2, 2], [2, 3], [2, 4], [2, 5], [2, 6], [2, 7])

    # 並列処理
    with Pool(3) as pool:
        list_result = pool.starmap(spam.calculation, multi_args)
        print(list_result)

---

解決策3: initializerを指定する。

こちらは２回で済みます。

import types
import sys

def dynamicModule(Param):
    spam_module = types.ModuleType('spam', 'dynamic generated module')
    spam_class = """
def calculation(Num1, Num2):
    result = """ + Param + """
    return result
"""
    exec(spam_class, spam_module.__dict__)
    sys.modules['spam'] = spam_module

    print("DynamicModule生成") #### XXX: <----- 確認用。計2回呼ばれます

    return spam_module


if __name__ == "__main__":
    from multiprocessing import Pool

    # インプットパラメータ
    multi_args = ([2, 2], [2, 3], [2, 4], [2, 5], [2, 6], [2, 7])

    Param = 'Num1 * Num2'
    # DynamicModule生成
    dynamicModule(Param)
    # DynamicModuleインポート
    import spam

    # 並列処理
    with Pool(3, dynamicModule, (Param,)) as pool:
        list_result = pool.starmap(spam.calculation, multi_args)
        print(list_result)

Comments

[m_osa]

ありがとうございました。3つのいずれの解決策でも問題が解消しました。
DynamicModule生成は1度だけでよいので、解決策1で行おうと思います。

[teamikl]

一点を説明補足で。スレッドの場合はプロセスのような制限がないと書いてしまいましたが、
メインスレッドでの実行を要求されるケースでは、逆にスレッドが実行時に制限になることも有るので、
場合によっては、解決策 (1) ではなく (3) の方法が必要になることも有ります。
(例えばsqlite3でデータベース接続が絡む場合等)

計算のみの場合は、特に影響ありません。
モジュール生成コストもプロセス起動時に一度のみなので、
それ程気にする必要はないと思います。

[m_osa]

ご丁寧にありがとうございました。場合によって使い分けをするようにします。

Answer 2

問題が再現できる最小限のコードを提示可能ですか？

エラーログより、multiprocessing.pool.Pool を利用は把握できましたが、
以下の点が不明です。

• if __name__ == "__main__": の有無
• ParaProc関数の定義位置 <-- 本来はこちらも重要なのですが
• 動的なモジュールの実装方法 <-- エラー自体はこちらの内容な気がします

該当のエラー自体は特定できません。（多数原因が考えられます)

---

モジュール自体は関数を纏めてる入れ物なので、それほど重要ではありません。
(がエラーメッセージは ModuleNotFound なので、そちらの原因はDynamicModuleの内容次第です)

ただし、ドキュメントに説明がある様に
別プロセスに渡す関数自体は import 可能なものである必要があります。
動的にモジュールを構築してる場合、そちらが引っかかりそう。

動的にモジュールを生成した場合のサンプル

提示されたコードを動かす為だけに構成したので、
特に動的にする意味のあるものではありませんが、
大抵の場合、動的な要因を引数にする等、別アプローチで解消した方が良いです。

## 重要
# https://docs.python.org/ja/3/library/multiprocessing.html
# > このパッケージに含まれる機能を使用するためには、
# > 子プロセスから __main__ モジュールをインポートできる必要があります。

def func(n):
    return n * n


def dynamic_module():
    from types import ModuleType

    ## ThreadPool の場合はここで関数を定義しても可能
    # def func(n):
    #     return n * n
    
    module = ModuleType("DynamicModule")
    module.func = func

    ## import dynmod でインポートしたい場合
    import sys
    sys.modules["dynmod"] = module

    return module


dynamic_module()
import dynmod


if __name__ == "__main__":
    from multiprocessing.pool import Pool, ThreadPool

    with Pool(5) as pool:
        result = pool.map(dynmod.func, range(10))
        print(result)

Comments

[m_osa]

ご回答いただき、ありがとうございます。
問題が再現できるコードは以下になります。
※理解が乏しいこともあり、上記の内容とは少しロジックを変えております。

[teamikl]

出来れば質問文の編集でお願いします。インデントが解らないので

[m_osa]

申し訳ありません。質問文に再現コードを追記しました。