Python Multiprocessing: Wie kann die Anzahl der wartenden Prozesse begrenzt werden?

Question

Wenn eine große Anzahl von Aufgaben (mit großen Parametern) mit Pool.apply_async ausgeführt wird, werden die Prozesse zugeordnet und in einen Wartezustand versetzt, und es gibt keine Begrenzung für die Anzahl der wartenden Prozesse. Dies kann unter durch den Verzehr der gesamten Speichers, wie im Beispiel am Ende:

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

ich für eine Art und Weise bin auf der Suche die Warteschlange zu begrenzen, so dass es nur eine begrenzte Anzahl von wartenden Prozessen ist, und Pool.apply_async ist blockiert, während die Warteschlange voll ist.

Answer 1

multiprocessing.Pool hat eine _taskqueue Mitglied des Typs multiprocessing.Queue, die einen optionalen maxsize Parameter übernimmt; Leider baut es ohne den maxsize Parametersatz.

würde ich Subklassifizieren multiprocessing.Pool mit einer Copy-Paste von multiprocessing.Pool.__init__ empfehlen, die maxsize-_taskqueue Konstruktor übergibt.

Monkey-Patching das Objekt (entweder den Pool oder die Warteschlange) auch funktionieren würde, aber Sie würden pool._taskqueue._maxsize und pool._taskqueue._sem monkeypatch haben, so wäre es recht spröde sein:

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

Answer 2

Sie explizite Queue hinzufügen könnte mit maxsize-Parameter und verwenden Sie in diesem Fall queue.put() anstelle von pool.apply_async(). Dann Arbeitsprozesse könnten:

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

Wenn Sie die Anzahl der erzeugten Eingabeargumente/Ergebnisse einschränken möchten, die in etwa der Zahl der aktiven Arbeitsprozesse im Speicher sind, dann könnten Sie pool.imap*() Methoden verwenden:

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

warten, wenn pool._taskqueue über die gewünschte Größe ist:

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