Pythonマルチプロセッシング：待機中のプロセス数を制限する方法は？

Question

Pool.apply_asyncを使用して多数のタスク（パラメータが大きい）を実行すると、プロセスが割り当てられて待機状態になり、待機プロセス数に制限はありません。 、

import multiprocessing 
import numpy as np 

def f(a,b): 
    return np.linalg.solve(a,b) 

def test(): 

    p = multiprocessing.Pool() 
    for _ in range(1000): 
     p.apply_async(f, (np.random.rand(1000,1000),np.random.rand(1000))) 
    p.close() 
    p.join() 

if __name__ == '__main__': 
    test() 

私は待っているプロセスの数が限られているように、待ちキューを制限する方法を探しています：これは、以下の例のように、すべてのメモリを食べることによって終わることができ待ち行列が一杯になっている間にPool.apply_asyncがブロックされます。

Answer 1

multiprocessing.Poolは、multiprocessing.Queueのメンバーであり、任意のmaxsizeパラメータを取ります。残念ながら、それはmaxsizeパラメータセットなしでそれを構築する。

私はmaxsizeコンストラクタ_taskqueueに渡しmultiprocessing.Pool.__init__のコピー＆ペーストしてサブクラス化multiprocessing.Poolをお勧めします。

モンキー・パッチの適用対象（プールまたはキューのどちらか）も動作しますが、あなたはそれが非常に脆くなりpool._taskqueue._maxsizeとpool._taskqueue._semをモンキーパッチする必要があると思います：

pool._taskqueue._maxsize = maxsize 
pool._taskqueue._sem = BoundedSemaphore(maxsize) 

Answer 2

あなたは明示的にキューを追加することができますmaxsizeパラメータを使用し、pool.apply_async()の代わりにqueue.put()を使用します。そして、ワーカー・プロセスができます

#!/usr/bin/env python 
import multiprocessing 
import numpy as np 

def f(a_b): 
    return np.linalg.solve(*a_b) 

def main(): 
    args = ((np.random.rand(1000,1000), np.random.rand(1000)) 
      for _ in range(1000)) 
    p = multiprocessing.Pool() 
    for result in p.imap_unordered(f, args, chunksize=1): 
     pass 
    p.close() 
    p.join() 

if __name__ == '__main__': 
    main() 

Answer 3

：あなたはアクティブなワーカー・プロセスの約数にメモリ内にある作成した入力引数/結果の数を制限したい場合は

for a, b in iter(queue.get, sentinel): 
    # process it 

あなたはpool.imap*()メソッドを使用することができます

pool._taskqueueが希望のサイズを超えている場合は待ちます。

import multiprocessing 
import numpy as np 
import time 

def f(a,b): 
    return np.linalg.solve(a,b) 

def test(max_apply_size=100): 
    p = multiprocessing.Pool() 
    for _ in range(1000): 
     p.apply_async(f, (np.random.rand(1000,1000),np.random.rand(1000))) 

     while pool._taskqueue.qsize() > max_apply_size: 
      time.sleep(1) 

    p.close() 
    p.join() 

if __name__ == '__main__': 
    test()