Multiprocessing Python Pool.map appelle aquire?

Question

J'ai un fichier numérique de 640x480 images, chacune ayant une longueur de 630 images. Le tableau total est donc 630x480x640. Je veux générer une image moyenne, ainsi que calculer l'écart-type pour chaque pixel dans toutes les 630 images.

Ceci est facilement accompli par

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

Cependant, depuis que je suis en cela pour 50 ou si de tels réseaux, et un poste de travail de fil 8 cœurs/16, je me suis dit que je serais cupide et paralléliser les choses avec multiprocessing.Pool.

J'ai donc fait ce qui suit:

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 

Cependant, j'ai vu qu'une petite speedup. En mettant les déclarations d'impression dans chunk_avg_map j'ai pu déterminer que seulement un ou deux processus sont lancés à la fois, au lieu de 16 (comme je l'attendre).

J'ai alors couru mon code par cprofile dans ipython:

%prun current_image_anal.main() 

Le résultat a indiqué que, de loin, le plus de temps a été consacré à des appels à acquérir:

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

Ce que je comprends être quelque chose à faire avec le verrouillage, mais je ne comprends pas pourquoi mon code le ferait. Quelqu'un a-t-il une idée?

[EDIT] Comme l'a demandé, voici un script run-mesure qui illustre le problème. Vous pouvez profiler par ce que signifie que vous aimez, mais quand je l'ai fait je l'ai trouvé que les lions part du temps a été repris avec des appels d'acquérir, plutôt que de dire ou std comme je l'aurais attendu.

#!/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

Je crois que le problème est que la quantité de temps CPU nécessaire pour traiter chaque morceau est faible par rapport à la quantité de temps qu'il faut pour copier l'entrée et de sortie et des processus de travail. J'ai modifié votre code par exemple pour diviser la sortie en 16 morceaux et même d'imprimer la différence dans le temps CPU (time.clock()) entre le moment où une série de chunk_avg_map() commence et se termine. Sur mon système, chaque exécution prenait un peu moins d'une seconde de temps processeur, mais l'utilisation du temps processeur global pour le groupe de processus (système + temps utilisateur) était supérieure à 38 secondes. Une surcharge de copie de 0,75 seconde apparente par bloc laisse votre programme effectuer des calculs à peine plus rapidement que multiprocessing peut fournir les données, ce qui entraîne l'utilisation simultanée de deux processus de travail.

Si je modifie le code de telle sorte que les « données d'entrée » est juste xrange(16) et construire le tableau aléatoire dans chunk_avg_map() alors je vois la perte de temps sysem + utilisateur à environ 19 secondes, les 16 processus de travail d'exécution en même temps.