ProcessPoolExecutor（Python2.7) Queue error.

Question

Python2.7.12の環境で、concurrent.futures.ProcessPoolExecutor を使い
並列処理を実装しています。
(訳あって、multiprocessing の並列処理関数は使っていません。Queue関数は使用。)

class A():
    def __init__(self):
        self.queue =  multiprocessing.Queue(maxsize=1024)

    
    def proc():
        todo = []
        proc_list = get_classes()  # 処理をさせたいクラス群を取得
        with concurrent.futures.ProcessPoolExecutor() as executor: 
           for p in proc_list:
               future = executor.submit(p.func, self.queue) # ★
               todo.append(future)

        for future in futures.as_completed(todo):
            res = future.result()

上記のコードを実行すると★部分で以下のエラーが発生します。
Traceback (most recent call last):
File "/usr/lib/python2.7/multiprocessing/queues.py", line 268, in _feed
send(obj)
File "/usr/lib/python2.7/multiprocessing/queues.py", line 77, in getstate
assert_spawning(self)
File "/usr/lib/python2.7/multiprocessing/forking.py", line 52, in assert_spawning
' through inheritance' % type(self).name
RuntimeError: Queue objects should only be shared between processes through inheritance

ProcessPoolExecutorをThreadPoolExecutorに変更するとエラーなく正常処理します。

以下の条件で回避方法がありましたらご教授下さい。
(a)マルチコアでもGlobal Interpreter Lock機構の制限を受けるThreadPoolExecutorは使わない
(b)multiprocessingの並列処理も使わない
(c)ProcessPoolExecutorまたは上記(a)(b)以外の手法で並列化を実現したい
(d)multiprocessing.Queueは、並列化する関数(p.func)で参照したいのでできれば使いたいが
使わなくても同等の機能(キュー機構)が使えればそちらでも構わない

よろしくお願い致します。
以上です。

Answer 1

骨が折れますが、リモートマネージャー経由でmultiprocessing.Queueオブジェクトを共有するのが現実的な回避方法かと。

https://docs.python.org/ja/3/library/multiprocessing.html#using-a-remote-manager

ProcessPoolExecutorでタスクをsubmitする側でリモートマネージャーのサーバを動かしてmultiprocessing.Queueオブジェクトを登録しておき、タスク側でリモートマネージャーのクライアントを使用してそのQueueオブジェクト（のProxyオブジェクト）を取得する流れになると思います。

• GILフリー => CPythonでやるならマルチプロセス一択
• マルチプロセス対応のキュー機構 => multiprocessing.Queue or 標準ライブラリ外の何か

なので、multiprocessingのマネージャー機構を使うことを勧めています。ProcessPoolExecutorではありませんが、Windows環境でもLockやValueをプロセス間で共有できたので、他の環境でも動くはずです。

あと余談ですが、concurrentパッケージを使うなら3.5や3.6あたりで試したほうが良いと思います。2.7のconcurrentパッケージはbackportですよね？

Comments

[spike0726]

丁寧な回答ありがとうございます。
ご提案のmultiprocessingのマネージャー機構というのを使ってみます！
また、Pythonのバージョンについては、開発環境の制限で、2.7に限定されています。
すみません、backportというのが理解できておりませんが、
concurrent.futures は、「pip install futures」でインストールしました。

以上、よろしくお願い致します。