Ir al contenido
wannacode

Grupo de hilos y tareas

En resumen: Un grupo de hilos son unos pocos trabajadores delante de una cola de tareas en lugar de un hilo nuevo por tarea; en Python es ThreadPoolExecutor, y ProcessPoolExecutor para calcular.

Problema

Un hilo por cada imagen

La tienda empieza unas rebajas. Los vendedores suben imágenes de sus productos, y cada imagen necesita una miniatura para la lista de productos. Hacer una miniatura es trabajo que calcula: recorre cada píxel (mira trabajo que espera y trabajo que calcula).

La primera versión inicia un hilo por cada imagen subida:

threads = []
for image in uploads:
    thread = threading.Thread(target=save_thumbnail, args=(image,))
    thread.start()
    threads.append(thread)

Con 20 imágenes todo funciona en tu computadora: «Listas: 20 miniaturas». En la primera hora de las rebajas llegan 10 000 a la vez.

  • Cada hilo necesita su propia pila y un lugar en el sistema operativo. Miles de hilos se comen la memoria, y cerca del límite del sistema start() falla con RuntimeError: can't start new thread.
  • Aunque haya memoria, por el GIL solo un hilo calcula a la vez. Los otros 9999 esperan y se turnan.

El servidor no se vuelve más lento; se cae. Y un servidor caído tampoco atiende a los compradores.

La idea

Unos pocos trabajadores y una cola

Un grupo de hilos son unos pocos trabajadores, creados una vez, y delante de ellos una cola de tareas. Una tarea es «reduce esta imagen». Un trabajador libre toma la siguiente tarea de la cola, la termina y vuelve por una nueva.

  • El número de trabajadores es constante. 10 000 imágenes no hacen 10 000 hilos, sino 10 000 tareas en la cola. Una tarea es un objeto pequeño; un hilo es caro.
  • Los trabajadores se reutilizan. No se crea ni se destruye un hilo por cada imagen.
  • La cola absorbe el pico. Cuando las subidas llegan más rápido de lo que los trabajadores pueden atender, las tareas esperan en lugar de tumbar el servidor.

Esto es productor y consumidor: el código que acepta las subidas pone tareas en la cola, y los trabajadores son los consumidores.

Ya conociste el grupo en la lección sobre hilos: ThreadPoolExecutor, submit y Future.result(). Aquí veremos qué pasa por dentro, cómo ordenar los resultados, qué pasa con los errores y cuántos trabajadores necesitas.

Míralo

Tres trabajadores

Hacer una miniatura aquí es una pausa, 2 ms por cada KB, y el tamaño sale del nombre de la imagen. Así la salida es la misma en cada ejecución:

# Crear una miniatura: aquí es una pausa, y el tamaño sale del nombre.
def thumbnail(image):
    if image.startswith("dañada"):
        raise ValueError("archivo dañado")
    kb = 20 + 3 * len(image)
    time.sleep(2 * kb / 1000)
    return kb

Ocho imágenes, un grupo de tres trabajadores:

with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = list(pool.map(thumbnail, images))
total = 0
for image, kb in zip(images, sizes):
    print(f"{image}: {kb} KB")
    total += kb
print(f"Total: {total} KB")
Paso a paso, en palabras
  1. 8 imágenes esperan. El grupo son 3 hilos: ThreadPoolExecutor(max_workers=3).
  2. map convierte cada imagen en una tarea de la cola. Los tres hilos toman las tres primeras.
  3. Hilo 3 termina primero: Top es la tercera imagen, y su resultado va al tercer lugar. El trabajador no se detiene: toma la siguiente tarea.
  4. La cola está vacía. Hilo 1 terminó y no tiene nada que tomar, así que espera. Los demás terminan sus últimas imágenes.
  5. map devuelve los tamaños en el orden de las imágenes, no en el orden en que terminaron. Total 379 KB; el final de with cierra el grupo.
images = ["Bolso.jpg", "Zapato.jpg", "Top.jpg", "Falda.jpg", "Pantalón.jpg", "Camisa.jpg", "Botín.jpg", "Boa.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = list(pool.map(thumbnail, images))
total = 0
for image, kb in zip(images, sizes):
    print(f"{image}: {kb} KB")
    total += kb
print(f"Total: {total} KB")
Salida
Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
Top.jpg: 41 KB
Falda.jpg: 47 KB
Pantalón.jpg: 56 KB
Camisa.jpg: 50 KB
Botín.jpg: 47 KB
Boa.jpg: 41 KB
Total: 379 KB
  • ThreadPoolExecutor(max_workers=3) crea tres hilos y delante de ellos una cola.
  • map crea una tarea por imagen y devuelve los resultados en el orden de las imágenes, sea cual sea la que terminó primero. Si no esperas a todas, las tomas una por una de map: espera exactamente la siguiente.
  • with cierra el grupo al final: no acepta tareas nuevas, espera a las iniciadas y detiene los hilos.

Los hilos ayudan aquí porque hacer una miniatura es una pausa. Una miniatura real calcula, y por el GIL los hilos no calculan a la vez. Para eso hay procesos, un poco más abajo.

Más precisamente, para curiosos

Sin max_workers el grupo tiene min(32, os.cpu_count() + 4) hilos. Su cola no tiene fin: si las subidas llegan siempre más rápido, crece hasta agotar la memoria.

Cómo es en los otros lenguajes: en Java el grupo es un ExecutorService (allí los hilos sí calculan en paralelo de verdad), en C# hay un grupo para todo el programa, y Node no tiene un grupo de Workers ya hecho.

Pruébalo

Encuentra el error

Un compañero reescribió las miniaturas con un grupo, y las rebajas no fueron más rápidas. Seis imágenes tardan lo mismo que sin grupo:

start = time.monotonic()
sizes = []
with ThreadPoolExecutor(max_workers=3) as pool:
    for image in images:
        size = pool.submit(thumbnail, image)
        sizes.append(size.result())
print(f"Tardó más de 0,5 s: {time.monotonic() - start > 0.5}")
Salida
Tardó más de 0,5 s: True

Elige la línea con el error.

La idea

Resultados en el orden de entrada

Las tareas terminan en un orden que no conoces: depende de las imágenes, de la carga, de qué trabajador estaba libre. Así que nunca imprimas «en el orden en que terminaron» si quieres la misma salida cada vez.

Dos formas dan el mismo resultado en cada ejecución:

  • En el orden de entrada. Guardas las promesas en una lista en el orden de las tareas y las esperas en ese orden. Así funciona el código de arriba.
  • Una suma o una cuenta. Si solo necesitas el total, el orden no importa: 47 + 50 es lo mismo que 50 + 47.

A veces quieres los resultados en el orden en que terminaron: para una barra de «3 de 8 listas» o para mostrar las primeras imágenes listas. as_completed los da así. Entonces recoges solo cosas que no dependen del orden:

count = 0
total = 0
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(thumbnail, image) for image in images]
    for future in as_completed(futures):
        total += future.result()
        count += 1
print(f"Listas: {count} de {len(images)}, en total: {total} KB")
Salida
Listas: 8 de 8, en total: 379 KB
as_completed da cada Future en cuanto está listo. result() sobre él no espera.

La idea

Un error dentro de una tarea

Una de las imágenes está dañada, y thumbnail lanza una excepción. No sale en el hilo que inició la tarea: ese hilo siguió su camino hace rato. El grupo la captura y la guarda en la promesa del resultado. Sale cuando esperas ese resultado:

images = ["Bolso.jpg", "Zapato.jpg", "dañada.jpg", "Top.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(thumbnail, image) for image in images]
for image, future in zip(images, futures):
    try:
        print(f"{image}: {future.result()} KB")
    except ValueError as error:
        print(f"{image}: error: {error}")
Salida
Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
dañada.jpg: error: archivo dañado
Top.jpg: 41 KB
result() vuelve a lanzar la misma excepción (ValueError). Las otras imágenes no sufren. Con pool.map el error sale cuando el bucle llega a la imagen dañada, y los resultados posteriores se pierden. Así que si esperas errores, usa submit y result() para cada una.

Pruébalo

Adónde fue el error

La segunda imagen está dañada. ¿Qué imprime el programa?

images = ["Bolso.jpg", "dañada.jpg", "Top.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    for image in images:
        pool.submit(thumbnail, image)
print(f"Listo: {len(images)} imágenes")
Respuesta

La idea

Cuántos trabajadores

Depende de lo que hagan:

  • Trabajo que calcula (miniaturas, PDF, compresión): tantos como núcleos. Un noveno trabajador en 8 núcleos no tiene dónde ejecutarse; solo se turna con los demás.
  • Trabajo que espera (correo, pago, base de datos): más. Mientras un trabajador espera a la red, el núcleo queda libre para otro. Si una tarea espera W ms y calcula C ms, caben unos 1 + W / C trabajadores en un núcleo.
def pool_size(cores, wait_ms, compute_ms):
    return cores * (1 + wait_ms // compute_ms)




print(f"Miniaturas (solo cálculo): {pool_size(8, 0, 50)}")
print(f"Correos (90 ms de espera, 10 ms de cálculo): {pool_size(8, 90, 10)}")
Salida
Miniaturas (solo cálculo): 8
Correos (90 ms de espera, 10 ms de cálculo): 80

El número de núcleos viene de os.cpu_count(). La fórmula es un punto de partida, no una ley: mide. Si los correos pasan por un servidor con un límite de 20 conexiones, 80 trabajadores solo esperarán en una cola delante de él.

Python tiene una trampa: por el GIL, los hilos no calculan a la vez. Para el trabajo que calcula, el grupo se hace de procesos: ProcessPoolExecutor, cada proceso con su propio Python y su propio GIL. Las tareas y los resultados viajan entre los procesos como copias:

from concurrent.futures import ProcessPoolExecutor




def checksum(invoice):
    return sum((invoice * 31 + i) % 256 for i in range(200_000))




if __name__ == "__main__":
    invoices = [1001, 1002, 1003, 1004]
    with ProcessPoolExecutor(max_workers=4) as pool:
        sums = list(pool.map(checksum, invoices))
    for invoice, total in zip(invoices, sums):
        print(f"Factura {invoice}: {total}")
Salida
Factura 1001: 25497376
Factura 1002: 25499360
Factura 1003: 25501344
Factura 1004: 25503328

if __name__ == "__main__": evita que el código principal se ejecute otra vez en cada proceso: en macOS y Windows un proceso empieza cargando el archivo desde el principio.

Pruébalo

Completa

¿Cuántos trabajadores para una tarea que espera waitMs y calcula computeMs?

def pool_size(cores, wait_ms, compute_ms):
    return cores * (1 + )
Más precisamente, para curiosos

Para el trabajo que espera, Python también tiene asyncio: un hilo espera miles de peticiones (la lección async no es un hilo). El grupo de hilos queda para las bibliotecas que esperan sin poder usar await.

Ejercicio

Las facturas de las rebajas

Resuelve la tarea en tu editor.

Después de las rebajas cada pedido necesita una factura. invoice(amount) crea el PDF (aquí una pausa de 50 ms) y devuelve el importe con 20 % de IVA. Un pedido con un importe de 0 o menos no es válido, y la función lanza una excepción con el texto «importe no válido».

Los pedidos son del 1001 al 1008 con los importes 50 €, 125 €, 80 €, 0 €, 240 €, 35 €, 90 € y 15 €. Haz las facturas en un grupo de 4 trabajadores. Imprímelas en el orden de los pedidos, y la no válida con el texto del error. Al final: la suma de las facturas buenas y cuántas fallaron.

Salida
Pedido 1001: 60 €
Pedido 1002: 150 €
Pedido 1003: 96 €
Pedido 1004: error: importe no válido
Pedido 1005: 288 €
Pedido 1006: 42 €
Pedido 1007: 108 €
Pedido 1008: 18 €
Total: 762 €, con error: 1

Más tareas

  • Una cola limitada. Haz que el grupo mantenga como máximo 10 tareas esperando en la cola. ¿Qué pasa con la undécima? (threading.BoundedSemaphore alrededor de submit)
  • Progreso. Imprime «3 de 8 listas» después de cada factura terminada, y al final otra vez la lista ordenada.
  • Un timeout. Una factura que tarda más de 1 segundo cuenta como fallida, sin detener a las demás.

Test

Ponte a prueba

1. ¿Por qué un grupo de pocos hilos y no un hilo por cada imagen subida?
2. Un grupo de 3 trabajadores reduce 8 imágenes. La tercera termina primera. ¿Dónde está su tamaño cuando las lees en el orden de la lista con get, result, Promise.all o sizes[i]?
3. Una tarea del grupo lanza una excepción. ¿Cuándo la ves?
4. El servidor tiene 8 núcleos. Cada correo espera 90 ms a la red y calcula 10 ms. ¿Cuántos trabajadores mantienen ocupados los núcleos, más o menos?
5. ¿Cuántos trabajadores para hacer miniaturas, si solo calcula y el servidor tiene 8 núcleos?

Creado por Ivo Gergov como parte de la campaña de Evo Tech por una educación accesible en programación ·