Perché i Lavoratori Multiprocessi Python non muoiono?

Question

Sto usando la funzionalità di multiprocessing di Python per mappare alcune funzioni attraverso alcuni elementi. Qualcosa sulla falsariga di questo:

def computeStuff(arguments, globalData, concurrent=True): 
    pool = multiprocessing.Pool(initializer=initWorker, initargs=(globalData,)) 
    results = pool.map(workerFunction, list(enumerate(arguments))) 
    return results 

def initWorker(globalData): 
    workerFunction.globalData = globalData 

def workerFunction((index, argument)): 
    ... # computation here 

Generalmente corro test in ipython utilizzando sia CPython e PyPy. Ho notato che i processi generati spesso non vengono uccisi, quindi iniziano ad accumularsi, ognuno utilizzando un concerto di ram. Questo succede quando si preme ctrl-k durante un calcolo, che invia il multiprocessing in una grande frenesia di confusione. Ma anche quando finiscono i calcoli, quei processi non moriranno in Pypy.

In base alla documentazione, quando il pool viene raccolto dei rifiuti, deve chiamare terminate() e terminare tutti i processi. Cosa sta succedendo qui? Devo chiamare esplicitamente lo close()? In caso affermativo, esiste una sorta di gestore del contesto che gestisce correttamente la chiusura delle risorse (ad esempio i processi)?

Questo è su Mac OS X Yosemite.

Answer 1

La garbage collection di PyPy è pigro, quindi non riuscire a chiamare close significa che lo Pool viene pulito "a volte", ma ciò potrebbe non significare "in qualunque momento presto".

Una volta che Pool è correttamente close d, gli operatori escono quando terminano le attività. Un modo semplice per garantire la Pool è chiuso in pre-3.3 di Python è:

from contextlib import closing 

def computeStuff(arguments, globalData, concurrent=True): 
    with closing(multiprocessing.Pool(initializer=initWorker, initargs=(globalData,))) as pool: 
     return pool.map(workerFunction, enumerate(arguments)) 

Nota: ho anche tolto la conversione esplicita a list (inutile, dal momento che map itererà all'iteratore enumerate per voi), e ha restituito il risultati direttamente (non è necessario assegnare a un nome solo per tornare sulla riga successiva).

Se si desidera garantire la chiusura immediata nel caso di eccezione (su pre-3.3 Python), si utilizzerà un blocco try/finally o si scrive un semplice gestore di contesto (che potrebbe essere riutilizzato per altri luoghi in cui si utilizza un Pool):

from contextlib import contextmanager 

@contextmanager 
def terminating(obj): 
    try: 
     yield obj 
    finally: 
     obj.terminate() 

def computeStuff(arguments, globalData, concurrent=True): 
    with terminating(multiprocessing.Pool(initializer=initWorker, initargs=(globalData,))) as pool: 
     return pool.map(workerFunction, enumerate(arguments)) 

l'approccio terminating è superiore in quanto garantisce immediatamente l'uscita processi; in teoria, se stai utilizzando thread altrove nel tuo programma principale, gli operatori Pool potrebbero essere biforcati con thread non demone, che manterrebbero i processi attivi anche quando il thread del task worker è terminato; terminating nasconde questo uccidendo i processi forzatamente.

Se il vostro interprete Python è 3.3 o superiore, l'approccio terminating è built-in per Pool, in modo che nessun involucro speciale è necessaria per l'istruzione with, with multiprocessing.Pool(initializer=initWorker, initargs=(globalData,)) as pool: opere direttamente.