

CPUバインドな並列タスクがあって途中でバリア同期？をとって集計タスクを追加してから処理を再開したいです

[code]

    |   |   |
    A1  B1  C1
    ↓   ↓   ↓
    +---+---+---+ バリア同期
    |   |   |   |
    A2  B2  C2  M2
    ↓   ↓   ↓   
    +---+---+---+ バリア同期
    |　　　　　　|   |   |
    A3  B3  C3  M3
    
[/code]

正確には  
A1 で作ったデータを A2  
B1 で作ったデータを B2  
C1 で作ったデータを C2  
で使って続きを続行しつつ  
A1 B1 C1 が全部おわった段階で別の集計タスクをさらに追加したいだけなので  
A2 B2 C2 は他の終了を待つ必要はないですが  
A B C のタスク粒度に差がないので楽に実現できるならちょっとぐらい待ってもかまわないです

* * *

マルチスレッドなら共有変数が使えるのでセマフォを使うなりいくらでもやりようがあるんですが  
python のスレッドは OS ネイティブのものではなく  
イベントループで実現されていてCPUを複数使ってくれないみたいです

プロセスなら複数のCPUにバインドされるみたいですが  
メモリ空間が別になってしまうので最初と最後の args で渡す以外に  
途中で情報共有する方法がわかりません

サブプロセスの終了＋メインプロセスの JOIN で終了を待つことはできますが  
プロセスを exit してしまうと A2 を新しく作っても A1 とメモリ空間を共有してないので A1 のデータを使うことができません

もちろんファイルシステムやデータベース等外部サービスにカウンタをもっておいて  
各プロセスでカウンタをインクリメントして  
M のタスクでカウンタを見張り続けるみたいなことをすればできそうですが  
システムが複雑になってしまうので軽く同期を実現するいい方法はありませんか？

* * *

そもそも python のスレッドで複数のCPUを使う方法があれば教えていただけると助かります


CC BY-SA 4.0
2022-02-09T03:19:57.767
86220
2022-02-10T03:58:52.897
2022-02-09T03:41:02.907
question
python
python の並列タスク間でバリア同期をとりつつ処理を再開したい


Pythonでのマルチプロセスはいくつかあり, 標準のものでは以下のがあるようです

  * [multiprocessing --- プロセスベースの並列処理](https://docs.python.org/ja/3/library/multiprocessing.html)
  * [concurrent.futures.ProcessPoolExecutor](https://docs.python.org/ja/3/library/concurrent.futures.html#processpoolexecutor)

(前者の) `multiprocessing` では, キュー (Queue), パイプ (Pipe), Lock, Shared memory
が利用でき, また以下の記述も見られます

[code]

    マネージャーオブジェクトを使用して同期プリミティブを作成できることも覚えておいてください。
    詳細は マネージャー を参照してください。
    
[/code]

* * *

#### Threadについて

Python … CPython でのスレッドは OSのスレッドで複数 CPUで動作可能です。GILがあるので Pythonの処理は同時に走らないけど,
たとえば NumPy による演算と Python そのものとかなら走るはず。

(Threadについての追記)  
[threading.get_native_id()](https://docs.python.org/ja/3/library/threading.html#threading.get_native_id)
には次のように記されています。(Thread内で取得すれば OS-native Thread ID が別々であることが判る)

[code]

    Return the native integral Thread ID of the current thread assigned by the kernel. 
    
[/code]

(ちなみにプロセス IDは `os.getpid()` で取得可能)

通常の処理を Threadにするとオーバーヘッドがかかり微妙に遅くなるところ,  
NumPyの処理がある分, Thread側が微妙に速くなるケース (colabにて動作確認)  
([primesfrom2to](https://stackoverflow.com/questions/2068372/fastest-way-to-
list-all-primes-below-n) は SO から)

[code]

    import concurrent.futures
    import numpy as np
    
    def primesfrom2to(n):
        """ Input n>=6, Returns a array of primes, 2 <= p < n """
        sieve = np.ones(n//3 + (n%6==2), dtype=bool)
        for i in range(1,int(n**0.5)//3+1):
            if sieve[i]:
                k=3*i+1|1
                sieve[       k*k//3     ::2*k] = False
                sieve[k*(k-2*(i&1)+4)//3::2*k] = False
        return np.r_[2,3,((3*np.nonzero(sieve)[0][1:]+1)|1)]
    
    def primesum(num):
        return primesfrom2to(num).sum()
    
    def main(flag):
        nums = [70000000,80000000,90000000,100000000,110000000,120000000]
        if flag:
            print('normal:')
            for n in nums:
                print(primesum(n))
        else:
            print('thread:')
            with concurrent.futures.ThreadPoolExecutor() as executor:
                for number in executor.map(primesum, nums):
                    print(number)
    
[/code]


CC BY-SA 4.0
2022-02-09T08:15:32.883
86229
2022-02-10T03:58:52.897
2022-02-10T03:58:52.897
43025
43025
86220
answer