Python itertools with multiprocessing - 巨大なリスト対非効率的なCPU使用率のイテレータ

Question

私は関数の引数として繰り返されるn個の要素（以下「ペア」という）のバリエーションを処理します。明らかに、 "r"リストがすべてのメモリを消費するのに十分な大きさでない限り、すべて正常に動作します。問題は、最終的に6つの要素に対して16回以上繰り返しなければならないことです。私はこのために雲の中に40コアシステムを使用しています。

は私が持つ問題を解決しようとした..私は、私が代わりに先行し、ここでの問題が始まる巨大なリストを作成するイテレータを使用する必要があると考えてい

if __name__ == '__main__': 
    pool = Pool(39) 
    r = itertools.product(pairs,repeat=16) 
    pool.map(f, r) 

：

コードは、次のようなルックスに見えます次のコード：

if __name__ == '__main__': 
    pool = Pool(39) 
    for r in itertools.product(pairs,repeat=14): 
    pool.map(f, r) 

メモリの問題はなくなりますが、CPU使用率はコアあたり5％のようです。コードのシングルコアバージョンは、これより高速です。

あなたが私に少しを導くことができれば、私は本当に感謝..

感謝。

Answer 1

（あなたはすべての出力を保存することができれば、それは前提としているので、あなたも、すべての入力を保存することができます）あなたの元のコードは、独自のコード（itertools.productリターン発電機）での先行listを作成するのではなく、pool.mapは全体の発電を実現しています。

ここではpool.mapを使用しないでください。注文結果が必要な場合は、pool.imapを使用するか、結果の順序が重要でない場合は、pool.imap_unorderedを使用してください。あなたは副作用のためpool.mapを使用している場合ので、

if __name__ == '__main__': 
    pool = Pool(39) 
    for result in pool.imap(f, itertools.product(pairs, repeat=16)): 
     print(result) 

：どちらかの呼び出しの結果を繰り返し処理（listでラップしていない）、そして彼らが来るとして結果を処理し、メモリは問題ではありません結果を出すだけでいいですが、結果と順序は関係ありません。imap_unorderedを使用してパフォーマンスを大幅に向上させ、実際に何も格納せずに効率的に "結果"を排除するためにcollections.dequeを使用してください。maxlenの0が最も速いイテレータを強制的に実行して結果を保存せずに済ませる最小のメモリ方法）：

from collections import deque 

if __name__ == '__main__': 
    pool = Pool(39) 
    deque(pool.imap_unordered(f, itertools.product(pairs, repeat=16)), 0) 

最後に、私は労働者を指定するのに少し疑念があります39 Pool; multiprocessingは、CPUのバウンドタスクに大きな効果があります。より多くの労働者をCPUコアよりも使用して利益を得ている場合は、multiprocessingがIPCのコストよりも多くのコストを要し、より多くのワーカーを使用すると、より多くのデータをバッファリングするだけで問題を隠すことができます。

作業が大部分I/Oバウンドの場合は、スレッドベースのプールを使用してみてください。これにより、親プロセスと子プロセス間のIPCコストと同様に、ピクルスリングとアンピケットのオーバーヘッドが回避されます。プロセスベースのプールとは異なり、PythonスレッドはGILの問題を抱えているため、CPUはPythonで動作します（I/OのGIL解放コール、.dll/.soファイルへの​​呼び出し、およびnumpyなどの特定のサードパーティエクステンションGILは単一のコアに限定されています（そして、CPUバインド作業のためのPython 2.xでは、GILの競合を解決し、コンテキストスイッチを実行することが多いので、Python 3はほとんどの廃棄物を取り除きます）。しかし、あなたの作業が大部分I/Oバウンドの場合、I/OをブロックするとGILが解放され、他のスレッドが実行できるようになります。そのため、ほとんどのスレッドがI/Oを遅らせる限り多くのスレッドを持つことができます。他のワーカーや親プロセスに影響を与えずに、 "共有"状態に書き込むことができると仮定して、各ワーカーのアドレススペースに依存しないようにプログラムを設計していない限り、

from multiprocessing import Pool 

：

from multiprocessing.dummy import Pool 

、あなたは、スレッドの代わりに、プロセスに基づいて、プールのmultiprocessing.dummyバージョンを取得。

Answer 2

39の作品のプールに1つのペアを送信しているので、2番目のコード例は遅いです。あなたのリクエストを処理しているワーカーは1人で、38人は何もしません！メインスレッドからワーカープロセスにデータをパイプする際にオーバーヘッドが生じるため、処理が遅くなります。

いくつかのペアを「バッファ」してから、ペアのセットを実行してメモリ使用量のバランスを取っても、マルチプロセス環境を利用することができます。

import itertools 
from multiprocessing import Pool 

def foo(x): 
    return sum(x) 

cpus = 3 
pool = Pool(cpus) 
# 10 is buffer size multiplier - the number of pair that each process will get 
buff_size = 10*cpus 
buff = [] 
for i, r in enumerate(itertools.product(range(20), range(10))): 
    if (i % buff_size) == (buff_size-1): 
     print pool.map(foo, buff) 
     buff = [] 
    else: 
     buff.append(r) 

if len(buff) > 0: 
    print pool.map(foo, buff) 
    buff = [] 

上記の出力は、システムの正しいバランスを得るためにbuff_size乗数と遊ぶこの

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
[3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 5, 6, 7, 8, 9, 10, 11, 12, 13] 
[6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 8, 9, 10, 11, 12, 13, 14, 15, 16] 
[9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 11, 12, 13, 14, 15, 16, 17, 18, 19] 
[12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 14, 15, 16, 17, 18, 19, 20, 21, 22] 
[15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 17, 18, 19, 20, 21, 22, 23, 24, 25] 
[18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28] 

ようになります！