Python non può allocare memoria utilizzando multiprocessing.pool

Question

Il mio codice (parte di un algoritmo di ottimizzazione genetica) esegue alcuni processi in parallelo, attende che tutti finiscano, legge l'output e quindi ripete con un input diverso. Tutto funzionava bene quando ho provato con 60 ripetizioni. Dal momento che ha funzionato, ho deciso di utilizzare un numero più realistico di ripetizioni, 200. Ho ricevuto questo errore:

File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner 
self.run() 
File "/usr/lib/python2.7/threading.py", line 504, in run 
self.__target(*self.__args, **self.__kwargs) 
File "/usr/lib/python2.7/multiprocessing/pool.py", line 302, in _handle_workers 
pool._maintain_pool() 
File "/usr/lib/python2.7/multiprocessing/pool.py", line 206, in _maintain_pool 
self._repopulate_pool() 
File "/usr/lib/python2.7/multiprocessing/pool.py", line 199, in _repopulate_pool 
w.start() 
File "/usr/lib/python2.7/multiprocessing/process.py", line 130, in start 
self._popen = Popen(self) 
File "/usr/lib/python2.7/multiprocessing/forking.py", line 120, in __init__ 
self.pid = os.fork() 
OSError: [Errno 12] Cannot allocate memory 

Ecco un frammento del mio codice che utilizza piscina:

def RunMany(inputs): 
from multiprocessing import cpu_count, Pool 
proc=inputs[0] 
pool=Pool(processes = proc) 
results=[] 
for arg1 in inputs[1]: 
    for arg2 in inputs[2]: 
     for arg3 in inputs[3]: 
      results.append(pool.apply_async(RunOne, args=(arg1, arg2, arg3))) 
casenum=0 
datadict=dict() 
for p in results: 
    #get results of simulation once it has finished 
    datadict[casenum]=p.get() 
    casenum+=1 
return datadict 

La funzione RunOne crea un oggetto in classe I creato, usa un pacchetto python computazionalmente pesante per risolvere un problema di chimica che impiega circa 30 secondi e restituisce l'oggetto con l'output del risolutore di chimica.

Quindi, il mio codice chiama RunMany in serie e RunMany chiama RunOne in parallelo. Nei miei test, ho chiamato RunOne utilizzando 10 processori (il computer ha 16) e un pool di 20 chiamate a RunOne. In altre parole, len (arg1) * len (arg2) * len (arg3) = 20. Tutto ha funzionato bene quando il mio codice ha chiamato RunMany 60 volte, ma ho esaurito la memoria quando l'ho chiamato 200 volte.

Ciò significa che alcuni processi non si ripuliscono correttamente da soli? Ho una perdita di memoria? Come posso determinare se ho una perdita di memoria e come posso scoprire la causa della perdita? L'unico elemento che sta crescendo nel mio ciclo di ripetizioni a 200 è un elenco di numeri che cresce da 0 a 200. Ho un dizionario di oggetti di una classe personalizzata che ho creato, ma è limitato a una lunghezza di 50 voci: ogni volta che il ciclo viene eseguito, elimina un elemento dal dizionario e lo sostituisce con un altro elemento.

Edit: Ecco un frammento del codice che chiama RunMany

for run in range(nruns): 
    #create inputs object for RunMany using genetic methods. 
    #Either use starting "population" or create "child" inputs from successful previous runs 
    datadict = RunMany(inputs) 

    sumsquare=0 
    for i in range(len(datadictsenk)): #input condition 
     sumsquare+=Compare(datadict[i],Target[i]) #compare result to target 

    with open(os.path.join(mainpath,'Outputs','output.txt'),'a') as f: 
     f.write('\t'.join([str(x) for x in [inputs.name, sumsquare]])+'\n') 

    Objective.append(sumsquare) #add sum of squares to list, to be plotted outside of loop 
    population[inputs]=sumsquare #add/update the model in the "population", using the inputs object as a key, and it's objective function as the value 
    if len(population)>initialpopulation: 
     population = PopulationReduction(population) #reduce the "population" by "killing" unfit "genes" 
    avgtime=(datetime.datetime.now()-starttime2)//(run+1) 
    remaining=(nruns-run-1)*avgtime 
    print(' Finished '+str(run+1)+'/' +str(nruns)+'. Elapsed: '+str(datetime.datetime.now().replace(microsecond=0)-starttime)+' Remaining: '+str(remaining)+' Finish at '+str((datetime.datetime.now()+remaining).replace(microsecond=0))+'~~~', end="\r") 

Answer 1

Come mostrato nei commenti alla mia domanda, la risposta è venuto da Puciek.

La soluzione era chiudere il pool di processi una volta terminato. Ho pensato che sarebbe stato chiuso automaticamente perché la variabile results è locale a RunMany e sarebbe stata eliminata dopo il RunMany completato. Tuttavia, python non funziona sempre come previsto.

Il codice fisso è:

def RunMany(inputs): 
from multiprocessing import cpu_count, Pool 
proc=inputs[0] 
pool=Pool(processes = proc) 
results=[] 
for arg1 in inputs[1]: 
    for arg2 in inputs[2]: 
     for arg3 in inputs[3]: 
      results.append(pool.apply_async(RunOne, args=(arg1, arg2, arg3))) 
#new section 
pool.close() 
pool.join()  
#end new section 
casenum=0 
datadict=dict() 
for p in results: 
    #get results of simulation once it has finished 
    datadict[casenum]=p.get() 
    casenum+=1 
return datadict