El riel de comandas: queue

Sexta noche, viernes. El personal de sala entra corriendo en la cocina con las comandas y se las da al primero que encuentra. Minho y Sua agarran la misma comanda a la vez, y salen dos boles del mismo fideo. Mientras tanto, el señor Tak espera cuchillo en mano a que le toque algo.

La chef Yoon Seoyeon clava una barra larga de metal sobre la entrada de la cocina: un riel de comandas. «La sala solo cuelga aquí. Los cocineros solo retiran de aquí. Y nadie le habla a nadie».

6-1. Quien cuelga y quien retira (8 min)

Separar el lado que genera el trabajo (la sala) del lado que lo procesa (los cocineros), con una cola entre ambos, es un diseño que se llama patrón productor-consumidor. En Python, esa cola es queue.Queue.

import queue

rail = queue.Queue()        # riel de comandas
for n in range(1, 5):
    rail.put(f"#{n}")       # colgar
print(rail.qsize(), rail.get(), rail.get(), rail.qsize())
# 4 #1 #2 2

La primera comanda que se cuelga es la primera que sale (FIFO: primero en entrar, primero en salir). Ahora pongamos en marcha como hilos a una persona de sala y a dos cocineros.

import threading
import time

rail = queue.Queue()
served = []

def hall():                          # productor: toma pedidos y los cuelga
    for n in range(1, 7):
        time.sleep(0.05)
        rail.put(f"#{n}")

def cook(name):                      # consumidor: los retira del riel y los prepara
    for _ in range(3):
        order = rail.get()           # si el riel está vacío, espera aquí
        time.sleep(0.1)
        served.append((name, order))

ts = [threading.Thread(target=hall),
      threading.Thread(target=cook, args=("minho",)),
      threading.Thread(target=cook, args=("sua",))]
for t in ts: t.start()
for t in ts: t.join()
print(sorted(order for _, order in served))
# ['#1', '#2', '#3', '#4', '#5', '#6']

• Los seis pedidos se han preparado exactamente una vez. Nunca agarran dos cocineros la misma comanda.
• No hay ni un Lock en todo el código. Queue ya lleva dentro un cerrojo y la Condition de la lección 4, así que put y get atienden a una persona cada vez por sí solos. La cola sustituye al cerrojo.
• Si el riel está vacío, get() espera hasta que se cuelgue una comanda. El cocinero no tiene que estar dando vueltas preguntando «¿hay pedidos?».

Si tus hilos tienen que pasarse datos, antes de poner un cerrojo a una variable compartida, piensa primero en pasárselos por una cola. Cuanto menos código manipule cerrojos a mano, menos accidentes como los de las lecciones 3 y 5.

6-2. Cómo saber que todos los pedidos están listos: task_done y join (8 min)

En el ejemplo anterior, cada cocinero tenía fijado «preparo tres y me voy a casa». En una cocina de verdad no sabes cuántos pedidos van a llegar. Los cocineros miran el riel sin parar, y a la chef solo le interesa saber si ya están listos todos los pedidos que se colgaron.

rail = queue.Queue()

def cook(name):
    while True:
        order = rail.get()
        time.sleep(0.1)
        rail.task_done()          # «esta comanda está lista»

for name in ("minho", "sua", "tak"):
    threading.Thread(target=cook, args=(name,), daemon=True).start()

start = time.perf_counter()
for n in range(1, 10):
    rail.put(f"#{n}")
rail.join()                       # hasta que estén listas todas las comandas colgadas
print(f"all 9 done: {time.perf_counter() - start:.2f}s")
# all 9 done: 0.30s

Por dentro, la cola lleva la cuenta de los pedidos que aún no se han terminado.

• Sacar una comanda con get() no significa haberla terminado. El número solo baja cuando llamas a task_done() después de cocinar.
• Los hilos de los cocineros vigilan el riel para siempre con while True, así que los hemos creado con daemon=True. Cuando todos los pedidos están listos, join() se desbloquea, y al terminar el programa los cocineros se van a casa con él.
• Si llamas a task_done() más veces de las que has sacado, salta ValueError: task_done() called too many times.

6-3. Cuando el riel se llena (10 min)

Puedes fijar cuántos huecos tiene el riel. Con Queue(maxsize=2) solo caben dos comandas. Si haces put() en un riel lleno, espera quien cuelga. Es como si, cuando la cocina va atrasada, la sala bajara el ritmo al tomar pedidos.

rail = queue.Queue(maxsize=2)     # solo caben dos comandas
rail.put("#1")
rail.put("#2")
print(rail.full())
# True

try:
    rail.put("#3", timeout=0.5)   # si en 0.5 s no queda un hueco libre
except queue.Full:
    print("Full after 0.5s")
# Full after 0.5s

try:
    rail.put_nowait("#3")         # = put("#3", block=False)
except queue.Full:
    print("Full right away")
# Full right away

print(rail.get_nowait(), rail.get(timeout=0.5))
# #1 #2

try:
    rail.get(timeout=0.5)         # el riel está vacío
except queue.Empty:
    print("Empty after 0.5s")
# Empty after 0.5s

Si no pones maxsize, o lo pones a 0, el riel crece sin límite. Si la sala va mucho más rápido que la cocina, las comandas se amontonan sin fin y llenan la memoria. Con un tamaño fijo, el lado rápido espera al lento. Esto se llama contrapresión (backpressure).

rail = queue.Queue(maxsize=2)

def cook():                           # un solo cocinero, lento
    while True:
        rail.get()
        time.sleep(0.1)
        rail.task_done()

threading.Thread(target=cook, daemon=True).start()

start = time.perf_counter()
for n in range(1, 7):
    rail.put(f"#{n}")                 # sala rápida: si no hay hueco, espera en put
print(f"hall done at {time.perf_counter() - start:.1f}s")
rail.join()
print(f"kitchen done at {time.perf_counter() - start:.1f}s")
# hall done at 0.3s
# kitchen done at 0.6s

La sala quería colgar las seis comandas en un instante, pero como el riel solo tiene dos huecos, se pasó 0,3 segundos esperando en put. Si quitas maxsize, la sala termina enseguida y las comandas se acumulan en el riel.

Compruébalo tú en la línea de tiempo de abajo. Las filas de arriba son el personal de sala; las de abajo, los cocineros. Cada persona de sala tarda 0,1 segundos en tomar un pedido (naranja) y además pasa un rato esperando a que quede un hueco en el riel (amarillo).

• Con los valores iniciales (2 cocineros, riel de 3), tarda unos 1,6 segundos y en el riel llega a haber como máximo 3 comandas. La sala empieza a esperar hacia los 0,3 segundos.
• Si subes el riel a 8, la sala no espera nada y termina hacia los 0,5 segundos. Aun así, el tiempo total sigue siendo de unos 1,6 segundos. Si la cocina es lenta, un riel más grande no lo arregla.
• Con 4 cocineros, el tiempo total baja a cerca de 1,0 segundos. Para saber dónde está el cuello de botella, mira en qué fila salen las barras amarillas largas.

6-4. La señal de cierre: centinelas y shutdown (9 min)

En lugar de dejar a los cocineros abandonados con daemon=True, a veces quieres avisarles de que «se acabó el servicio» y que se vayan a casa de forma ordenada. El método clásico es el centinela (sentinel): colgar una comanda especial que dice «fin del servicio».

STOP = None                         # centinela: la comanda de «fin del servicio»
rail = queue.Queue()
served = []

def cook(name):
    while True:
        order = rail.get()
        if order is STOP:
            break                   # al recibir el centinela, se va a casa
        time.sleep(0.05)
        served.append(order)

cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
    rail.put(f"#{n}")
for _ in cooks:
    rail.put(STOP)                  # se cuelgan tantos como cocineros haya
for t in cooks: t.join()
print(len(served), [t.is_alive() for t in cooks])
# 6 [False, False]

Hay que colgar tantos centinelas como cocineros. Si cuelgas solo uno, solo se va un cocinero y los demás se quedan para siempre esperando delante de un riel vacío. Como el riel es FIFO, los centinelas solo salen después de todos los pedidos colgados antes.

A partir de Python 3.13 se puede cerrar el propio riel. Después de llamar a Queue.shutdown(), cualquier put() lanza queue.ShutDown. get() entrega primero todas las comandas que quedan y después lanza queue.ShutDown. Ya no hace falta contar cocineros.

rail = queue.Queue()
served = []

def cook(name):
    while True:
        try:
            order = rail.get()
        except queue.ShutDown:      # el riel está cerrado y no quedan comandas
            return
        time.sleep(0.05)
        served.append(order)
        rail.task_done()

cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
    rail.put(f"#{n}")
rail.shutdown()                     # no se aceptan más; las comandas colgadas se terminan
for t in cooks: t.join()
print(len(served))
# 6

try:
    rail.put("#7")
except queue.ShutDown:
    print("rail is closed")
# rail is closed

Cuando hay que cerrar a toda prisa, se usa shutdown(immediate=True). Se tiran todas las comandas que quedan en el riel, y los get() que estaban esperando lanzan queue.ShutDown en el acto.

rail = queue.Queue()
served = []

def cook(name):
    while True:
        try:
            order = rail.get()
        except queue.ShutDown:
            return
        time.sleep(0.1)
        served.append(order)
        rail.task_done()

cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
    rail.put(f"#{n}")
time.sleep(0.15)                    # los dos están con #3 y #4
rail.shutdown(immediate=True)       # se tiran #5 y #6, que seguían en el riel, y se cierra ya
rail.join()                         # se desbloquea cuando acaban #3 y #4, que ya tenían en la mano
for t in cooks: t.join()
print(sorted(served), rail.qsize())
# ['#1', '#2', '#3', '#4'] 0

Ni siquiera immediate=True le quita a nadie el pedido que ya tiene en la mano. Por eso join() se desbloquea después de que se llame a task_done() para esos pedidos.

6-5. Rieles con otra forma (5 min)

El módulo queue tiene más rieles que sacan las cosas en otro orden.

PriorityQueue: saca primero el valor más pequeño. Sirve para preparar antes los pedidos de las mesas VIP. Lo habitual es meter tuplas (prioridad, número, pedido).

import itertools

vip_first = queue.PriorityQueue()
ticket = itertools.count(1)          # a igual prioridad, por orden de llegada
for prio, dish in [(2, "noodle"), (1, "VIP soup"), (2, "rice"), (1, "VIP noodle")]:
    vip_first.put((prio, next(ticket), dish))

while not vip_first.empty():
    print(vip_first.get())
# (1, 2, 'VIP soup')
# (1, 4, 'VIP noodle')
# (2, 1, 'noodle')
# (2, 3, 'rice')

Sin el número del medio, cuando dos prioridades empatan se comparan los terceros valores. Si el pedido es algo que no se puede comparar, como un dict, revienta.

pq = queue.PriorityQueue()
pq.put((1, {"dish": "soup"}))
try:
    pq.put((1, {"dish": "rice"}))   # a igual prioridad, intenta comparar dos dict y...
except TypeError as e:
    print(e)
# '<' not supported between instances of 'dict' and 'dict'

También están LifoQueue, que saca primero lo último, como una pila de platos, y SimpleQueue, la más sencilla de todas.

plates = queue.LifoQueue()           # pila de platos: primero el último que se puso
for p in ("p1", "p2", "p3"):
    plates.put(p)
print(plates.get(), plates.get(), plates.get())
# p3 p2 p1

simple = queue.SimpleQueue()         # el riel más sencillo: sin maxsize, task_done ni join
simple.put("#1")
print(simple.get(), hasattr(simple, "task_done"))
# #1 False

Cierre (10 min)

El objetivo de esta noche era que tres cocineros prepararan 12 pedidos exactamente una vez cada uno, sin que faltara ninguno, y saber con certeza cuándo estaba todo listo. Dos personas de sala toman pedidos por turnos, y el riel tiene cuatro huecos.

rail = queue.Queue(maxsize=4)        # riel de cuatro huecos
served = []

def hall(orders):                    # sala: toma pedidos y los cuelga
    for n in orders:
        time.sleep(0.02)
        rail.put(f"#{n}")

def cook(name):
    while True:
        try:
            order = rail.get()
        except queue.ShutDown:
            return
        time.sleep(0.1)
        served.append((name, order))
        rail.task_done()

start = time.perf_counter()
halls = [threading.Thread(target=hall, args=(range(k, 13, 2),)) for k in (1, 2)]
cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua", "tak")]
for t in halls + cooks: t.start()
for t in halls: t.join()             # cuando la sala termina de tomar pedidos
rail.shutdown()                      # se cierra el riel
rail.join()                          # y se espera a que estén listas todas las comandas colgadas
for t in cooks: t.join()
orders = [o for _, o in served]
print(len(orders), len(set(orders)), f"{time.perf_counter() - start:.2f}s")
# 12 12 0.44s  (el tiempo es aproximado)

12 platos preparados y 12 pedidos distintos. Ningún pedido salió dos veces y no faltó ninguno. El señor Tak también hizo su parte. Y en este código no hay ni una línea con Lock.

La chef le da un golpecito al riel. «Ahora la sala y la cocina ya no tienen que hablarse. Pero ¿y si el viernes que viene entra un pedido de grupo de 30 platos? ¿Vas a contratar a 30 cocineros?»

Ejercicios

1. Cambia cook de 6-2 para que solo el pedido #5 lance un ValueError. ¿Se desbloqueará rail.join()? Para no quedarte esperando para siempre, lánzalo con waiter = threading.Thread(target=rail.join, daemon=True), haz waiter.join(timeout=1) e imprime waiter.is_alive() para comprobarlo. Después, arréglalo con try/finally.
2. En la línea de tiempo, prueba con 4 personas de sala, 1 cocinero y un riel de 1 hueco. Explica en qué fila aparecen las barras amarillas y dónde está el cuello de botella.
3. Cambia el código del cierre para usar PriorityQueue y que los pedidos pares se preparen primero, como VIP. shutdown() funciona igual en PriorityQueue.