Cocineros de agencia: concurrent.futures

Séptima noche, diez y media. Suena el teléfono, y la chef cuelga y dice: «Cena de empresa. Treinta raciones».

Minho se reajusta el delantal y pregunta: «¿Creo 30 hilos?». La chef niega con la cabeza. «Si contratas gente, todo corre de tu cuenta: alojarla, pagarle, despedirla. Hoy llamamos a una agencia. Ellos se ocupan de las personas; nosotros solo les pasamos el trabajo».

7-1. Encargar el trabajo a una agencia: ThreadPoolExecutor (10 min)

concurrent.futures.ThreadPoolExecutor es una agencia que tiene preparados unos cuantos hilos y, cuando llega trabajo, se lo da a uno que esté libre. Se encarga de todo lo que en la lección 2 hacías a mano: crear el Thread, start, join y preparar la lista donde guardar los resultados.

import time
from concurrent.futures import ThreadPoolExecutor

def cook(n):
    time.sleep(0.2)
    return f"#{n} ready"

with ThreadPoolExecutor(max_workers=3) as agency:   # tres cocineros de agencia
    fut = agency.submit(cook, 1)        # al encargar el trabajo, te dan un «resguardo» (Future)
    print(fut.done())                   # todavía está cocinando
    print(fut.result())                 # espera a que termine y recoge el resultado
    print(fut.done())
# False
# #1 ready
# True

• submit(función, argumentos...) encarga el trabajo y devuelve al instante un Future. No es la comida: es «el resguardo con el que recogerás la comida». Si haces print(fut), verás su estado, algo como <Future at 0x... state=running>.
• fut.result() espera a que el plato esté listo y devuelve el valor de retorno de la función. Aquí se resuelve el problema de la lección 2, donde, como «los hilos no devuelven valores», guardábamos los resultados en una lista.
• Al salir del bloque with se llama a shutdown(wait=True), que espera a que termine todo el trabajo encargado y recoge los hilos.

Ahora encarguemos 30. Solo hay cinco cocineros, así que a cada uno le tocan seis pedidos, uno detrás de otro.

import threading

def cook(n):
    time.sleep(0.1)
    return threading.current_thread().name

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5, thread_name_prefix="agency") as agency:
    futures = [agency.submit(cook, n) for n in range(1, 31)]   # 30 pedidos
# al salir del bloque with, todas las tareas han terminado
names = {f.result() for f in futures}
print(len(futures), sorted(names), f"{time.perf_counter() - start:.2f}s")
# 30 ['agency_0', 'agency_1', 'agency_2', 'agency_3', 'agency_4'] 0.60s

Hay 30 pedidos, pero solo se crearon cinco hilos. 30 pedidos de 0,1 segundos repartidos entre cinco: 0,6 segundos. El coste de crear y destruir hilos se paga una sola vez, y los mismos cocineros van recibiendo un pedido tras otro. Esto se llama grupo de hilos (thread pool).

7-2. Orden de llegada contra orden de entrada: as_completed y map (10 min)

Hay dos maneras de recoger los resultados. Esta vez haremos que el pedido #1 sea el que más tarde.

from concurrent.futures import as_completed

def cook(n):
    time.sleep(0.1 * (5 - n))     # #1 es el que más tarda (0.4 s … 0.1 s)
    return f"#{n}"

with ThreadPoolExecutor(max_workers=4) as agency:
    start = time.perf_counter()
    futures = [agency.submit(cook, n) for n in range(1, 5)]
    for f in as_completed(futures):                 # en el orden en que terminan
        print(f.result(), f"{time.perf_counter() - start:.1f}s", end="  ")
    print()
    start = time.perf_counter()
    for dish in agency.map(cook, range(1, 5)):      # en el orden en que se encargaron
        print(dish, f"{time.perf_counter() - start:.1f}s", end="  ")
    print()
# #4 0.1s  #3 0.2s  #2 0.3s  #1 0.4s
# #1 0.4s  #2 0.4s  #3 0.4s  #4 0.4s

Aunque #4 termine a los 0,1 segundos, map no lo entrega hasta los 0,4 segundos, cuando termina #1. Tiene que respetar el orden de entrada. El tiempo total es el mismo, pero cambia cuándo sale el primer plato.

as_completed solo devuelve el Future, así que, para saber de qué pedido es cada resultado, lo habitual es emparejarlos en un dict.

with ThreadPoolExecutor(max_workers=4) as agency:
    table_of = {agency.submit(cook, n): f"T{n}" for n in range(1, 5)}   # Future → mesa
    for fut in as_completed(table_of):
        print(table_of[fut], fut.result())
# T4 #4
# T3 #3
# T2 #2
# T1 #1

El modo «Cocineros de agencia» de la línea de tiempo de abajo también funciona con un ThreadPoolExecutor de verdad. En este escenario, cuanto más alto es el número del pedido, antes termina.

• Con los dos modos tarda alrededor de 1,0 segundos, y las barras tienen casi la misma forma. Los cocineros trabajan igual; solo cambia el orden en que se recogen los resultados.
• Con as_completed, el orden de llegada sale revuelto, por ejemplo #3 → #2 → #1 → #6 → #5 → #4 → …. De los pedidos que empiezan a la vez los tres cocineros, el más corto termina primero.
• Con map siempre es #1 → #2 → … → #10.
• Si subes los cocineros al máximo, 8, termina hacia los 0,4 segundos, y el orden de llegada de as_completed queda casi al revés: #8 → #7 → … → #1. Los ocho pedidos empiezan a la vez y terminan primero los de número alto, que son más cortos.

7-3. El plato quemado aparece al presentar el resguardo (7 min)

Si salta una excepción dentro del trabajo encargado, la excepción se guarda en silencio dentro del Future. En el momento en que llamas a result(), vuelve a saltar ahí mismo.

def cook(n):
    if n == 3:
        raise ValueError(f"#{n} burnt")    # el pedido 3 se ha quemado
    return f"#{n}"

with ThreadPoolExecutor(max_workers=2) as agency:
    futures = [agency.submit(cook, n) for n in range(1, 5)]
# hasta aquí, todo tranquilo, como si no hubiera pasado nada

for f in futures:
    try:
        print(f.result())
    except ValueError as e:                # salta al llamar a result()
        print("error:", e)
# #1
# #2
# error: #3 burnt
# #4

print(repr(futures[2].exception()), futures[0].exception())
# ValueError('#3 burnt') None

• Si nunca llamas a result(), la excepción desaparece sin mostrarse en ningún sitio. Es como si el plato quemado no llegara al cliente, pero nadie se enterara de que se quemó.
• exception() te entrega la excepción sin lanzarla. Si todo fue bien, devuelve None.
• map lanza la excepción en el momento de sacar ese resultado, y los resultados posteriores ya no se pueden recoger. Si necesitas el resto aunque uno falle, usa submit y as_completed.

Si quieres hacer algo justo cuando termine, cuelga una campanilla con add_done_callback. Le pasas una función que recibe como argumento un Future ya terminado.

def cook(n):
    time.sleep(0.1)
    return f"#{n}"

def ring_bell(fut):                      # función que se llama al terminar (callback). Recibe un Future
    print("bell:", fut.result(), threading.current_thread().name)

with ThreadPoolExecutor(max_workers=2, thread_name_prefix="agency") as agency:
    for n in (1, 2):
        agency.submit(cook, n).add_done_callback(ring_bell)
# p. ej.:
# bell: #1 agency_0
# bell: #2 agency_1

El callback se ejecuta en el hilo del cocinero de agencia que terminó el trabajo (si lo cuelgas en un Future ya terminado, se ejecuta al momento en el hilo de quien lo colgó). Así que, si dentro del callback modificas datos compartidos, se aplican tal cual las reglas de cerrojos de la lección 3. Las excepciones dentro de un callback solo quedan en el log; por lo demás, se ignoran.

7-4. Cancelar, y esperar solo a uno (8 min)

cancel() solo puede cancelar trabajo que todavía no ha empezado. Un plato que ya está al fuego no tiene vuelta atrás.

with ThreadPoolExecutor(max_workers=1) as agency:     # un solo cocinero
    futures = [agency.submit(time.sleep, 0.2) for _ in range(3)]
    time.sleep(0.05)                                  # tiempo para que el primer pedido llegue al fuego
    print([f.cancel() for f in futures])              # lo que ya empezó no se puede retirar
    print([f.cancelled() for f in futures], [f.done() for f in futures])
# [False, True, True]
# [False, True, True] [False, True, True]

• cancel() devuelve True si consigue cancelar, y False si el trabajo ya está en marcha o ha terminado.
• done() es True si el trabajo terminó o se canceló. El primer pedido todavía se está cocinando, así que da False.

Para rescindir de golpe el contrato con la agencia, usa shutdown(cancel_futures=True). Cancela todo el trabajo que estaba en cola y solo espera hasta el final al que ya había empezado.

agency = ThreadPoolExecutor(max_workers=2)
futures = [agency.submit(time.sleep, 0.2) for _ in range(6)]
time.sleep(0.05)
agency.shutdown(cancel_futures=True)     # lo que está en cola se cancela; lo que está al fuego se termina
print(sum(f.cancelled() for f in futures), sum(not f.cancelled() for f in futures))
# 4 2

Cuando quieres actuar en cuanto termine al menos una de varias tareas, usa wait.

from concurrent.futures import FIRST_COMPLETED, wait

def cook(sec):
    time.sleep(sec)
    return sec

with ThreadPoolExecutor(max_workers=3) as agency:
    futures = [agency.submit(cook, s) for s in (0.3, 0.1, 0.2)]
    done, pending = wait(futures, return_when=FIRST_COMPLETED)   # en cuanto termine una
    print([f.result() for f in done], len(pending))
    done, pending = wait(futures, timeout=0.15)                  # por defecto, ALL_COMPLETED
    print(len(done), len(pending))
# [0.1] 2
# 2 1

wait devuelve dos conjuntos: lo terminado y lo pendiente. return_when admite FIRST_COMPLETED, FIRST_EXCEPTION y ALL_COMPLETED (el valor por defecto), y si pasa el timeout, vuelve aunque no haya terminado todo. Como no lanza ninguna excepción, tienes que comprobar tú si pending está vacío.

7-5. El mismo contrato, otra agencia (5 min)

Lo bueno de concurrent.futures es que el contrato es el mismo. submit, map, Future y as_completed no cambian; basta con cambiar el nombre de la agencia.

from concurrent.futures import InterpreterPoolExecutor, ProcessPoolExecutor, ThreadPoolExecutor

def knead(n):
    return sum(range(n))

if __name__ == "__main__":               # imprescindible para el grupo de procesos (lección 8)
    for Agency in (ThreadPoolExecutor, ProcessPoolExecutor, InterpreterPoolExecutor):
        with Agency(max_workers=2) as agency:
            print(Agency.__name__, list(agency.map(knead, [10, 100, 1000])))
# ThreadPoolExecutor [45, 4950, 499500]
# ProcessPoolExecutor [45, 4950, 499500]
# InterpreterPoolExecutor [45, 4950, 499500]

Guarda este ejemplo en un archivo .py y ejecútalo; no uses el intérprete interactivo. Por qué el grupo de procesos exige if __name__ == "__main__": lo veremos la próxima noche.

Si omites max_workers, el grupo de hilos llama a min(32, os.process_cpu_count() + 4) cocineros (hasta 3.12 era os.cpu_count()). En este PC de 12 núcleos, 16. Se suman 4 porque el trabajo de espera necesita más hilos que núcleos, y se corta en 32 para que no se disparen. En un servicio real, conviene fijarlo a mano pensando en cuántas conexiones aguanta el servidor del otro lado.

Cierre (10 min)

El objetivo de esta noche era encargar los 30 pedidos a seis cocineros de agencia, sacarlos a medida que se terminaban en menos de 1 segundo, y no dejar escapar ningún plato quemado. Cada pedido tarda entre 0,1 y 0,2 segundos, y el pedido 13 se quema.

def cook(n):
    time.sleep(0.1 + 0.05 * (n % 3))     # entre 0.1 y 0.2 s
    if n == 13:
        raise ValueError(f"#{n} burnt")
    return f"#{n}"

served, burnt = [], []
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=6, thread_name_prefix="agency") as agency:
    futures = {agency.submit(cook, n): n for n in range(1, 31)}
    for fut in as_completed(futures):     # se sacan a medida que terminan
        try:
            served.append(fut.result())
        except ValueError:
            burnt.append(futures[fut])
print(len(served), burnt, f"{time.perf_counter() - start:.2f}s")
# 29 [13] 0.80s  (el tiempo es aproximado)

29 platos servidos a medida que salían, y el 13, quemado, apuntado en la lista para repetirlo. 0,8 segundos. Sin código para crear hilos y sin preparar de antemano una lista para los resultados.

La clienta de la ventana vuelve a estar sola esta noche, con un bol de fideos delante. Mientras los platos del grupo salen en tropel, mira una vez su reloj de pulsera y apunta algo breve en una libretita.

La chef despide a los cocineros de agencia y dice: «Mañana tenemos una montaña de masa. Si llamo a cuatro cocineros de agencia y les doy la masa, ¿irá cuatro veces más rápido?».

Ejercicios

1. En el código del cierre, cambia as_completed por agency.map(cook, range(1, 31)). ¿Qué pasa con el 13 y cuántos platos llegas a recoger?
2. En la línea de tiempo, baja los cocineros a 1 (tarda unos 2,8 segundos) y compara as_completed con map. ¿Por qué el orden de llegada pasa a ser el mismo?
3. Escribe con wait(..., return_when=FIRST_COMPLETED) y shutdown(cancel_futures=True) un código que, de 30 pedidos, recoja solo el primer plato que termine y cancele el resto. ¿Cuántos platos habían empezado ya?