Come utilizzare Python multiprocessing Pool.map per riempire l'array numpy in un ciclo for

Question

Desidero riempire un array 2D-numpy all'interno di un ciclo for e collegare il calcolo utilizzando il multiprocessing.

import numpy 
from multiprocessing import Pool 


array_2D = numpy.zeros((20,10)) 
pool = Pool(processes = 4) 

def fill_array(start_val): 
    return range(start_val,start_val+10) 

list_start_vals = range(40,60) 
for line in xrange(20): 
    array_2D[line,:] = pool.map(fill_array,list_start_vals) 
pool.close() 

print array_2D 

L'effetto di esecuzione esso è che Python effettua 4 sottoprocessi e occupa 4 core CPU, ma la finitura Indifferente esecuzione e la matrice non viene stampato. Se provo a scrivere la matrice sul disco, non succede nulla.

Qualcuno può dirmi perché?

Answer 1

Il problema è dovuto all'esecuzione del ciclo pool.map in per, Il risultato del metodo map() è funzionalmente equivalente alla mappa built-in(), ad eccezione del fatto che le singole attività vengono eseguite in parallelo. così nel tuo caso il pool.map (fill_array, list_start_vals) saranno chiamati 20 volte e iniziare a correre parallelo per ogni iterazione del ciclo for, sottostante Codice dovrebbe funzionare

Codice:

#!/usr/bin/python 

import numpy 
from multiprocessing import Pool 

def fill_array(start_val): 
    return range(start_val,start_val+10) 

if __name__ == "__main__": 
    array_2D = numpy.zeros((20,10)) 
    pool = Pool(processes = 4)  
    list_start_vals = range(40,60) 

    # running the pool.map in a for loop is wrong 
    #for line in xrange(20): 
    # array_2D[line,:] = pool.map(fill_array,list_start_vals) 

    # get the result of pool.map (list of values returned by fill_array) 
    # in a pool_result list 
    pool_result = pool.map(fill_array,list_start_vals) 

    # the pool is processing its inputs in parallel, close() and join() 
    #can be used to synchronize the main process 
    #with the task processes to ensure proper cleanup. 
    pool.close() 
    pool.join() 

    # Now assign the pool_result to your numpy 
    for line,result in enumerate(pool_result): 
     array_2D[line,:] = result 

    print array_2D 

Answer 2

Il seguenti lavori. Innanzitutto è una buona idea proteggere la parte principale del codice all'interno di un blocco principale per evitare strani effetti collaterali. Il risultato di poo.map() è un elenco contenente le valutazioni per ciascun valore nell'iteratore list_start_vals, in modo tale che non sia necessario creare array_2D in precedenza.

import numpy as np 
from multiprocessing import Pool 

def fill_array(start_val): 
    return list(range(start_val, start_val+10)) 

if __name__=='__main__': 
    pool = Pool(processes=4) 
    list_start_vals = range(40, 60) 
    array_2D = np.array(pool.map(fill_array, list_start_vals)) 
    pool.close() # ATTENTION HERE 
    print array_2D 

forse si avranno problemi con l'uso pool.close(), dai commenti di @hpaulj si può semplicemente rimuovere questa riga nel caso in cui si hanno problemi ...

Answer 3

Se si desidera continuare a utilizzare riempire la matrice, è possibile utilizzare pool.apply_async anziché pool.map. Lavorare da risposta di Saullo:

import numpy as np 
from multiprocessing import Pool 

def fill_array(start_val): 
    return range(start_val, start_val+10) 

if __name__=='__main__': 
    pool = Pool(processes=4) 
    list_start_vals = range(40, 60) 
    array_2D = np.zeros((20,10)) 
    for line, val in enumerate(list_start_vals): 
     result = pool.apply_async(fill_array, [val]) 
     array_2D[line,:] = result.get() 
    pool.close() 
    print array_2D 

Questo viene eseguito un po 'più lento del map. Ma non produce un errore di runtime come il mio test della versione della mappa: Exception RuntimeError: RuntimeError('cannot join current thread',) in <Finalize object, dead> ignored