Solo tres fogones: herramientas de sincronización

Cuarta noche, diez menos veinte. Hay una reserva de grupo, así que han venido tres cocineros más solo por hoy. Seis cocineros, tres fogones.

Minho es el primero en plantarse delante de un fogón y encenderlo. «Si lo caliento antes, voy más rápido.» La chef, sin decir nada, cierra la llave del gas. «Prohibido encender antes de abrir. Y con seis personas amontonadas delante de los fogones, aquí no se mueve nadie.»

Sua levanta la mano. «¿Y si colgamos solo tres fichas numeradas delante de los fogones? Solo puede ponerse al fuego quien tenga una.» La chef sonríe por primera vez. «Esa es la primera herramienta de esta noche.»

4-1. Tres fogones: Semaphore (12 min)

El Lock de la lección 3 era una tiza: una persona a la vez. Semaphore(n) son n fichas numeradas. Pueden entrar hasta n personas a la vez, y la persona número n+1 espera a que alguien devuelva su ficha.

import threading
import time

stoves = threading.Semaphore(3)         # tres fogones
start = time.perf_counter()

def cook(name):
    with stoves:                        # espera a que quede un fogón libre y lo toma
        print(f"{time.perf_counter() - start:.1f}s {name} on")
        time.sleep(0.3)

names = [f"cook{i}" for i in range(1, 7)]
threads = [threading.Thread(target=cook, args=(n,)) for n in names]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(f"total {time.perf_counter() - start:.2f}s")
# por ejemplo:
# 0.0s cook1 on
# 0.0s cook2 on
# 0.0s cook3 on
# 0.3s cook4 on
# 0.3s cook5 on
# 0.3s cook6 on
# total 0.60s

• Al entrar en with stoves: se toma una ficha (acquire) y al salir se devuelve (release). Si quedan 0 fichas, quien quiere entrar espera.
• Tres en el segundo 0,0 y tres en el 0,3. Con tres fogones, los seis pedidos se hacen en dos tandas y tardan 0,6 segundos. Quién toma primero un fogón puede cambiar en cada ejecución.
• Semaphore(1) deja pasar a una persona a la vez, como un Lock. La diferencia: si sueltas un Lock que ya está suelto, salta RuntimeError; en cambio, un semáforo acepta más llamadas a release() y simplemente suma fichas en silencio (ver BoundedSemaphore más abajo).
• En la práctica se usa para limitar recursos que no aguantan que se amontone todo de golpe: «como mucho cinco conexiones simultáneas», «como mucho tres descargas a la vez».

Cambia el número de fogones en la línea de tiempo de abajo. Las barras amarillas son el tiempo esperando un fogón; las naranjas, el tiempo cocinando de verdad.

• 3 fogones: tres filas pasan directamente a naranja; las otras tres esperan en amarillo 0,3 segundos y luego cambian a naranja. En total, unos 0,6 segundos.
• 1 fogón: aunque haya seis cocineros, cocinan de uno en uno. El último espera 1,5 segundos, y el total es de unos 1,8 segundos. Igual que una persona sola.
• 6 fogones: casi no hay espera, unos 0,3 segundos. Cuando hay tantos fogones como cocineros, ya no se gana más.
• Si añades cocineros pero no fogones, lo único que crece son las barras amarillas. El cuello de botella no es la gente, sino los fogones.

BoundedSemaphore: atrapar el error de colgar una ficha de más

Semaphore no lleva la cuenta de cuántas fichas le devuelven. Si un error hace que llames a release() una vez de más, las fichas pasan a ser cuatro sin que nadie se entere. Hay tres fogones, pero se ponen cuatro al fuego.

loose = threading.Semaphore(3)
loose.release()                         # se soltó una de más por error
print(sum(loose.acquire(blocking=False) for _ in range(10)))   # ahora hay cuatro fogones
# 4

strict = threading.BoundedSemaphore(3)
strict.release()
# ValueError: Semaphore released too many times

BoundedSemaphore lanza ValueError si le devuelven más fichas de las fijadas al principio. Si usas un semáforo para limitar una cantidad, BoundedSemaphore es casi siempre la mejor opción: el error no pasa desapercibido, revienta en el acto.

4-2. «¡Abrimos!»: Event (9 min)

Un Event es una luz de señal en la pared de la cocina. Empieza apagado, y cuando alguien lo enciende (set()), todos los que esperaban esa señal (wait()) se ponen en marcha a la vez.

opened = threading.Event()              # empieza apagado
start = time.perf_counter()

def cook(name):
    opened.wait()                       # espera a que se encienda la señal
    print(f"{time.perf_counter() - start:.1f}s {name} fire")

threads = [threading.Thread(target=cook, args=(n,)) for n in ["minho", "sua", "tak"]]
for th in threads:
    th.start()
time.sleep(0.5)
print(opened.is_set())
opened.set()                            # «¡Abrimos!»
for th in threads:
    th.join()
print(opened.is_set())
# False
# 0.5s minho fire     (el orden de estas tres líneas cambia en cada ejecución)
# 0.5s sua fire
# 0.5s tak fire
# True

También puedes ponerle un límite de tiempo a wait(). En ese caso, importa lo que devuelve: True si la señal se encendió, False si se acabó el tiempo y sigue apagada.

closing = threading.Event()
print(closing.wait(timeout=0.2))        # esperó 0,2 s y no hubo señal
closing.set()
print(closing.wait(timeout=0.2))        # si ya está encendida, True al instante
closing.clear()                         # se apaga otra vez
print(closing.is_set())
# False
# True
# False

Hay un patrón muy habitual: crear la señal de «fin del servicio» con un Event y hacer que el hilo trabajador lo compruebe en cada vuelta del bucle con if closing.is_set(): break. A diferencia del daemon de la lección 2, que se corta a medias, así el hilo recoge por sí mismo y termina.

La línea de tiempo de abajo es una cocina en la que la chef grita «¡Abrimos!» a los 0,5 segundos.

• Todas las filas esperan en amarillo de 0 a 0,5 segundos y, en el 0,5, pasan a naranja a la vez. En total, unos 0,8 segundos.
• Pongas 1 cocinero u 8, la línea de salida está en 0,5 segundos. Un solo set() despierta a todos los que esperan.

4-3. «Despiértame cuando llegue el género»: Condition (10 min)

Un Event solo tiene dos estados: encendido o apagado. Pero las esperas de una cocina suelen ser más concretas. Minho espera huevos; Sua, tofu. Cada vez que llega algo a la estantería hay que comprobar «¿ha llegado lo mío?» y, si no, volver a esperar. Para eso está Condition.

Condition es un cerrojo + una lista de «despiértame». Protege el estado compartido (la estantería del género) con el cerrojo y, cuando ese estado cambia, despierta a la gente de la lista.

pantry = []                             # estantería del género (estado compartido)
arrived = threading.Condition()         # cerrojo de la estantería + lista de «despiértame»
start = time.perf_counter()

def cook(name, need):
    with arrived:
        ok = arrived.wait_for(lambda: need in pantry, timeout=0.5)
        print(f"{time.perf_counter() - start:.1f}s {name} {need if ok else 'gave up'}")

def supplier():
    for item in ["tofu", "egg"]:        # el pescado hoy no llega
        time.sleep(0.1)
        with arrived:
            pantry.append(item)
            arrived.notify_all()        # despertar a todos los que esperan

threads = [threading.Thread(target=cook, args=a) for a in [("minho", "egg"), ("sua", "tofu"), ("tak", "fish")]]
threads.append(threading.Thread(target=supplier))
for th in threads:
    th.start()
for th in threads:
    th.join()
# 0.1s sua tofu
# 0.2s minho egg
# 0.5s tak gave up

• La estantería solo se mira y se modifica con el cerrojo tomado mediante with arrived:. La misma regla de la lección 3.
• wait_for(condición) espera hasta que la condición sea verdadera. Mientras espera, suelta el cerrojo un momento, y así el proveedor puede dejar género en la estantería. Al despertar, vuelve a tomar el cerrojo y comprueba otra vez la condición.
• Con timeout, cuando se acaba el tiempo devuelve el último valor de la condición. El pescado no llegó nunca, así que el señor Tak recibió False a los 0,5 segundos y se rindió.
• notify_all() despierta a todos los de la lista. Cada uno comprueba su propia condición y, si no se cumple, vuelve a dormir. Cuando llegó el tofu, Minho y el señor Tak también se despertaron, pero como no era lo suyo, siguieron esperando.

notify() despierta solo a uno de los que esperan. Va bien para trabajos en los que da igual quién se lo lleve, por ejemplo «que cualquier camarero se lleve un plato terminado».

plates = 0
counter = threading.Condition()

def cook():
    global plates
    time.sleep(0.1)
    with counter:
        plates += 1
        counter.notify()                # despertar solo a uno

def waiter(name):
    global plates
    with counter:
        counter.wait_for(lambda: plates > 0)
        plates -= 1
        print(name, "took a plate")

ws = [threading.Thread(target=waiter, args=(n,)) for n in ["w1", "w2"]]
cs = [threading.Thread(target=cook) for _ in range(2)]
for th in ws + cs:
    th.start()
for th in ws + cs:
    th.join()
print(plates)
# por ejemplo:
# w1 took a plate
# w2 took a plate
# 0

Como por cada plato solo se despierta a uno, hay menos casos en que dos se despiertan a la vez y se pelean por el mismo plato. En cambio, si cada persona espera una condición distinta, notify() puede despertar a quien no toca, así que usa notify_all(). Si dudas, notify_all() es lo seguro.

4-4. Los cuatro platos de la mesa de grupo, a la vez: Barrier (5 min)

En una mesa de cuatro hay que servir los cuatro platos a la vez. Si sale uno antes, los otros tres comensales esperan sentados mientras se enfría. Barrier(n) espera en el pase hasta que lleguen los n, y cuando están todos los deja salir juntos.

start = time.perf_counter()
table = threading.Barrier(4, action=lambda: print("serve 4 plates"))
done = []

def cook(name, sec):
    time.sleep(sec)                     # cada uno a su ritmo
    i = table.wait()                    # esperar en el pase hasta que lleguen los cuatro
    done.append((round(time.perf_counter() - start, 1), i, name))

crew = [("minho", 0.1), ("sua", 0.2), ("chef", 0.3), ("tak", 0.4)]
threads = [threading.Thread(target=cook, args=c) for c in crew]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(sorted(done))
# serve 4 plates
# [(0.4, 0, 'minho'), (0.4, 1, 'sua'), (0.4, 2, 'chef'), (0.4, 3, 'tak')]

• Minho terminó a los 0,1 segundos, pero los cuatro pasan juntos a los 0,4, cuando llega el más lento, el señor Tak.
• wait() devuelve a cada uno un número distinto, de 0 a n-1. En CPython es el orden de llegada: 0, 1, 2, 3. Con ese número puedes decidir «lo que hace solo uno» (por ejemplo, escribir un log con if i == 0:).
• La función de action= se ejecuta una sola vez, cuando llega el último. Es la señal para sacar los platos.
• Cuando todos han pasado, el Barrier vuelve a su estado inicial y sirve para la siguiente mesa. Si un wait(timeout=) se pasa de tiempo, el Barrier se rompe y todos los que esperaban reciben BrokenBarrierError.

4-5. Resumen de las herramientas de esta noche (4 min)

Las cinco están en threading, y todas se construyen sobre el cerrojo de la lección 3. De hecho, Semaphore, Event y Barrier usan un Condition por dentro, y Condition lleva dentro un cerrojo (un RLock si no le das otro).

Cierre (10 min)

El objetivo de esta noche era que nadie encendiera el fuego antes de la señal de apertura, usar como mucho tres fogones a la vez y sacar los seis pedidos en menos de 1 segundo. Con Event marcamos la línea de salida, con Semaphore(3) los fogones y con Lock contamos «cuántos hay ahora al fuego».

opened = threading.Event()
stoves = threading.Semaphore(3)
lock = threading.Lock()
on_fire = peak = 0
log = []
start = time.perf_counter()

def cook(n):
    global on_fire, peak
    opened.wait()                       # 1. nada de fuego antes de «¡Abrimos!»
    with stoves:                        # 2. como mucho tres fogones a la vez
        with lock:
            on_fire += 1
            peak = max(peak, on_fire)
        time.sleep(0.3)
        with lock:
            on_fire -= 1
            log.append((round(time.perf_counter() - start, 1), f"#{n}"))

threads = [threading.Thread(target=cook, args=(n,)) for n in range(1, 7)]
for th in threads:
    th.start()
time.sleep(0.3)                         # preparativos para recibir a los clientes
opened.set()
for th in threads:
    th.join()
print(sorted(log))
print(f"peak {peak}, total {time.perf_counter() - start:.2f}s")
# por ejemplo:
# [(0.6, '#1'), (0.6, '#2'), (0.6, '#3'), (0.9, '#4'), (0.9, '#5'), (0.9, '#6')]
# peak 3, total 0.90s

• Hasta los 0,3 segundos no cocina nadie. El primer plato sale a los 0,6 y el último a los 0,9: dentro del objetivo de 1 segundo.
• peak da 3 todas las veces. Qué números de pedido salen a los 0,6 segundos puede cambiar en cada ejecución.
• on_fire y peak los modifican los seis cocineros, así que van envueltos en lock. Aunque estemos dentro del semáforo.

Hacia el cierre, una clienta sentada sola junto a la ventana termina un cuenco de fideos. Mira una vez su reloj de pulsera, apunta algo en una libretita, paga en silencio y se marcha.

La chef va cerrando las llaves de los fogones una a una. «Hoy todo el mundo ha respetado la cola. Pero ¿y si, además de los fogones, solo hubiera un cuchillo y una tabla? Si Minho tiene el cuchillo y espera la tabla, y Sua tiene la tabla y espera el cuchillo, ¿qué pasa?»

Ejercicios

1. En la línea de tiempo de los fogones, deja 6 cocineros y 6 pedidos, cambia los fogones de 1 a 6 y apunta en una tabla el tiempo total. ¿Por qué con 4 y con 5 fogones se tarda lo mismo?
2. Construye tú el patrón de la señal de «fin del servicio» de 4-2. Crea closing = threading.Event() y haz que un hilo cocinero prepare un plato cada 0,1 segundos y se detenga cuando closing.is_set(). El hilo principal llama a closing.set() a los 0,35 segundos y luego hace join(). ¿Cuántos platos salieron?
3. Piensa qué pasaría si en el ejemplo del Barrier de 4-4 cambias Barrier(4) por Barrier(5). Para que no se quede parado para siempre de verdad, cámbialo a table.wait(timeout=1), ejecútalo y comprueba qué excepción salta.