Multiprocesamiento de Python: entendiendo la lógica detrás de `chunksize`

python-3.x parallel-processing (3)

¿Qué factores determinan un argumento de chunksize óptimo para métodos como multiprocessing.Pool.map() ? El método .map() parece utilizar una heurística arbitraria para su tamaño predeterminado (se explica a continuación); ¿Qué motiva esa elección y hay un enfoque más reflexivo basado en alguna situación / configuración particular?

Ejemplo - decir que soy:

Pasar un iterable a .map() que tiene ~ 15 millones de elementos;
Trabajando en una máquina con 24 núcleos y utilizando los processes = os.cpu_count() predeterminados processes = os.cpu_count() dentro de multiprocessing.Pool() .

Mi ingenua idea es darles a cada uno de los 24 trabajadores una porción del mismo tamaño, es decir, 15_000_000 / 24 o 625,000. Los trozos grandes deberían reducir la rotación / gastos generales al tiempo que utilizan a todos los trabajadores. Pero parece que a esto le faltan algunos inconvenientes potenciales de entregar lotes grandes a cada trabajador. ¿Es esta una imagen incompleta, y qué me estoy perdiendo?

Parte de mi pregunta proviene de la lógica predeterminada de si chunksize=None : tanto .map() como .starmap() llaman a .map_async() , que tiene este aspecto:

def _map_async(self, func, iterable, mapper, chunksize=None, callback=None, error_callback=None): # ... (materialize `iterable` to list if it''s an iterator) if chunksize is None: chunksize, extra = divmod(len(iterable), len(self._pool) * 4) # ???? if extra: chunksize += 1 if len(iterable) == 0: chunksize = 0

¿Cuál es la lógica detrás de divmod(len(iterable), len(self._pool) * 4) ? Esto implica que el tamaño del trozo estará más cerca de 15_000_000 / (24 * 4) == 156_250 . ¿Cuál es la intención al multiplicar len(self._pool) por 4?

Esto hace que el tamaño resultante sea un factor 4 más pequeño que mi "lógica ingenua" desde arriba, que consiste en simplemente dividir la longitud del iterable por el número de trabajadores en pool._pool .

Por último, también hay un snippet de los documentos de Python en .imap() que impulsa aún más mi curiosidad:

El argumento chunksize es el mismo que el utilizado por el método map() . Por mucho tiempo, los iterables que usan un gran valor para chunksize pueden hacer que el trabajo se complete mucho más rápido que usar el valor predeterminado de 1.

_{Respuesta relacionada que es útil pero un nivel demasiado alto:
multiprocesamiento de Python: ¿por qué los bloques grandes son más lentos?} _.

Sobre esta respuesta

Esta respuesta es la Parte II de la respuesta aceptada above .

7. Naive vs algoritmo de tamaño de grupo

Antes de entrar en detalles, considera los dos gifs a continuación. Para un rango de diferentes longitudes iterable , muestran cómo los dos algoritmos comparados dividen el iterable pasado (será una secuencia para ese entonces) y cómo las tareas resultantes podrían distribuirse. El orden de los trabajadores es aleatorio y el número de tareas distribuidas por trabajador en realidad puede diferir de estas imágenes para tareas ligeras o tareas en un escenario amplio. Como se mencionó anteriormente, la sobrecarga tampoco se incluye aquí. Sin embargo, para cálculos lo suficientemente pesados en un escenario denso con tamaños de datos transmitidos despreciables, los cálculos reales dibujan una imagen muy similar.

Como se muestra en el capítulo " 5. Algoritmo de tamaño de bloque de Pool ", con el algoritmo de tamaño de bloque de Pool, el número de trozos se estabilizará en n_chunks == n_workers * 4 para iterables lo suficientemente grandes, mientras continúa cambiando entre n_chunks == n_workers y n_chunks == n_workers + 1 con el enfoque ingenuo. Para el algoritmo ingenuo se aplica: Debido a que n_chunks % n_workers == 1 es True para n_chunks == n_workers + 1 , se n_chunks == n_workers + 1 una nueva sección donde solo se empleará un solo trabajador.

Algoritmo de tamaño de ingenuo:

Podría pensar que creó tareas en el mismo número de trabajadores, pero esto solo será cierto para los casos en que no haya resto para len_iterable / n_workers . Si no es un resto, habrá una nueva sección con sólo una tarea para un solo trabajador. En ese punto su computación ya no será paralela.

Abajo puede ver una figura similar a la que se muestra en el capítulo 5, pero que muestra la cantidad de secciones en lugar de la cantidad de partes. Para el algoritmo de tamaño completo de Pool ( n_pool2 ), n_sections se estabilizará en el infame factor codificado 4 . Para el algoritmo ingenuo, n_sections se alternará entre uno y dos.

Para el algoritmo de tamaño de Pool, la estabilización a n_chunks = n_workers * 4 través del tratamiento adicional antes mencionado , evita la creación de una nueva sección aquí y mantiene la participación de ralentí limitada a un trabajador durante el tiempo suficiente. No solo eso, sino que el algoritmo seguirá reduciendo el tamaño relativo de la acción de ralentí , lo que lleva a un valor RDE que converge hacia el 100%.

"Lo suficientemente largo" para n_workers=4 es len_iterable=210 por ejemplo. Para que sean iguales o mayores que eso, el reparto de ralentí se limitará a un trabajador, un rasgo que originalmente se perdió debido a la 4 multiplicación múltiple en el algoritmo de tamaño en primer lugar.

El ingenuo algoritmo de tamaño también converge hacia el 100%, pero lo hace más lento. El efecto convergente depende únicamente del hecho de que la parte relativa de la cola se contrae en los casos en que habrá dos secciones. Esta cola con un solo trabajador empleado está limitada a la longitud del eje x n_workers - 1 , el resto máximo posible para len_iterable / n_workers .

¿En qué difieren los valores reales de RDE para el ingenuo y el algoritmo de tamaño de Pool?

Abajo encontrará dos mapas de calor que muestran los valores de RDE para todas las longitudes iterables hasta 5000, para todos los números de trabajadores desde 2 hasta 100. La escala de color va de 0.5 a 1 (50% -100%). Notará áreas mucho más oscuras (valores RDE más bajos) para el algoritmo ingenuo en el mapa de calor izquierdo. En contraste, el algoritmo de tamaño de Pool de la derecha dibuja una imagen mucho más brillante.

El gradiente diagonal de las esquinas oscuras de la parte inferior izquierda frente a las esquinas brillantes de la parte superior derecha muestra nuevamente la dependencia de la cantidad de trabajadores para lo que se denomina un "iterable largo".

¿Qué tan malo puede ser con cada algoritmo?

Con el algoritmo de tamaño de bloque de Pool, un valor RDE de 81.25% es el valor más bajo para el rango de trabajadores y las longitudes iterables especificadas anteriormente:

Con el ingenuo algoritmo chunksize, las cosas pueden ponerse mucho peor. La RDE calculada más baja aquí es 50.72%. En este caso, ¡casi la mitad del tiempo de cómputo solo se está ejecutando un trabajador! Así que, cuidado, orgullosos dueños de Knights Landing . ;)

8. Verificación de la realidad

En los capítulos anteriores consideramos un modelo simplificado para el problema de distribución puramente matemático, despojado de los detalles esenciales que hacen del multiprocesamiento un tema tan espinoso en primer lugar. Para comprender mejor hasta qué punto el Modelo de Distribución (DM) por sí solo puede contribuir a explicar la utilización observada del trabajador en la realidad, ahora examinaremos los Horarios Paralelos dibujados por cálculos reales .

Preparar

Las siguientes gráficas tratan con ejecuciones paralelas de una función ficticia simple, unida a la CPU, que se invoca con diversos argumentos para que podamos observar cómo varía la Programación Paralelo dibujada en función de los valores de entrada. El "trabajo" dentro de esta función consiste solo en iteración sobre un objeto de rango. Esto ya es suficiente para mantener un núcleo ocupado ya que pasamos grandes números. Opcionalmente, la función toma un extra de taskel-unique data que solo se devuelve sin cambios. Dado que cada tarea comprende exactamente la misma cantidad de trabajo, todavía estamos tratando con un escenario denso aquí.

La función está decorada con una envoltura que toma marcas de tiempo con resolución ns (Python 3.7+). Las marcas de tiempo se utilizan para calcular el intervalo de tiempo de un panel de tareas y, por lo tanto, habilitar el dibujo de un Programa paralelo empírico.

@stamp_taskel def busy_foo(i, it, data=None): """Dummy function for CPU-bound work.""" for _ in range(int(it)): pass return i, data def stamp_taskel(func): """Decorator for taking timestamps on start and end of decorated function execution. """ @wraps(func) def wrapper(*args, **kwargs): start_time = time_ns() result = func(*args, **kwargs) end_time = time_ns() return (current_process().name, (start_time, end_time)), result return wrapper

El método de starmap de Pool también está decorado de tal manera que solo el propio starmap-call está cronometrado. "Inicio" y "final" de esta llamada determinan el mínimo y el máximo en el eje x del Programa paralelo producido.

Vamos a observar el cálculo de 40 tareas en cuatro procesos de trabajo en una máquina con estas especificaciones: Python 3.7.1, Ubuntu 18.04.2, Intel® Core ™ i7-2600K CPU @ 3.40GHz x 8

Los valores de entrada que se variarán son el número de iteraciones en el bucle for (30k, 30M, 600M) y el tamaño adicional de los datos de envío (por taskel, númpy-ndarray: 0 MiB, 50 MiB).

... N_WORKERS = 4 LEN_ITERABLE = 40 ITERATIONS = 30e3 # 30e6, 600e6 DATA_MiB = 0 # 50 iterable = [ # extra created data per taskel (i, ITERATIONS, np.arange(int(DATA_MiB * 2**20 / 8))) # taskel args for i in range(LEN_ITERABLE) ] with Pool(N_WORKERS) as pool: results = pool.starmap(busy_foo, iterable)

Las ejecuciones que se muestran a continuación se seleccionaron a mano para tener el mismo orden de trozos, por lo que puede detectar mejor las diferencias en comparación con el Horario Paralelo del Modelo de Distribución, pero no olvide que el orden en que los trabajadores obtienen su tarea no es determinista.

Predicción de DM

Para reiterar, el Modelo de Distribución "predice" un Programa Paralelo como lo hemos visto anteriormente en el capítulo 6.2:

1er RUN: 30k iteraciones y 0 datos MiB por tarea

Nuestra primera carrera aquí es muy corta, las tareas son muy "ligeras". La totalidad de la pool.starmap() llamada solo tomó 14.5 ms en total. Notará que, contrariamente a lo que ocurre con el DM , el ralentí no se limita a la sección de cola, sino que también se produce entre tareas e incluso entre tareas. Eso es porque nuestro horario real aquí naturalmente incluye todo tipo de gastos generales. La inactividad aquí significa simplemente todo lo que está fuera de una tarea. Posible ralentí real durante un taskel no se captura como ya se mencionó anteriormente.

Además, se puede ver que no todos los trabajadores realizan sus tareas al mismo tiempo. Esto se debe al hecho de que todos los trabajadores se alimentan de una fuente compartida inqueue y solo un trabajador puede leer al mismo tiempo. Lo mismo se aplica para el outqueue . Esto puede causar grandes trastornos tan pronto como esté transmitiendo tamaños de datos no marginales, como veremos más adelante.

Además, puede ver que a pesar del hecho de que cada tarea comprende la misma cantidad de trabajo, el tiempo real medido para una tarea varía mucho. Las tareas distribuidas al trabajador-3 y al trabajador-4 necesitan más tiempo que las procesadas por los dos primeros trabajadores. Para esta ejecución, sospecho que se debe a que el turbo boost ya no está disponible en los núcleos para worker-3/4 en ese momento, por lo que procesaron sus tareas con una frecuencia de reloj más baja.

Todo el cálculo es tan ligero que el hardware o los factores de caos introducidos por el sistema operativo pueden sesgar drásticamente el PS . El cálculo es una "hoja en el viento" y la predicción de DM tiene poca importancia, incluso para un escenario teóricamente adecuado.

2do RUN: 30M iteraciones y 0 datos MiB por tarea

Aumentar el número de iteraciones en el bucle for de 30,000 a 30 millones, da como resultado un Programa Paralelo real que está cerca de una coincidencia perfecta con la pronosticada por los datos proporcionados por el DM , hurray! El cómputo por tarea ahora es lo suficientemente pesado como para marginar las partes inactivas al inicio y en el medio, permitiendo que solo se vea la gran proporción de inactividad que predijo el DM .

3er RUN: 30M iteraciones y 50 datos MiB por tarea

Manteniendo las iteraciones de 30M, pero además, el envío de 50 MiB por tarea hace que la imagen vuelva a aparecer. Aquí el efecto de cola es bien visible. Worker-4 necesita esperar más tiempo para su segunda tarea que Worker-1. ¡Ahora imagina este horario con 70 trabajadores!

En caso de que las tareas sean computacionalmente ligeras, pero proporcionen una cantidad notable de datos como carga útil, el cuello de botella de una sola cola compartida puede evitar cualquier beneficio adicional de agregar más trabajadores al Grupo, incluso si están respaldados por núcleos físicos. En tal caso, Worker-1 podría hacerse con su primera tarea y esperando una nueva incluso antes de que Worker-40 haya recibido su primera tarea.

Ahora debería ser obvio por qué los tiempos de cómputo Pool no siempre disminuyen linealmente con el número de trabajadores. El envío de cantidades relativamente grandes de datos puede llevar a escenarios en los que la mayor parte del tiempo se dedica a esperar a que los datos se copien en el espacio de direcciones de un trabajador y solo se pueda alimentar a un trabajador a la vez.

4to RUN: 600M iteraciones y 50 datos MiB por tarea

Aquí enviamos 50 MiB nuevamente, pero aumentamos el número de iteraciones de 30M a 600M, lo que eleva el tiempo total de cómputo de 10 sa 152 s. El Horario Paralelo dibujado nuevamente , está cerca de una coincidencia perfecta con la predicha, la sobrecarga a través de la copia de datos queda marginalizada.

9. Conclusión

La multiplicación discutida 4 aumenta la flexibilidad de programación, pero también aprovecha la desigualdad en las distribuciones de tareas. Sin esta multiplicación, el reparto de ralentí se limitaría a un solo trabajador, incluso para iterables cortos (para DM con escenario denso) El algoritmo de tamaño de grupo requiere que los datos de entrada sean de cierto tamaño para recuperar ese rasgo.

Como es de esperar que esta respuesta se haya demostrado, el algoritmo de tamaño de bloque de Pool conduce a una mejor utilización del núcleo en promedio en comparación con el enfoque ingenuo, al menos para el caso promedio y mientras no se considere la sobrecarga. El algoritmo ingenuo aquí puede tener una Eficiencia de Distribución (DE) tan baja como ~ 51%, mientras que el algoritmo de tamaño de Pool tiene un mínimo de ~ 81%. Sin embargo, DE no comprende la sobrecarga de paralelización (PO) como IPC. El Capítulo 8 ha demostrado que la DE aún puede tener un gran poder predictivo para el escenario denso con una sobrecarga marginal.

A pesar del hecho de que el algoritmo de tamaño de Pool logra una mayor DE en comparación con el enfoque ingenuo, no proporciona distribuciones de tareas óptimas para cada constelación de entrada. Si bien un simple algoritmo de fragmentación estática no puede optimizar (ECA, incluida la sobrecarga), la Eficiencia de Paralelización (PE), no hay una razón intrínseca por la que no siempre pueda proporcionar una Eficiencia de Distribución Relativa (RDE) del 100%, es decir, la misma DE como con chunksize=1 . Un simple algoritmo de tamaño consiste solo en matemática básica y es libre de "cortar el pastel" de cualquier manera.

A diferencia de la implementación de un "igual-tamaño-fragmentación" algoritmo de piscina, un algoritmo "incluso de tamaño-fragmentación" proporcionaría una RDE del 100% para todos los len_iterable / n_workers combinación. Un algoritmo de fragmentación de tamaño uniforme sería un poco más complicado de implementar en la fuente de Pool, pero se puede modular sobre el algoritmo existente simplemente empaquetando las tareas externamente (lo vincularé desde aquí en caso de que coloque una Q / A en como hacer eso).

Respuesta corta

El algoritmo de tamaño de Pool es una heurística. Proporciona una solución simple para todos los escenarios de problemas imaginables que intenta incluir en los métodos de Pool. Como consecuencia, no se puede optimizar para ningún escenario específico .

El algoritmo divide arbitrariamente lo iterable en aproximadamente cuatro veces más trozos que el enfoque ingenuo. Más trozos significan más sobrecarga, pero mayor flexibilidad de programación. Como se mostrará esta respuesta, esto conduce a una mayor utilización de los trabajadores en promedio, pero sin la garantía de un tiempo de cómputo general más corto para cada caso.

"Es bueno saberlo", podría pensar, "pero, ¿cómo me ayuda esto con mis problemas concretos de multiprocesamiento?" Bueno, no es así. La respuesta corta más honesta es: "no hay una respuesta corta", "el multiprocesamiento es complejo" y "depende". Un síntoma observado puede tener diferentes raíces, incluso para escenarios similares.

Esta respuesta trata de proporcionarle conceptos básicos que lo ayudan a obtener una imagen más clara de la caja negra de programación de Pool. También trata de brindarle algunas herramientas básicas para reconocer y evitar posibles acantilados en la medida en que estén relacionados con el tamaño de los trozos.

Tabla de contenido

Parte I

Definiciones

Metas de paralelización

Escenarios de paralelización

Riesgos de Chunksize> 1

Algoritmo de tamaño de grupo

Cuantificando la eficiencia del algoritmo

6.1 modelos

6.2 Horario paralelo

6.3 Eficiencias

6.3.1 Eficiencia de distribución absoluta (ADE)

6.3.2 Eficiencia de distribución relativa (RDE)

Parte II

Ingenuo contra el algoritmo de chunksize

Verificación de la realidad

Conclusión

Es necesario aclarar algunos términos importantes primero.

1. Definiciones

Pedazo

Una parte aquí es una parte del argumento iterable especificado en una llamada de método de grupo. El tema de esta respuesta es cómo se calcula el tamaño del trozo y qué efectos puede tener esto.

Tarea

La representación física de una tarea en un proceso de trabajo en términos de datos se puede ver en la siguiente figura.

La figura muestra un ejemplo de llamada a pool.map() , que se muestra a lo largo de una línea de código, tomada de la función multiprocessing.pool.worker , donde se desempaqueta una tarea que se lee de la inqueue . worker es la función principal subyacente en MainThread de un pool-worker-process. El argumento func especificado en el método de agrupación solo coincidirá con la variable func dentro de la función worker para métodos de llamada única como apply_async y para imap con chunksize=1 . Para el resto de los métodos de agrupación con un chunksize la función de func procesamiento será una función de mapeador ( mapstar o starmapstar ). Esta función mapea el parámetro de func especificado por el usuario en cada elemento de la parte transmitida del iterable (-> "tareas de mapa"). El tiempo que esto lleva, define una tarea también como una unidad de trabajo .

Taskel

Si bien el uso de la palabra "tarea" para el procesamiento completo de un fragmento coincide con el código dentro de multiprocessing.pool , no hay ninguna indicación de cómo una sola llamada a la func especificada por el usuario, con un elemento del fragmento como argumento (s) ), debe ser referido. Para evitar confusiones que surjan de los conflictos de nombres (piense en maxtasksperchild para el método __init__ de __init__ ), esta respuesta se referirá a las unidades de trabajo individuales dentro de una tarea como taskel .

Un taskel (de task + elemento ) es la unidad de trabajo más pequeña dentro de una tarea . Es la ejecución única de la función especificada con la func -parámetro de un método Pool agrupación, llamada con argumentos obtenidos de un solo elemento del fragmento transmitido. Una tarea consiste en tareas de chunksize grande .

Sobrecarga de paralelización (PO)

La PO consiste en la sobrecarga interna de Python y la sobrecarga para la comunicación entre procesos (IPC). La sobrecarga por tarea dentro de Python viene con el código necesario para empaquetar y desempaquetar las tareas y sus resultados. La sobrecarga de IPC viene con la sincronización necesaria de hilos y la copia de datos entre diferentes espacios de direcciones (se necesitan dos pasos de copia: padre -> cola -> hijo). La cantidad de sobrecarga de IPC depende del tamaño del sistema operativo, hardware y datos, lo que dificulta las generalizaciones sobre el impacto.

2. Metas de paralelización

Cuando se usa el multiprocesamiento, nuestro objetivo general (obviamente) es minimizar el tiempo total de procesamiento para todas las tareas. Para alcanzar este objetivo general, nuestro objetivo técnico debe ser optimizar la utilización de los recursos de hardware .

Algunos objetivos secundarios importantes para lograr el objetivo técnico son:

minimizar la sobrecarga de paralelización (lo más famoso, pero no solo: IPC )
alta utilización en todos los cpu-cores
Mantener el uso de memoria limitado para evitar que el sistema operativo se pagine en exceso

Al principio, las tareas deben ser lo suficientemente pesadas (intensivas) desde el punto de vista informático, para recuperar el PO que tenemos que pagar por la paralelización. La relevancia de la PO disminuye a medida que aumenta el tiempo de cálculo absoluto por tarea. O, para decirlo de otra manera, cuanto mayor sea el tiempo de cálculo absoluto por tarea para su problema, menos relevante tendrá la necesidad de reducir la PO. Si su cálculo tomará horas por tarea, la sobrecarga de IPC será insignificante en comparación. La principal preocupación aquí es evitar los procesos de trabajo inactivos después de que se hayan distribuido todas las tareas. Manteniendo todos los núcleos cargados significa que estamos paralelizando tanto como sea posible.

3. Escenarios de paralelización

¿Qué factores determinan un argumento de tamaño óptimo para métodos como multiprocessing.Pool.map ()?

El factor principal en cuestión es cuánto tiempo de cálculo puede variar entre nuestras tareas individuales. Para nombrarlo, la elección de un tamaño óptimo se determina mediante ...

Coeficiente de variación ( CV ) para tiempos de cálculo por tarea.

Los dos escenarios extremos en una escala, siguiendo el alcance de esta variación, son:

Todas las tareas necesitan exactamente el mismo tiempo de cálculo.
Un taskel podría tardar segundos o días en terminar.

Para una mejor memorización, me referiré a estos escenarios como:

Escenario denso
Escenario amplio

Escenario denso

En un escenario denso , sería conveniente distribuir todos los grupos de tareas a la vez, para mantener al mínimo el intercambio de contexto y el IPC necesarios. Esto significa que queremos crear solo tantos fragmentos, como procesos de trabajo hay. Como ya se indicó anteriormente, el peso de la OP aumenta con los tiempos de cómputo más cortos por tarea.

Para un rendimiento máximo, también queremos que todos los procesos de trabajo estén ocupados hasta que se procesen todas las tareas (sin trabajadores inactivos). Para este objetivo, los trozos distribuidos deben ser del mismo tamaño o cerca.

Escenario amplio

El ejemplo principal para un escenario amplio sería un problema de optimización, en el que los resultados convergen rápidamente o el cálculo puede demorar horas, si no días. Por lo general, no es predecible qué combinación de "tareas ligeras" y "tareas pesadas" contendrá una tarea en tal caso, por lo tanto, no es recomendable distribuir demasiados tareas en un lote de tareas a la vez. Distribuir menos tareas al mismo tiempo de lo posible, significa aumentar la flexibilidad de programación. Esto es necesario aquí para alcanzar nuestro objetivo secundario de alta utilización de todos los núcleos.

Si los métodos Pool agrupación, de forma predeterminada, estuvieran totalmente optimizados para el escenario denso, crearían cada vez menos tiempos óptimos para cada problema ubicado más cerca del escenario amplio.

4. Riesgos de Chunksize> 1

Considere este ejemplo de pseudocódigo simplificado de un Escenario amplio que se puede ejecutar, que queremos pasar a un método de agrupación:

good_luck_iterable = [60, 60, 86400, 60, 86400, 60, 60, 84600]

En lugar de los valores reales, pretendemos ver el tiempo de cálculo necesario en segundos, para simplificar solo 1 minuto o 1 día. Suponemos que el grupo tiene cuatro procesos de trabajo (en cuatro núcleos) y que el chunksize se establece en 2 . Debido a que se mantendrá la orden, los trozos que se envíen a los trabajadores serán estos:

[(60, 60), (86400, 60), (86400, 60), (60, 84600)]

Ya que tenemos suficientes trabajadores y el tiempo de cómputo es lo suficientemente alto, podemos decir que, en primer lugar, cada proceso de trabajo hará que una parte funcione. (Esto no tiene que ser el caso para completar tareas rápidamente). Además, podemos decir que todo el procesamiento tomará alrededor de 86400 + 60 segundos, porque ese es el tiempo de cómputo total más alto para una parte en este escenario artificial y distribuimos partes solo una vez.

Ahora considere este iterable, que tiene un solo elemento que cambia su posición en comparación con el iterable anterior:

bad_luck_iterable = [60, 60, 86400, 86400, 60, 60, 60, 84600]

... y los trozos correspondientes:

[(60, 60), (86400, 86400), (60, 60), (60, 84600)]

¡Solo mala suerte con la clasificación de nuestro iterable casi el doble (86400 + 86400) nuestro tiempo total de procesamiento! El trabajador que recibe la pieza viciosa (86400, 86400) está impidiendo que la segunda tarea pesada en su tarea se distribuya a uno de los trabajadores inactivos que ya terminó con sus (60, 60) piezas. Obviamente, no arriesgaríamos un resultado tan desagradable si establecemos chunksize=1 .

Este es el riesgo de tamaños más grandes. Con los tamaños más altos, cambiamos la flexibilidad de programación por menos gastos generales y, en casos como el anterior, es un mal negocio.

Cómo veremos en el capítulo 6. La cuantificación de la eficiencia del algoritmo , los tamaños más grandes también pueden conducir a resultados subóptimos para los escenarios densos .

5. Algoritmo de tamaño de la piscina

A continuación encontrará una versión ligeramente modificada del algoritmo dentro del código fuente. Como puede ver, corté la parte inferior y la envolví en una función para calcular el argumento de chunksize externo. También reemplacé 4 con un parámetro factor y subcontraté las llamadas len() .

# mp_utils.py def calc_chunksize(n_workers, len_iterable, factor=4): """Calculate chunksize argument for Pool-methods. Resembles source-code within `multiprocessing.pool.Pool._map_async`. """ chunksize, extra = divmod(len_iterable, n_workers * factor) if extra: chunksize += 1 return chunksize

Para asegurarnos de que todos estemos en la misma página, esto es lo que hace divmod :

divmod(x, y) es una función incorporada que devuelve (x//y, x%y) . x // y es la división de piso, devolviendo el cociente redondeado hacia abajo desde x / y , mientras que x % y es la operación de módulo devolviendo el resto de x / y . Por lo tanto, por ejemplo, divmod(10, 3) devuelve (3, 1) .

Ahora, cuando vea chunksize, extra = divmod(len_iterable, n_workers * 4) , notará que n_workers aquí es el divisor y en x / y y multiplicación por 4 , sin más ajustes if extra: chunksize +=1 más adelante, lleva a un tamaño de pieza inicial al menos cuatro veces más pequeño (para len_iterable >= n_workers * 4 ) de lo que sería de otra manera.

Para ver el efecto de la multiplicación por 4 en el resultado de tamaño intermedio considere esta función:

def compare_chunksizes(len_iterable, n_workers=4): """Calculate naive chunksize, Pool''s stage-1 chunksize and the chunksize for Pool''s complete algorithm. Return chunksizes and the real factors by which naive chunksizes are bigger. """ cs_naive = len_iterable // n_workers or 1 # naive approach cs_pool1 = len_iterable // (n_workers * 4) or 1 # incomplete pool algo. cs_pool2 = calc_chunksize(n_workers, len_iterable) real_factor_pool1 = cs_naive / cs_pool1 real_factor_pool2 = cs_naive / cs_pool2 return cs_naive, cs_pool1, cs_pool2, real_factor_pool1, real_factor_pool2

La función anterior calcula el tamaño de los trozos ingenuo ( cs_naive ) y el primer paso del tamaño del algoritmo de tamaño de bloque de Pool ( cs_pool1 ), así como el tamaño del bloque completo para el algoritmo de Pool completo ( cs_pool2 ). Además, calcula los factores reales rf_pool1 = cs_naive / cs_pool1 y rf_pool2 = cs_naive / cs_pool2 , que nos dicen cuántas veces los tamaños calculados ingenuamente son más grandes que las versiones internas de Pool.

A continuación se muestran dos figuras creadas con la salida de esta función. La figura de la izquierda solo muestra los tamaños de n_workers=4 hasta una longitud iterable de 500 . La figura de la derecha muestra los valores para rf_pool1 . Para la longitud iterable 16 , el factor real se convierte en >=4 (para len_iterable >= n_workers * 4 ) y su valor máximo es 7 para las longitudes iterables 28-31 . Esa es una desviación masiva del factor original 4 al que converge el algoritmo para iterables más largos. ''Más largo'' aquí es relativo y depende del número de trabajadores especificados.

Recuerde que chunksize cs_pool1 aún carece de ajustes extra con el resto de divmod contenido en cs_pool2 del algoritmo completo.

El algoritmo continúa con:

if extra: chunksize += 1

Ahora, en los casos en los que hay un resto (un extra de la operación divmod-operation), el aumento del tamaño de trozo en 1 obviamente no puede funcionar para cada tarea. Después de todo, si lo fuera, no habría un resto para empezar.

Como puede ver en las figuras a continuación, el " tratamiento adicional " tiene el efecto, que el factor real para rf_pool2 ahora converge hacia 4 desde debajo de 4 y la desviación es algo más suave. Desviación estándar para n_workers=4 y len_iterable=500 caídas desde 0.5233 para rf_pool1 a 0.4115 para rf_pool2 .

Finalmente, aumentar chunksize por 1 tiene el efecto, que la última tarea transmitida solo tiene un tamaño de len_iterable % chunksize or chunksize .

Sin embargo, cuanto más interesante y cómo veremos más adelante, más como consecuencia, se puede observar el efecto del tratamiento adicional para la cantidad de trozos generados ( n_chunks ). Durante el tiempo suficiente, el algoritmo de tamaño de trozos completado de Pool ( n_pool2 en la figura a continuación) estabilizará el número de trozos en n_chunks == n_workers * 4 . Por el contrario, el algoritmo ingenuo (después de un eructo inicial) sigue alternando entre n_chunks == n_workers y n_chunks == n_workers + 1 medida que n_chunks == n_workers + 1 la longitud del iterable.

A continuación, encontrará dos funciones de información mejoradas para Pool''s y el ingenuo chunksize-algorithm. La salida de estas funciones será necesaria en el siguiente capítulo.

# mp_utils.py from collections import namedtuple Chunkinfo = namedtuple( ''Chunkinfo'', [''n_workers'', ''len_iterable'', ''n_chunks'', ''chunksize'', ''last_chunk''] ) def calc_chunksize_info(n_workers, len_iterable, factor=4): """Calculate chunksize numbers.""" chunksize, extra = divmod(len_iterable, n_workers * factor) if extra: chunksize += 1 # `+ (len_iterable % chunksize > 0)` exploits that `True == 1` n_chunks = len_iterable // chunksize + (len_iterable % chunksize > 0) # exploit `0 == False` last_chunk = len_iterable % chunksize or chunksize return Chunkinfo( n_workers, len_iterable, n_chunks, chunksize, last_chunk )

No se confunda con el aspecto probablemente inesperado de calc_naive_chunksize_info . El extra de divmod no se usa para calcular el tamaño del bloque.

def calc_naive_chunksize_info(n_workers, len_iterable): """Calculate naive chunksize numbers.""" chunksize, extra = divmod(len_iterable, n_workers) if chunksize == 0: chunksize = 1 n_chunks = extra last_chunk = chunksize else: n_chunks = len_iterable // chunksize + (len_iterable % chunksize > 0) last_chunk = len_iterable % chunksize or chunksize return Chunkinfo( n_workers, len_iterable, n_chunks, chunksize, last_chunk )

6. Cuantificando la eficiencia del algoritmo

Ahora, después de que vimos cómo la salida del algoritmo de tamaño de Pool ve diferente en comparación con la salida del algoritmo ingenuo ...

¿Cómo saber si el enfoque de Pool realmente mejora algo?
¿Y qué podría ser exactamente algo ?

Como se muestra en el capítulo anterior, para iterables más largos (un mayor número de tareas), el algoritmo de tamaño de bloque de Pool divide aproximadamente el iterable en cuatro veces más trozos que el método ingenuo. Los trozos más pequeños significan más tareas y más tareas significan más Sobrecarga de paralelización (PO) , un costo que debe sopesarse con el beneficio de una mayor flexibilidad de programación (recuérdese "Riesgos de tamaño de archivo> 1" ).

Por razones bastante obvias, el algoritmo básico de tamaño de Pool no puede comparar la flexibilidad de programación con la OP para nosotros. La sobrecarga de IPC depende del tamaño del sistema operativo, hardware y datos. El algoritmo no puede saber en qué hardware ejecutamos nuestro código, ni tiene una idea de cuánto tardará en completarse una tarea. Es una heurística que proporciona una funcionalidad básica para todos los escenarios posibles. Esto significa que no se puede optimizar para ningún escenario en particular. Como se mencionó anteriormente, la PO también se vuelve cada vez menos preocupante con el aumento de los tiempos de cálculo por tarea (correlación negativa).

Cuando recuerdas los Objetivos de paralelización del capítulo 2, un punto fue:

alta utilización en todos los cpu-cores

Como se mencionó anteriormente, el algoritmo de tamaño de Pool puede intentar mejorar es la minimización de los procesos de trabajo inactivos, respectivamente, la utilización de cpu-cores .

Las personas que se preguntan acerca de los núcleos no utilizados / procesos de trabajo inactivos en situaciones en las que usted esperaría que todos los procesos de trabajadores estuvieran ocupados hacen una pregunta repetida sobre SO con respecto al multiprocessing.Pool . Si bien esto puede tener muchas razones, los procesos de trabajo inactivos hacia el final de un cómputo son una observación que podemos hacer a menudo, incluso con Escenarios densos (tiempos de cómputo iguales por tarea) en los casos en que el número de trabajadores no es un divisor del número de trozos ( n_chunks % n_workers > 0 ).

La pregunta ahora es:

¿Cómo podemos traducir prácticamente nuestra comprensión de los chunksizes a algo que nos permita explicar la utilización observada por el trabajador, o incluso comparar la eficiencia de diferentes algoritmos en ese sentido?

6.1 modelos

Para obtener información más detallada aquí, necesitamos una forma de abstracción de cálculos paralelos que simplifique la realidad demasiado compleja hasta un grado manejable de complejidad, al tiempo que conserva el significado dentro de los límites definidos. Tal abstracción se llama modelo . Una implementación de dicho " Modelo de Paralelización" (PM) genera metadatos (marcas de tiempo) asignados por el trabajador como lo harían los cálculos reales, si se recolectaran los datos. Los metadatos generados por el modelo permiten predecir métricas de cálculos paralelos bajo ciertas restricciones.

Uno de los dos submodelos dentro del PM aquí definido es el Modelo de Distribución (DM) . El DM explica cómo se distribuyen las unidades atómicas de trabajo (tareas) en tiempo y trabajadores paralelos , cuando no se consideran otros factores que no sean el algoritmo de tamaño de chunks respectivo, el número de trabajadores, el valor de entrada (número de tareas) y su duración de cálculo. . Esto significa que no se incluye ninguna forma de sobrecarga.

Para obtener un PM completo, el DM se amplía con un modelo de sobrecarga (OM) , que representa varias formas de sobrecarga de paralelización (PO) . Dicho modelo debe calibrarse para cada nodo individualmente (dependencias de hardware, sistema operativo). La cantidad de formas de sobrecarga que se representan en un OM se deja abierta, por lo que pueden existir múltiples OM con diversos grados de complejidad. El nivel de precisión que las necesidades de OM implementadas se determina por el peso total de PO para el cálculo específico. Las tareas más cortas conducen a un mayor peso de PO , lo que a su vez requiere un OM más preciso si intentamos predecir las eficiencias de paralelización (PE) .

6.2 Horario Paralelo (PS)

La Programación Paralela es una representación bidimensional del cálculo paralelo, donde el eje x representa el tiempo y el eje y representa un conjunto de trabajadores paralelos. El número de trabajadores y el tiempo total de cálculo marcan la extensión de un rectángulo, en el que se dibujan rectángulos más pequeños. Estos rectángulos más pequeños representan unidades atómicas de trabajo (tareas).

A continuación encontrará la visualización de un PS dibujado con datos del DM del algoritmo de tamaño de bloque para el escenario denso .

El eje x se divide en unidades de tiempo iguales, donde cada unidad representa el tiempo de cálculo que requiere una tarea.
El eje y se divide en el número de procesos de trabajo que utiliza el grupo.
Aquí se muestra un panel de tareas como el rectángulo de color cian más pequeño, puesto en una línea de tiempo (un programa) de un proceso de trabajo anónimo.
Una tarea es una o varias tareas en una línea de tiempo de trabajador resaltada continuamente con el mismo tono.
Las unidades de tiempo de inactividad se representan a través de azulejos de color rojo.
El horario paralelo se divide en secciones. La última sección es la sección de cola.

Los nombres de las partes compuestas se pueden ver en la siguiente imagen.

En un PM completo que incluye un OM , el recurso compartido de ralentí no se limita a la cola, sino que también incluye espacio entre tareas e incluso entre tareas.

6.3 Eficiencias

Nota:

Desde las versiones anteriores de esta respuesta, "Eficacia de paralelización (PE)" ha sido renombrada a "Eficiencia de distribución (DE)". PE ahora se refiere a la sobrecarga, incluida la eficiencia.

Los modelos introducidos anteriormente permiten cuantificar la tasa de utilización del trabajador. Podemos distinguir:

Eficiencia de distribución (DE) : calculada con la ayuda de un DM (o un método simplificado para el escenario denso ).
Eficiencia de paralelización (PE) : calculada con la ayuda de un PM calibrado (predicción) o calculada a partir de metadatos de cálculos reales.

Es importante tener en cuenta que las eficiencias calculadas no se correlacionan automáticamente con una computación general más rápida para un problema de paralelización dado. La utilización del trabajador en este contexto solo distingue entre un trabajador que tiene un grupo de tareas iniciado pero no terminado y un trabajador que no tiene un grupo de tareas tan "abierto". Esto significa que no se registra la posible inactividad durante el período de tiempo de un taskel.

Todas las eficiencias mencionadas anteriormente se obtienen básicamente calculando el cociente de la división Ocupado / Horario paralelo . La diferencia entre DE y PE viene con la participación ocupada que ocupa una parte más pequeña de la programación paralela general para el PM de gastos generales.

Esta respuesta solo discutirá un método simple para calcular DE para el escenario denso. Esto es lo suficientemente adecuado para comparar diferentes algoritmos de tamaño, ya que ...

... el DM es la parte del PM , que cambia con diferentes algoritmos de tamaño empleado.
... el escenario denso con duraciones de cómputo iguales por panel de tarea representa un "estado estable", para el cual estos intervalos de tiempo se eliminan de la ecuación. Cualquier otro escenario solo llevaría a resultados aleatorios, ya que la ordenación de las tareas sería importante.

6.3.1 Eficiencia de distribución absoluta (ADE)

Esta eficiencia básica se puede calcular en general al dividir la Participación ocupada a través de todo el potencial del Programa paralelo :

Eficiencia de distribución absoluta (ADE) = Participación ocupada / Programación paralela

Para el escenario denso , el código de cálculo simplificado se ve así:

# mp_utils.py def calc_ade(n_workers, len_iterable, n_chunks, chunksize, last_chunk): """Calculate Absolute Distribution Efficiency (ADE). `len_iterable` is not used, but contained to keep a consistent signature with `calc_rde`. """ if n_workers == 1: return 1 potential = ( ((n_chunks // n_workers + (n_chunks % n_workers > 1)) * chunksize) + (n_chunks % n_workers == 1) * last_chunk ) * n_workers n_full_chunks = n_chunks - (chunksize > last_chunk) taskels_in_regular_chunks = n_full_chunks * chunksize real = taskels_in_regular_chunks + (chunksize > last_chunk) * last_chunk ade = real / potential return ade

Si no hay un recurso compartido inactivo , el recurso compartido ocupado será igual al horario paralelo , por lo tanto, obtenemos un ADE del 100%. En nuestro modelo simplificado, este es un escenario en el que todos los procesos disponibles estarán ocupados durante todo el tiempo necesario para procesar todas las tareas. En otras palabras, todo el trabajo se paraliza efectivamente al 100 por ciento.

Pero, ¿por qué sigo refiriéndome al PE como PE absoluto aquí?

Para comprenderlo, debemos considerar un posible caso para el tamaño de chunksize (cs) que garantice la máxima flexibilidad de programación (también, la cantidad de Highlanders que puede haber. ¿Coincidencia?):

___________________________________ ~ UNO ~ ___________________________________

Si, por ejemplo, tenemos cuatro procesos de trabajo y 37 tareas, habrá trabajadores chunksize=1 incluso con chunksize=1 , solo porque n_workers=4 no es un divisor de 37. El resto de dividir n_workers=4 es 1. Este único El resto de tareas tendrán que ser procesadas por un solo trabajador, mientras que las tres restantes están inactivas.

Del mismo modo, todavía habrá un trabajador inactivo con 39 tareas, cómo se puede ver en la imagen a continuación.

Cuando compara el Horario Paralelo superior para chunksize=1 con la versión de abajo para chunksize=3 , notará que el Horario Paralelo superior es más pequeño, la línea de tiempo en el eje x más corta. Ahora debería ser obvio, cómo los tamaños más grandes inesperadamente también pueden llevar a un aumento en los tiempos de cómputo en general, incluso para escenarios densos .

Pero, ¿por qué no solo usar la longitud del eje x para cálculos de eficiencia?

Porque la sobrecarga no está contenida en este modelo. Será diferente para ambos tamaños, por lo tanto, el eje x no es realmente directamente comparable. La sobrecarga aún puede llevar a un tiempo de cómputo total más largo, como se muestra en el caso 2 de la figura a continuación.

6.3.2 Eficiencia de distribución relativa (RDE)

El valor de ADE no contiene la información si es posible una mejor distribución de las tareas con el tamaño de chunksize establecido en 1. Mejor aún, esto significa un menor Idling Share .

Para obtener un valor de DE ajustado para la máxima DE posible, tenemos que dividir el ADE considerado a través del ADE que obtenemos para chunksize=1 .

Eficiencia de distribución relativa (RDE) = ADE_cs_x / ADE_cs_1

Aquí está cómo se ve en el código:

# mp_utils.py def calc_rde(n_workers, len_iterable, n_chunks, chunksize, last_chunk): """Calculate Relative Distribution Efficiency (RDE).""" ade_cs1 = calc_ade( n_workers, len_iterable, n_chunks=len_iterable, chunksize=1, last_chunk=1 ) ade = calc_ade(n_workers, len_iterable, n_chunks, chunksize, last_chunk) rde = ade / ade_cs1 return rde

RDE , como se define aquí, en esencia es una historia sobre la cola de un Horario Paralelo . RDE está influenciado por el tamaño máximo efectivo contenido en la cola. (Esta cola puede ser del chunksize eje x en chunksize o last_chunk .) Esto tiene la consecuencia, que RDE naturalmente converge al 100% (par) para todo tipo de "looks de cola" como se muestra en la siguiente figura.

Un RDE bajo ...

Es un fuerte indicio del potencial de optimización.

naturalmente, se vuelve menos probable para los iterables más largos, porque la parte relativa de la cola del Programa paralelo general se reduce.

Encuentra la Parte II de esta respuesta aquí abajo .

Creo que parte de lo que te falta es que tu estimación ingenua supone que cada unidad de trabajo lleva la misma cantidad de tiempo, en cuyo caso tu estrategia sería la mejor. Pero si algunos trabajos terminan antes que otros, entonces algunos núcleos pueden quedar inactivos esperando que los trabajos lentos terminen.

Por lo tanto, al dividir los trozos en 4 veces más piezas, luego, si un trozo se terminó antes, el núcleo puede comenzar el siguiente (mientras que los otros núcleos siguen trabajando en su parte más lenta).

No sé por qué eligieron el factor 4 exactamente, pero sería una compensación entre minimizar la sobrecarga del código del mapa (que quiere los fragmentos más grandes posibles) y equilibrar los segmentos que toman diferentes cantidades de veces (lo que quiere el fragmento más pequeño posible). ).