Python multiprocesamiento Pool.map llama a aquire?

Question

Tengo un numpy.array de imágenes de 640x480, cada una de las cuales tiene 630 imágenes de largo. El conjunto total es, por tanto, 630x480x640. Quiero generar una imagen promedio, así como calcular la desviación estándar para cada píxel en las 630 imágenes.

Esto se logra fácilmente por

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

Sin embargo, desde que me estoy quedando esto para 50 o más este tipo de matrices, y tienen un/16 estación de trabajo 8 del núcleo del hilo, pensé que había que conseguir codicioso y paralelizar las cosas con multiprocesamiento.Pool.

Así que hizo lo siguiente:

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 

Sin embargo, vi sólo un pequeño aumento de velocidad. Al poner las declaraciones de impresión en chunk_avg_map, pude determinar que solo se lanzan uno o dos procesos a la vez, en lugar de 16 (como era de esperar).

entonces me encontré con mi código a través cprofile en IPython:

%prun current_image_anal.main() 

El resultado indicó que, con mucho, la mayor parte del tiempo la pasó en llamadas a adquirir:

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

que entiendo que ser algo que ver con el bloqueo, pero no entiendo por qué mi código estaría haciendo eso. ¿Alguien tiene alguna idea?

[EDIT] Conforme a lo solicitado, aquí es una secuencia de comandos de ejecución poder que demuestra el problema. Puede perfil it por cualquier medio que te gusta, pero cuando lo hice me encontré con que los leones parte del tiempo se recogió con llamadas a adquirir, en lugar de std significan o como yo hubiera esperado .

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

Creo que el problema es que la cantidad de tiempo de CPU que se tarda en procesar cada fragmento es pequeño en relación con la cantidad de tiempo que se necesita para copiar la entrada y salida desde y hacia los procesos de trabajo. Modifiqué su código de ejemplo para dividir la salida en 16 partes pares e imprimir la diferencia en el tiempo de CPU (time.clock()) entre cuando comienza y termina una ejecución de chunk_avg_map(). En mi sistema, cada ejecución individual tomó poco menos de un segundo de tiempo de CPU, pero el uso de tiempo total de la CPU para el grupo de procesos (sistema + tiempo de usuario) fue más de 38 segundos. Una aparente sobrecarga de 0,75 segundos por cada copia trozo deja sus cálculos programa que realiza sólo ligeramente más rápido que multiprocessing puede entregar los datos, lo que lleva a sólo dos procesos de trabajo cada vez es utilizar a la vez.

Si modifico el código de tal manera que los "datos de entrada" es sólo xrange(16) y construir la matriz aleatoria dentro de chunk_avg_map() entonces ver la caída sysem + usuario de tiempo de alrededor de 19 segundos y los 16 procesos de trabajo que ejecutan al mismo tiempo.