Python3.8.4の並列処理のプロセス間で、クラスを共有したい

Question

前提・実現したいこと

• ファイル「asd.py」で「class dsa」を用意
• 1つ目のプロセスで、一定間隔で「dsa.a」の値を変更(task関数)
• 2つ目のプロセスファイル「dsa.a」の値を利用(printing関数)。

ですが、両方どうすれば、共有できるのでしょうか？

※同時アクセスによるデータ整合性は無視する。

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

task =  1
printing =  0
printing =  0
task =  2
printing =  0
printing =  0
printing =  0
task =  3
printing =  0
printing =  0
printing =  0
task =  4
printing =  0
printing =  0
printing =  0

該当のソースコード

#main.py
import concurrent.futures as confu
import time
import asd

# クラス変数出力
def printing():
    while(1):
        time.sleep(1)
        print("printing = ", asd.dsa.a)

# クラス変数変更
def task():
    asd.dsa.a += 1
    print("task = ",asd.dsa.a)


def main():
    # -   -   -   -   処理開始    -   -   -   - #
    with confu.ProcessPoolExecutor(max_workers=2) as executor:
        executor.submit(printing)
        for i in range(50):
            executor.submit(task)
            time.sleep(3)

    # -   -   -   -   並列処理が終わった後    -   -   -   - #
    # print(asd.dsa.a)

if __name__ == '__main__':
    main()

# asd.py
class dsa:
    a = 0

試したこと

multiprocessingのValueにClassを入れる。
= 当然エラー

追記

「= 当然エラー」とはどんなコードを書いてどんなエラーが出たのでしょうか？

以下のとおり(main.py)です。なぜ、第一引数を「i」にしてるかは、適当に選んだからです。
第一引数の候補は、「sharedctypes.py」に存在する「typecode_to_type」の中から選ぶものです。（間違っていたらごめんなさい。）
その中にClassの引数が存在しなかったので、当然エラーと書きました。

#main.py
dsa = Value('i', asd.dsa)

TypeError: an integer is required

#sharedctypes.py
typecode_to_type = {
    'c': ctypes.c_char,     'u': ctypes.c_wchar,
    'b': ctypes.c_byte,     'B': ctypes.c_ubyte,
    'h': ctypes.c_short,    'H': ctypes.c_ushort,
    'i': ctypes.c_int,      'I': ctypes.c_uint,
    'l': ctypes.c_long,     'L': ctypes.c_ulong,
    'q': ctypes.c_longlong, 'Q': ctypes.c_ulonglong,
    'f': ctypes.c_float,    'd': ctypes.c_double
    }
```c

Comments

[meg_]

「= 当然エラー」とはどんなコードを書いてどんなエラーが出たのでしょうか？

[GUWAGUWA]

追記しました。

[quickquip]

なんとなくXY問題 https://ja.meta.stackoverflow.com/questions/2701/ っぽい雰囲気が感じられて気になります。
なぜそんなことをしたいのか? が明らかになっていないせいで、解決したいことに対して間違ったアプローチをしようとしているのでは? という感じが拭えません。

[GUWAGUWA]

XY問題については、初めて知りました。
確かに、XY問題問題が発生してますね！
質問する際に、処理を書かずに質問するのは、失礼かな？と考えてたので、付け焼き刃な知識で、処理を書くのがだめでした！
これから質問する際はXY問題もよく考えて質問します！ありがとうございました！

Answer 1

プロセス間で変数を共有するためには

• mp.Value()はProcessPoolExecutor()よりも前に作成する
• 引数initargsで共有メモリとして使用する変数を与える
• 共有メモリとして使用する変数は引数initializerで呼ばれる関数内でグローバル変数に複製する

とします。これを踏まえた上で、(少しトリッキーですが)以下のようにするのはどうでしょう?

# asd.py
import multiprocessing as mp

class dsa:
    a = mp.Value("i", 0)

# main.py
import concurrent.futures as confu
import time
import asd


# クラス変数出力
def printing():
    while 1:
        time.sleep(1)
        with asd.dsa.a.get_lock():
            print("printing = ", asd.dsa.a.value)


# クラス変数変更
def task():
    with asd.dsa.a.get_lock():
        asd.dsa.a.value += 1
        print("task = ", asd.dsa.a.value)


def initializer(x):
    asd.dsa.a = x


def main():
    # -   -   -   -   処理開始    -   -   -   - #
    with confu.ProcessPoolExecutor(
        initializer=initializer, initargs=(asd.dsa.a,), max_workers=2
    ) as executor:
        executor.submit(printing)
        for i in range(50):
            executor.submit(task)
            time.sleep(3)

    # -   -   -   -   並列処理が終わった後    -   -   -   - #
    # print(asd.dsa.a)


if __name__ == "__main__":
    main()

Comments

[GUWAGUWA]

回答ありがとうございます！２点ほど質問します。
1)これは、taskに入るたびにinitializer関数によって、グローバルなasd.dsa.aの内容が変わっているという解釈であっていますか？
2)今回のような内容をPythonでするには、これが最適なのでしょうか？もう少し分かりやすく実装できるモジュールなどはないのでしょうか？

[yymmt]

1) いいえ、最初にサブプロセスx2回呼ばれるのみです。taskに入るたびに呼ばれることはありません。グローバルなasd.dsa.aの内容をinitializer()で上書きして変えているのはその通りです。
2) クラス変数をプロセス間で共有するというのがトリッキーになっている原因の1つですので普通のグローバル変数に頼れば多少分かりやすいと思います。またマルチプロセスにしているのは何か理由がありますか?マルチスレッドであればもう少し分かりやすいと思われます。

[GUWAGUWA]

・クラス変数をプロセス間で共有するというのがトリッキーになっている原因の1つですので普通のグローバル変数に頼れば多少分かりやすいと思います。
＞なるほど！私は複数個の変数を一気に共有しようと考えてます。ですので、クラスを用いようとしました。
・またマルチプロセスにしているのは何か理由がありますか?
＞マルチプロセスにしている理由は、マルチスレッドよりも処理が軽いとサイトで見たからです。
最終的には画像処理ライブラリのOpenCVを用いようと考えてますので、処理が軽い法が良いのではと考え、マルチプロセスを利用しようと考えました。

[yymmt]

高速化のためにマルチプロセスを使うのは正しい選択のような気がします。ただ今回のような実装だと、せっかくスレッドとプロセスの抽象化を図ったconcurrent.futuresを使うメリットがなくなってしまうので、(共通化された)分かりやすい方法を求めるのは共感するところです(デコレータをつければ共有メモリに乗るみたいな)。
ソースコードの共通化という意味ではMessagePackやgRPC+ProtocolBuffersなどでも良いですが、速度を重視しているので流石に共有メモリの方が早そうな気がします(試していませんので憶測です)。

[GUWAGUWA]

MessagePackを行ってみましたが、書き込みと読み込みが発生するので、遅かったですね。
grpcはMessagePackよりも早かったので、そちらを利用します。ありがとうございました！