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有一個_taskqueue類型的成員multiprocessing.Queue，它需要一個可選的maxsize參數;不幸的是它構造它沒有maxsize參數集。

我建議將multiprocessing.Pool與multiprocessing.Pool.__init__的副本粘貼maxsize到_taskqueue的構造函數。

猴修補的對象（游泳池或隊列）也將工作，但你必須猴補丁pool._taskqueue._maxsize和pool._taskqueue._sem所以這將是非常脆弱：

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

Answer 2

你可以增加明確的隊列在這種情況下使用最大參數並使用queue.put()而不是pool.apply_async()。然後，工作進程可能：

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

如果要限制創建的輸入參數/結果是在內存中活動的工作流程的大致數數，那麼你可以使用pool.imap*()方法：

#!/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

如果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()