Come posso recuperare l'output da Multiprocessing in Python?

Question

Quindi, sto cercando di accelerare una routine utilizzando il modulo Multiprocessing in Python. Voglio essere in grado di leggere diversi file .csv, dividendo il lavoro tra i diversi core, per questo ho:

def csvreader(string): 

    from numpy import genfromtxt; 

    time,signal=np.genfromtxt(string, delimiter=',',unpack="true") 
    return time,signal 

Poi chiamo questa funzione dicendo:

if __name__ == '__main__': 
    for i in range(0,2): 
     p = multiprocessing.Process(target=CSVReader.csvreader, args=(string_array[i],)) 
     p.start() 

Il fatto è che questo non memorizza alcun output. Ho letto tutti i forum online e ho visto che potrebbe esserci un modo con multiprocessing.queue ma non lo capisco abbastanza bene. Esiste un metodo semplice e diretto?

Answer 1

Le migliori sono multiprocessing.Queue o multiprocessing.Pipe, progettate esattamente per questo problema. Ti consentono di inviare dati tra processi in modo semplice e sicuro.

Se si desidera restituire l'output della funzione csvreader, è necessario passare un altro argomento, ovvero lo multiprocessing.Queue tramite il quale i dati verranno inviati al processo principale. Invece di return inserire i valori, posizionarli sulla coda e il processo principale li recupererà in un momento successivo. Se non sono pronti quando il processo tenta di farli, per impostazione predefinita sarà solo bloccare (attesa) fino a quando non sono disponibili

La funzione sarebbe ora simile a questa:

def cvsreader(string, q): 
    q.put(np.genfromtxt(string, delimiter=',', unpack="true")) 

La routine principale sarebbe essere:

if __name__ == '__main__' 
    q = multiprocessing.Queue() 
    for i in range(2): 
     p = multiprocessing.Process(target=csvreader, args=(string_array[i], q,)) 
     p.start() 

# Do anything else you need in here 

time=np.empty(2,dtype='object') 
signal=np.empty(2,dtype='object') 
for i in range(2): 
    time[i], signal[i] = q.get() # Returns output or blocks until ready 
    # Process my output 

Nota che devi chiamare Queue.get() per ogni elemento che si desidera tornare.

Dai un'occhiata alla documentazione sullo multiprocessing module per ulteriori esempi e informazioni.

Answer 2

Utilizzando l'esempio da the introduction to the documentation:

if __name__ == '__main__': 
    pool = Pool(2) 
    results = pool.map(CSVReader.csvreader, string_array[:2]) 
    print(results)