Contratar más cocineros: lo básico de threading

Segunda noche, nueve y media. Tres cocineros de línea se atan el delantal. Minho ya tiene el cuchillo en la mano, Sua ordena las comandas una por una y el señor Tak limpia el fogón sin prisa.

A las diez entran los pedidos y cada uno pone su olla al fuego. Pero media hora después, cuando la chef pregunta «¿Quién hizo el udon de la mesa 3?», nadie sabe qué contestar.

La chef da un golpecito al temporizador que lleva al cuello. «Contratar cocineros, hecho. Hoy aprendemos a ponerles una etiqueta con su nombre, a darles trabajo y a esperar a que terminen.»

2-1. Incorporar a un cocinero: Thread, start, join (10 min)

En Python, un cocinero es un hilo (thread). Los hilos comparten la misma cocina (la memoria) y cada uno avanza con su propio trabajo. Se crean con threading.Thread.

import threading
import time

def cook(name, dish):
    print(f"{name}: start {dish}")
    time.sleep(0.3)          # esperar a que hierva la sopa
    print(f"{name}: done {dish}")

minho = threading.Thread(target=cook, args=("minho", "ramen"))
minho.start()                # Minho se pone a trabajar
print("chef: next order")    # la chef no espera: pasa a lo siguiente
minho.join()                 # esperar a que Minho termine
print("chef: served")
# minho: start ramen
# chef: next order
# minho: done ramen
# chef: served

• En target= va el trabajo del hilo (una función) y en args=, los argumentos de esa función, en forma de tupla. No llamas a la función, le pasas la función misma: por eso es cook y no cook().
• Crear el Thread(...) no hace nada todavía. Es como haber contratado al cocinero y nada más. Hasta que no llamas a start(), cook no empieza a ejecutarse en el hilo nuevo.
• start() vuelve enseguida. Así que, mientras Minho cocina el ramen, la chef (el hilo principal, el que arrancó el programa) pasa a la línea siguiente.
• join() espera a que ese hilo termine. Es «esperar en el pase hasta que Minho entregue el plato».

Si le das un plato a cada uno de los tres, queda así:

def cook(name, dish):
    time.sleep(0.3)
    print(f"{name}: done {dish}")

crew = [("minho", "ramen"), ("sua", "udon"), ("tak", "soba")]
threads = [threading.Thread(target=cook, args=c) for c in crew]

start = time.perf_counter()
for th in threads:
    th.start()        # primero arrancan los tres
for th in threads:
    th.join()         # y después se espera a todos
print(f"{time.perf_counter() - start:.2f}s")
# por ejemplo:
# sua: done udon
# minho: done ramen
# tak: done soba
# 0.30s

El orden en que terminan los tres cambia en cada ejecución. Como los tres acaban casi a la vez, a los 0,3 segundos, depende de a quién despierte primero el sistema operativo. En código concurrente, el «orden de la salida» no es una promesa.

No pongas start() y join() en el mismo bucle. Si alternas th.start(); th.join(), cada persona tiene que terminar para que salga la siguiente, y acabas tardando 0,9 segundos.

También puedes ponerle un límite a la espera. join(timeout=segundos) espera solo ese tiempo y vuelve; para saber si el hilo sigue trabajando, usa is_alive().

slow = threading.Thread(target=time.sleep, args=(1.0,))
slow.start()
slow.join(timeout=0.2)
print(slow.is_alive())   # True: todavía está hirviendo
slow.join()
print(slow.is_alive())   # False

• Con 1 cocinero tarda unos 1,8 segundos, igual que «Solo, uno por uno». Con un solo hilo no hay nada que ganar.
• Con 2 cocineros, unos 0,9 segundos; con 3, unos 0,6; con 6, unos 0,3.
• Con 4 y con 5 tarda lo mismo: unos 0,6 segundos. Si repartes 6 pedidos entre cuatro, a alguien le tocan dos. El tiempo lo marca quien más pedidos tiene.
• Cada fila es un hilo. Los cocineros de esta línea de tiempo también crean su Thread y llaman a start() y join(), igual que el código de arriba.

2-2. Llamar a run() es una simple llamada a función (5 min)

Thread también tiene un método run(). Por el nombre parece «ejecutar», pero si lo llamas directamente caes en una trampa.

def cook(name, dish):
    time.sleep(0.3)
    print(f"{threading.current_thread().name}: done {dish}")

threads = [threading.Thread(target=cook, args=c) for c in crew]

start = time.perf_counter()
for th in threads:
    th.run()          # run() en lugar de start()
print(f"{time.perf_counter() - start:.2f}s")
# MainThread: done ramen
# MainThread: done udon
# MainThread: done soba
# 0.90s

• run() es un método corriente que llama a target. No crea ningún hilo: lo ejecuta todo en orden en el hilo actual. Por eso las tres veces lo hizo MainThread, y tardó 0,3 × 3 = 0,9 segundos.
• Lo que lanza un hilo nuevo es start(). start() crea el hilo y, dentro de él, llama a run().

Un hilo solo puede arrancar una vez. Si llamas dos veces a start() sobre el mismo hilo, da error. Para repetir el trabajo, crea un Thread nuevo.

th = threading.Thread(target=time.sleep, args=(0.1,))
th.start()
th.start()
# RuntimeError: threads can only be started once

2-3. Ponerles nombre: name, current_thread, enumerate (7 min)

El problema de esta noche era «no sabemos quién lo hizo». A los hilos se les puede poner una etiqueta con su nombre.

def cook(dish):
    me = threading.current_thread()       # el hilo que ejecuta este código
    print(me.name, dish, threading.get_native_id())
    time.sleep(0.2)

a = threading.Thread(target=cook, args=("ramen",), name="minho")
b = threading.Thread(target=cook, args=("udon",))      # ¿y si no le damos nombre?
a.start(); b.start()
print([t.name for t in threading.enumerate()])
print(threading.active_count())
a.join(); b.join()
print(threading.current_thread() is threading.main_thread())
print([t.name for t in threading.enumerate()])
# por ejemplo:
# minho ramen 25680
# Thread-1 (cook) udon 38072
# ['MainThread', 'minho', 'Thread-1 (cook)']
# 3
# True
# ['MainThread']

• b, que no tiene nombre, se llama Thread-1 y no Thread-2 porque la numeración solo cuenta los hilos sin nombre.
• El número de get_native_id() cambia en cada ejecución. El número que usa Python internamente es th.ident (o threading.get_ident()), y ambos se fijan solo cuando el hilo ya arrancó.
• Un hilo que terminó su trabajo desaparece de enumerate(). Quien ya se fue a casa no está en la lista.

2-4. Los hilos no devuelven el plato (6 min)

Una función devuelve su resultado con return, pero el hilo no lo recoge. El valor de retorno de la función target simplemente se tira, y join() tampoco devuelve nada.

def make(dish):
    time.sleep(0.1)
    return f"{dish} ok"

th = threading.Thread(target=make, args=("ramen",))
th.start()
print(th.join())     # join no devuelve nada
# None

Por eso hay que preparar de antemano un recipiente para los resultados y que el hilo los deje ahí. Lo más fácil es asignarle a cada cocinero su propia casilla.

dishes = ["ramen", "udon", "soba"]
results = [None] * len(dishes)          # preparar los huecos antes

def cook(i, dish):
    results[i] = make(dish)             # cada uno escribe solo en su casilla

threads = [threading.Thread(target=cook, args=(i, d)) for i, d in enumerate(dishes)]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(results)
# ['ramen ok', 'udon ok', 'soba ok']

Si prefieres buscar por nombre, usa un dict.

done = {}

def cook2(name, dish):
    done[name] = make(dish)

ts = [threading.Thread(target=cook2, args=(n, d)) for n, d in [("minho", "ramen"), ("sua", "udon")]]
for t in ts:
    t.start()
for t in ts:
    t.join()
print(sorted(done.items()))
# [('minho', 'ramen ok'), ('sua', 'udon ok')]

• Si cada uno escribe en una casilla distinta o en una clave distinta, los cocineros no chocan. Cuando empiezan a leer y reescribir el valor de la misma casilla, la cosa cambia, y ese es el incidente de mañana por la noche.
• Lee los resultados siempre después de que termine join(). Si los lees antes, puede que alguna casilla siga en None.
• Si preparar el recipiente cada vez te parece un engorro, el Future que devuelve ThreadPoolExecutor en la lección 7 lo hace por ti. Con future.result() recibes el plato directamente.

2-5. daemon=True: el hilo al que no se espera al apagar las luces (6 min)

Aunque el hilo principal termine, un programa de Python espera a que terminen todos los hilos que no son daemon antes de cerrarse. Los hilos que creas por defecto no son daemon.

def clean():
    for i in range(1, 6):
        print("cleaning", i)
        time.sleep(0.1)

th = threading.Thread(target=clean)      # daemon=False (valor por defecto)
th.start()
time.sleep(0.25)
print("lights off")
# cleaning 1
# cleaning 2
# cleaning 3
# lights off
# cleaning 4
# cleaning 5

Aunque la chef grite «¡Apago las luces!», el restaurante no cierra hasta que quien está limpiando termine. Ahora añadamos solo daemon=True al mismo código.

th = threading.Thread(target=clean, daemon=True)
th.start()
time.sleep(0.25)
print("lights off", th.daemon)
# cleaning 1
# cleaning 2
# cleaning 3
# lights off True

• A los hilos daemon no se les espera cuando el programa termina. Las luces se apagan en mitad de la limpieza, y los pasos 4 y 5 no se hacen nunca.
• Si quieres esperarlo, llama a join() aunque sea daemon. join() espera sea daemon o no.
• Usa daemon solo para «trabajos que pueden cortarse a medias», como un hilo vigilante que imprime el estado cada cierto tiempo. Si le das a un daemon algo que tiene que terminarse, como escribir un archivo o registrar pedidos, te quedará un archivo escrito a medias.
• Si es daemon o no se decide antes de start(). Si no lo indicas, hereda el valor del hilo que lo creó; como el hilo principal no es daemon, normalmente es False.

2-6. Otras herramientas: heredar de Thread, Timer, local (6 min)

Heredar de Thread y cambiar run. Útil cuando quieres que el cocinero lleve consigo su propio estado (el plato que le toca, el resultado). Como vimos en 2-2, start() llama a run() en el hilo nuevo, así que basta con reescribir run().

class Cook(threading.Thread):
    def __init__(self, name, dish):
        super().__init__(name=name)     # primero, siempre el __init__ del padre
        self.dish = dish
        self.result = None

    def run(self):                      # start() llama a esto en el hilo nuevo
        time.sleep(0.1)
        self.result = f"{self.dish} ok"

tak = Cook("tak", "soba")
tak.start()
tak.join()
print(tak.name, tak.result)
# tak soba ok

threading.Timer: una vez, dentro de n segundos. Como el temporizador de la pasta: un hilo que llama a una función una sola vez cuando pasa el tiempo fijado. Si todavía no ha sonado, lo puedes apagar con cancel().

start = time.perf_counter()

def ring(msg):
    print(f"{time.perf_counter() - start:.1f}s", msg)

noodles = threading.Timer(0.5, ring, args=("noodles up",))
noodles.start()                     # suena una vez dentro de 0,5 s
eggs = threading.Timer(1.0, ring, args=("eggs up",))
eggs.start()
eggs.cancel()                       # si lo apagas antes de que suene, no pasa nada
noodles.join()
eggs.join()
print("end")
# 0.5s noodles up
# end

threading.local(): el bolsillo del delantal de cada cocinero. Una variable con el mismo nombre que guarda un valor distinto en cada hilo. El bolsillo depende de quién mete la mano.

pocket = threading.local()

def cook(name, n):
    pocket.name = name
    pocket.count = 0
    for _ in range(n):
        pocket.count += 1
        time.sleep(0.01)
    print(pocket.name, pocket.count)

ts = [threading.Thread(target=cook, args=a) for a in [("minho", 3), ("sua", 5)]]
for t in ts:
    t.start()
for t in ts:
    t.join()
print(hasattr(pocket, "name"))      # el bolsillo de la chef (hilo principal) está vacío
# minho 3
# sua 5
# False

Aunque pocket es una sola variable global, el count de Minho y el de Sua no se mezclan. Es una técnica habitual para que cada hilo tenga su propia conexión a la base de datos.

Cierre (10 min)

El objetivo de esta noche era repartir seis pedidos entre tres personas, terminarlos en menos de 0,6 segundos y dejar constancia de quién terminó qué y en qué segundo. Les ponemos nombre (name=) y hacemos que cada uno apunte una línea en un diario común cada vez que termina.

start = time.perf_counter()
log = []                                # (instante en que terminó, cocinero, pedido)

def cook(dishes):
    me = threading.current_thread().name
    for dish in dishes:
        time.sleep(0.3)
        log.append((round(time.perf_counter() - start, 1), me, dish))

plan = {"minho": ["#1", "#4"], "sua": ["#2", "#5"], "tak": ["#3", "#6"]}
threads = [threading.Thread(target=cook, args=(d,), name=n) for n, d in plan.items()]
for th in threads:
    th.start()
for th in threads:
    th.join()

for row in sorted(log):
    print(*row)
print(f"total {time.perf_counter() - start:.2f}s, {len(log)} orders")
# 0.3 minho #1
# 0.3 sua #2
# 0.3 tak #3
# 0.6 minho #4
# 0.6 sua #5
# 0.6 tak #6
# total 0.60s, 6 orders

• El argumento es una sola lista, por eso args=(d,) lleva coma.
• log.append(...) es una operación de lista que se completa de una vez, así que aunque la llamen varios hilos no se pierde ninguna línea. El diario se lee ordenado, después de join().

La chef repasa el diario y asiente. «Bien, ahora sí sabemos quién hizo qué. Mañana probamos una cosa más. Si cuatro personas suman lo que hizo cada una a un único número en la pizarra, ¿ese número saldrá bien?»

Ejercicios

1. En el ejemplo de los tres cocineros de 2-1, pon start() y join() dentro del mismo bucle. Predice cuántos segundos tardará y luego mídelo.
2. En el código del cierre, haz que solo el señor Tak tarde 0,5 segundos por pedido (pista: cambia el tiempo del sleep según el nombre del cocinero). ¿Quién marca el tiempo total? Compáralo con la línea de tiempo cuando solo una fila se alarga.
3. Con threading.Timer, crea un temporizador de cocina que suene «tres veces, una cada 0,3 segundos». Un Timer suena una sola vez, así que cada vez que suene tendrás que arrancar uno nuevo.