python - lock - Joblib paralelo múltiples cpu más lento que solo

python multiprocessing windows

Question

Acabo de empezar a usar el módulo Joblib y estoy tratando de entender cómo funciona la función paralela. A continuación se muestra un ejemplo de donde la paralelización conduce a tiempos de ejecución más largos, pero no entiendo por qué. Mi tiempo de ejecución en 1 cpu fue de 51 segundos frente a 217 segundos en 2 cpu.

Supongo que ejecutar el bucle en paralelo copiaría las listas a y b a cada procesador. Luego desplace item_n a una cpu y item_n + 1 a la otra cpu, ejecute la función y luego escriba los resultados de nuevo en una lista (en orden). Luego agarra los siguientes 2 elementos y así sucesivamente. Obviamente me estoy perdiendo algo

¿Es este un mal ejemplo o uso de joblib? ¿Simplemente estructuré mal el código?

Aquí está el ejemplo:

import numpy as np
from matplotlib.path import Path
from joblib import Parallel, delayed

## Create pairs of points for line segments
a = zip(np.random.rand(5000,2),np.random.rand(5000,2))

b = zip(np.random.rand(300,2),np.random.rand(300,2))

## Check if one line segment contains another. 
def check_paths(path, paths):
    for other_path in paths:
        res=''no cross''
        chck = Path(other_path)
        if chck.contains_path(path)==1:
            res= ''cross''
            break
    return res

res = Parallel(n_jobs=2) (delayed(check_paths) (Path(points), a) for points in b)

Answer 1

Además de la respuesta anterior, y para futuras referencias, hay dos aspectos de esta pregunta, y las recientes evoluciones de joblib ayudan con ambas.

Gastos generales de creación de agrupaciones paralelas : el problema aquí es que crear una agrupación paralela es costoso. Fue especialmente costoso aquí, ya que el código no protegido por el " principal " se ejecutó en cada trabajo en la creación del objeto paralelo. En la versión más reciente de Joblib (aún beta), Parallel se puede usar como administrador de contexto para limitar el número de veces que se crea una agrupación y, por lo tanto, el impacto de esta sobrecarga.

Gastos generales de envío : es importante tener en cuenta que el envío de un elemento del bucle for tiene una sobrecarga (mucho más grande que iterar un bucle for sin paralelo). Por lo tanto, si estos elementos de cálculo individuales son muy rápidos, esta sobrecarga dominará el cálculo. En el último joblib, joblib rastreará el tiempo de ejecución de cada trabajo y comenzará a agruparlos si son muy rápidos. Esto limita en gran medida el impacto de la sobrecarga de despacho en la mayoría de los casos (consulte la PR para el banco y la discusión).

Descargo de responsabilidad : Soy el autor original de joblib (solo digo para advertir contra posibles conflictos de interés en mi respuesta, aunque aquí creo que es irrelevante).

Answer 2

En resumen: no puedo reproducir tu problema. Si está en Windows, debe usar un protector para su bucle principal: documentación de joblib.Parallel . El único problema que veo es la sobrecarga de copia de datos, pero parece que los números no son realistas.

En mucho tiempo, aquí están mis tiempos con tu código:

En mi i7 3770k (4 núcleos, 8 hilos) obtengo los siguientes resultados para diferentes n_jobs :

For-loop: Finished in 33.8521318436 sec
n_jobs=1: Finished in 33.5527760983 sec
n_jobs=2: Finished in 18.9543449879 sec
n_jobs=3: Finished in 13.4856410027 sec
n_jobs=4: Finished in 15.0832719803 sec
n_jobs=5: Finished in 14.7227740288 sec
n_jobs=6: Finished in 15.6106669903 sec

Así que hay una ganancia en el uso de múltiples procesos. Sin embargo, aunque tengo cuatro núcleos, la ganancia ya se satura en tres procesos. Así que supongo que el tiempo de ejecución en realidad está limitado por el acceso a la memoria en lugar del tiempo del procesador.

Debe observar que los argumentos para cada entrada de bucle individual se copian al proceso que lo ejecuta. Esto significa que copia a para cada elemento en b . Eso es ineficaz. Así que en lugar de acceder a la global a . ( Parallel bifurcará el proceso, copiando todas las variables globales a los nuevos procesos generados, por lo que es accesible). Esto me da el siguiente código (con temporización y protección de bucle principal como recomienda la documentación de joblib :

import numpy as np
from matplotlib.path import Path
from joblib import Parallel, delayed
import time
import sys

## Check if one line segment contains another. 

def check_paths(path):
    for other_path in a:
        res=''no cross''
        chck = Path(other_path)
        if chck.contains_path(path)==1:
            res= ''cross''
            break
    return res

if __name__ == ''__main__'':
    ## Create pairs of points for line segments
    a = zip(np.random.rand(5000,2),np.random.rand(5000,2))
    b = zip(np.random.rand(300,2),np.random.rand(300,2))

    now = time.time()
    if len(sys.argv) >= 2:
        res = Parallel(n_jobs=int(sys.argv[1])) (delayed(check_paths) (Path(points)) for points in b)
    else:
        res = [check_paths(Path(points)) for points in b]
    print "Finished in", time.time()-now , "sec"

Resultados de tiempo:

 n_jobs=1: Finished in 34.2845709324 sec
 n_jobs=2: Finished in 16.6254048347 sec
 n_jobs=3: Finished in 11.219119072 sec
 n_jobs=4: Finished in 8.61683392525 sec
 n_jobs=5: Finished in 8.51907801628 sec
 n_jobs=6: Finished in 8.21842098236 sec
 n_jobs=7: Finished in 8.21816396713 sec
 n_jobs=8: Finished in 7.81841087341 sec

La saturación ahora se movió ligeramente a n_jobs=4 que es el valor que se espera.

check_paths realiza varios cálculos redundantes que pueden eliminarse fácilmente. En primer lugar, para todos los elementos en other_paths=a la Path(...) de la línea Path(...) se ejecuta en cada llamada. Precalcular eso. En segundo lugar, la cadena res=''no cross'' se escribe en cada giro de bucle, aunque solo puede cambiar una vez (seguido de una ruptura y retorno). Mueve la línea delante del bucle. Entonces el código se ve así:

import numpy as np
from matplotlib.path import Path
from joblib import Parallel, delayed
import time
import sys

## Check if one line segment contains another. 

def check_paths(path):
    #global a
    #print(path, a[:10])
    res=''no cross''
    for other_path in a:
        if other_path.contains_path(path)==1:
            res= ''cross''
            break
    return res

if __name__ == ''__main__'':
    ## Create pairs of points for line segments
    a = zip(np.random.rand(5000,2),np.random.rand(5000,2))
    a = [Path(x) for x in a]

    b = zip(np.random.rand(300,2),np.random.rand(300,2))

    now = time.time()
    if len(sys.argv) >= 2:
        res = Parallel(n_jobs=int(sys.argv[1])) (delayed(check_paths) (Path(points)) for points in b)
    else:
        res = [check_paths(Path(points)) for points in b]
    print "Finished in", time.time()-now , "sec"

con tiempos:

n_jobs=1: Finished in 5.33742594719 sec
n_jobs=2: Finished in 2.70858597755 sec
n_jobs=3: Finished in 1.80810618401 sec
n_jobs=4: Finished in 1.40814709663 sec
n_jobs=5: Finished in 1.50854086876 sec
n_jobs=6: Finished in 1.50901818275 sec
n_jobs=7: Finished in 1.51030707359 sec
n_jobs=8: Finished in 1.51062297821 sec

Un nodo lateral en su código, aunque realmente no he seguido su propósito, ya que no estaba relacionado con su pregunta, contains_path solo devolverá True if this path completely contains the given path. (ver documentation ). Por lo tanto, su función básicamente siempre devolverá no cross dada la entrada aleatoria.