Python multiprocessing Pool.map sta chiamando aquire?

Question

Ho un file numpy.array di 640x480, ognuna delle quali è lunga 630 immagini. L'array totale è quindi 630x480x640. Voglio generare un'immagine media, nonché calcolare la deviazione standard per ogni pixel su tutte le 630 immagini.

Questo è facilmente realizzabile da

avg_image = numpy.mean(img_array, axis=0) 
std_image = numpy.std(img_array, axis=0) 

Tuttavia, dato che corro questo per 50 o giù di lì tali matrici, ed avere una workstation filo 8 core/16, ho pensato che avrei avuto avidi e parallelizzare le cose con il multiprocessing.Pool .

Così ho fatto la seguente:

def chunk_avg_map(chunk): 
    #do the processing 
    sig_avg = numpy.mean(chunk, axis=0) 
    sig_std = numpy.std(chunk, axis=0) 
    return([sig_avg, sig_std]) 

def chunk_avg(img_data): 

    #take each row of the image 
    chunks = [img_data[:,i,:] for i in range(len(img_data[0]))] 

    pool = multiprocessing.Pool() 
    result = pool.map(chunk_avg_map, chunks) 
    pool.close() 
    pool.join() 
    return result 

Tuttavia, ho visto solo un piccolo aumento di velocità. Inserendo le dichiarazioni di stampa in chunk_avg_map, sono stato in grado di determinare che solo uno o due processi vengono lanciati alla volta, anziché 16 (come mi aspetterei).

Allora ho fatto funzionare il mio codice attraverso Cprofile in ipython:

%prun current_image_anal.main() 

Il risultato ha indicato che di gran lunga la maggior parte del tempo è stato speso nelle chiamate di acquisire:

ncalls tottime percall cumtime percall filename:lineno(function) 
    1527 309.755 0.203 309.755 0.203 {built-in method acquire} 

che capisco di essere qualcosa fare con il blocco, ma non capisco perché il mio codice lo farebbe. Qualcuno ha qualche idea?

[EDIT] Come richiesto, ecco uno script eseguibile che dimostra il problema. Puoi tracciarlo con qualsiasi mezzo tu voglia, ma quando l'ho fatto ho scoperto che la quota dei leoni parte del tempo era occupata con le chiamate da acquisire, piuttosto che dire o std come mi sarei aspettato .

#!/usr/bin/python 
import numpy 
import multiprocessing 

def main(): 
    fake_images = numpy.random.randint(0,2**14,(630,480,640)) 
    chunk_avg(fake_images) 

def chunk_avg_map(chunk): 
    #do the processing 
    sig_avg = numpy.mean(chunk, axis=0) 
    sig_std = numpy.std(chunk, axis=0) 
    return([sig_avg, sig_std]) 

def chunk_avg(img_data): 

    #take each row of the image 
    chunks = [img_data[:,i,:] for i in range(len(img_data[0]))] 

    pool = multiprocessing.Pool() 
    result = pool.map(chunk_avg_map, chunks) 
    pool.close() 
    pool.join() 
    return result 

if __name__ == "__main__": 
    main() 

Answer 1

Credo il problema è che la quantità di tempo di CPU necessario per elaborare ciascun blocco è piccola rispetto alla quantità di tempo necessario per copiare l'ingresso e l'uscita da e verso i processi di lavoro. Ho modificato il codice di esempio per dividere l'output in 16 parti uguali e per stampare la differenza di tempo della CPU (time.clock()) tra quando inizia e termina una corsa di chunk_avg_map(). Sul mio sistema ogni singola esecuzione richiedeva poco meno di un secondo del tempo di CPU, ma l'utilizzo complessivo della CPU per il gruppo di processi (sistema + tempo utente) era superiore a 38 secondi. Un evidente sovraccarico di copiatura di 0,75 secondi per chunk lascia il programma in grado di eseguire calcoli solo leggermente più rapidamente di multiprocessing in grado di fornire i dati, consentendo l'utilizzo simultaneo di due soli processi di lavoro.

Se modifico il codice in modo che i "dati di input" è solo xrange(16) e costruire la matrice casuale all'interno chunk_avg_map() poi vedere la goccia sysem + utente tempo di circa 19 secondi e tutti i 16 processi di lavoro eseguono contemporaneamente.