Python multiprocessing e condiviso serie NumPy

Question

Ho un problema, che è simile a questo:

import numpy as np 

C = np.zeros((100,10)) 

for i in range(10): 
    C_sub = get_sub_matrix_C(i, other_args) # shape 10x10 
    C[i*10:(i+1)*10,:10] = C_sub 

Quindi, a quanto pare non v'è alcuna necessità di eseguire questo come un calcolo di serie, dal momento che ogni sottomatrice può essere calcolato in modo indipendente. Vorrei utilizzare il modulo multiprocessing e creare fino a 4 processi per il ciclo for. Ho letto alcuni tutorial sul multiprocessing, ma non sono riuscito a capire come usarlo per risolvere il mio problema.

Grazie per il vostro aiuto

Answer 1

Un modo semplice per parallelizzare che il codice sarebbe quella di utilizzare un Pool dei processi:

pool = multiprocessing.Pool() 
results = pool.starmap(get_sub_matrix_C, ((i, other_args) for i in range(10))) 

for i, res in enumerate(results): 
    C[i*10:(i+1)*10,:10] = res 

Ho usato starmap poiché la funzione get_sub_matrix_C ha più di un argomento (starmap(f, [(x1, ..., xN)]) chiamate f(x1, ..., xN)).

Si noti tuttavia che la serializzazione/deserializzazione può richiedere tempo significativo e lo spazio, quindi potrebbe essere necessario utilizzare una soluzione di livello più basso per evitare tale sovraccarico.

---

Sembra che tu stia utilizzando una versione obsoleta di Python. È possibile sostituire starmap con piano map ma poi si deve fornire una funzione che prende un solo parametro:

def f(args): 
    return get_sub_matrix_C(*args) 

pool = multiprocessing.Pool() 
results = pool.map(f, ((i, other_args) for i in range(10))) 

for i, res in enumerate(results): 
    C[i*10:(i+1)*10,:10] = res 

Answer 2

La ricetta che segue, forse può fare il lavoro. Sentiti libero di chiedere.

import numpy as np 
import multiprocessing 

def processParallel(): 

    def own_process(i, other_args, out_queue): 
     C_sub = get_sub_matrix_C(i, other_args) 
     out_queue.put(C_sub)    

    sub_matrices_list = [] 
    out_queue = multiprocessing.Queue() 
    other_args = 0 
    for i in range(10): 
     p = multiprocessing.Process(
          target=own_process, 
          args=(i, other_args, out_queue)) 
     procs.append(p) 
     p.start() 

    for i in range(10): 
     sub_matrices_list.extend(out_queue.get()) 

    for p in procs: 
     p.join() 

    return sub_matrices_list  

C = np.zeros((100,10)) 

result = processParallel() 

for i in range(10): 
    C[i*10:(i+1)*10,:10] = result[i]