Se cae la reserva de grupo: concurrencia estructurada

Diez y media de la noche. Un grupo ocupa seis mesas de golpe. Lou toma los pedidos y los pasa a cocina, y la cocina arranca los seis menús a la vez. Diez minutos después, levantan la mano en la mesa 3: alguien avisó tarde de que tiene alergia a los cacahuetes. La persona que organiza el grupo dice: «Entonces hoy nos vamos todos».

Minho no lo oye y sigue con el guiso de la mesa 5 al fuego. Sua le apaga el fuego y dice: «No había ninguna regla que dijera que, si se cae una mesa, se para todo».

La chef Yoon Seoyeon pulsa el temporizador. «Lo que empieza junto tiene que terminar junto. Esta noche toca prometer eso en código».

10-1. El punto débil de gather: si falla uno, los demás siguen hirviendo (5 min)

Probemos a llevar la reserva de grupo con el asyncio.gather de la lección 9. La mesa 2 falla a los 0,1 segundos.

import asyncio

async def table(n, sec, fail=False):
    await asyncio.sleep(sec)
    if fail:
        raise ValueError(f"T{n} allergy")
    print(f"T{n} served")

async def main():
    try:
        await asyncio.gather(table(1, 0.3), table(2, 0.1, fail=True), table(3, 0.5))
    except ValueError as e:
        print("caught:", e)
    await asyncio.sleep(0.6)   # mantenemos el local abierto un rato más

asyncio.run(main())
# caught: T2 allergy
# T1 served
# T3 served

• gather deja subir la primera excepción al instante. Por eso caught: sale primero.
• Pero las otras dos mesas no se cancelan. Platos que ya no espera nadie siguen hirviendo por detrás y acaban imprimiendo served.
• En un programa real, eso es servir comida a clientes que ya se fueron, o peticiones que siguen en marcha aunque nadie vaya a recoger su resultado.

gather(return_exceptions=True) solo mete las excepciones en la lista de resultados; tampoco para al resto. Si lo que quieres es «si falla uno, se para todo», hace falta otra herramienta.

10-2. TaskGroup: lo que empieza junto termina junto (12 min)

asyncio.TaskGroup (3.11) maneja las tareas como un solo paquete. Esta forma de agruparlas y gestionarlas se llama concurrencia estructurada (structured concurrency).

import asyncio

async def table(n, sec):
    await asyncio.sleep(sec)
    return f"T{n}"

async def main():
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(table(n, 0.1 * n)) for n in range(1, 4)]
    # al llegar aquí, las tres tareas han terminado
    print([t.result() for t in tasks])

asyncio.run(main())
# ['T1', 'T2', 'T3']

TaskGroup hace tres promesas.

1. Al salir del bloque async with, todas las tareas creadas dentro han terminado. No hace falta hacerles await por separado.
2. Si una tarea lanza una excepción, cancela todas las demás que sigan en marcha.
3. Cuando termina de cancelar, reúne las excepciones de los fallos en un único ExceptionGroup y lo lanza.

Como tg.create_task guarda una referencia a cada tarea, tampoco hay que preocuparse de las «tareas que desaparecen al perder la referencia» de la lección 9. Ahora, la reserva de grupo en la que falla la mesa 3.

import asyncio

async def table(n, sec, fail=False):
    try:
        await asyncio.sleep(sec)
        if fail:
            raise ValueError(f"T{n} allergy")
        print(f"T{n} served")
    except asyncio.CancelledError:
        print(f"T{n} cancelled")
        raise                      # la cancelación recibida se vuelve a lanzar

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(table(1, 0.05))
            tg.create_task(table(2, 0.3))
            tg.create_task(table(3, 0.1, fail=True))
            tg.create_task(table(4, 0.5))
    except* ValueError as eg:
        print("group failed:", eg.exceptions)
    await asyncio.sleep(0.6)

asyncio.run(main())
# T1 served
# T2 cancelled      (el orden de T2 y T4 cambia en cada ejecución)
# T4 cancelled
# group failed: (ValueError('T3 allergy'),)

• T1 terminó antes del fallo, así que se sirve sin más.
• En cuanto T3 falla, a los 0,1 segundos, les llega la cancelación a T2 y T4, que seguían hirviendo. El orden en que se imprimen los mensajes de cancelación puede cambiar de una ejecución a otra.
• Aunque esperes 0,6 segundos más, no aparece ningún served más. A diferencia de gather, no queda ninguna olla hirviendo a escondidas.
• Dentro de una tarea cancelada, asyncio.CancelledError salta en el punto del await. Por eso puedes capturarla con except y recoger. La cancelación recibida hay que volver a lanzarla siempre (10-5).

La línea de tiempo de abajo ejecuta un TaskGroup de verdad en el servidor. Cuanto más alto es el número de mesa, un poco más tarda su plato (T1 0,3 s, T2 0,345 s, … T6 unos 0,53 s). La mesa que elijas como «Mesa que falla» falla justo cuando su plato estaba a punto de terminar.

• Con la configuración inicial (mesa que falla: 3), T1 y T2 terminan antes, y en el instante en que T3 falla, hacia los 0,4 segundos, las tres barras de T4–T6 se cortan en gris en el mismo momento. T1 y T2, que ya habían terminado, se quedan como están.
• Si la mesa que falla es la 1, a los 0,3 segundos se cortan las otras cinco; si es la 6, falla solo T6, después de que hayan terminado las otras cinco.
• Con la mesa que falla en «Ninguno», las seis barras llegan hasta el final. La más tardía, T6, termina hacia los 0,53 segundos.
• Si bajas el límite de tiempo a unos 0,4 segundos, solo terminan algunas de las primeras mesas, y las que seguían en marcha al llegar el límite se cortan en gris. Los platos que ya habían terminado no se deshacen.

10-3. ExceptionGroup y except* (7 min)

Varias tareas pueden fallar a la vez. Por eso TaskGroup no lanza una sola excepción, sino un paquete de excepciones.

import asyncio

async def boom(exc, sec):
    await asyncio.sleep(sec)
    raise exc

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(boom(ValueError("T2 allergy"), 0.1))
            tg.create_task(boom(KeyError("T5"), 0.1))
    except Exception as e:
        print(type(e).__name__, "|", e)
        print(e.exceptions)

asyncio.run(main())
# ExceptionGroup | unhandled errors in a TaskGroup (2 sub-exceptions)
# (ValueError('T2 allergy'), KeyError('T5'))

ExceptionGroup (3.11) es una excepción que guarda una lista de excepciones en .exceptions. Para capturarlas por tipo dentro del paquete, se usa except*.

eg = ExceptionGroup("kitchen", [ValueError("T2 allergy"), KeyError("T5"), ValueError("T6 salt")])

try:
    raise eg
except* ValueError as g:
    print("ValueError:", [str(e) for e in g.exceptions])
except* KeyError as g:
    print("KeyError:", g.exceptions)
# ValueError: ['T2 allergy', 'T6 salt']
# KeyError: (KeyError('T5'),)

• Con un except normal se ejecuta un solo bloque. Con except* se ejecutan todos los bloques que encajan. Arriba se ejecutan los dos.
• g siempre es un ExceptionGroup: un paquete más pequeño con solo las excepciones que encajan.
• Las excepciones que no encajan en ningún except* vuelven a subir en un paquete con las que quedan.

eg = ExceptionGroup("kitchen", [ValueError("T2"), KeyError("T5")])
try:
    try:
        raise eg
    except* ValueError:
        print("handled ValueError")
except ExceptionGroup as rest:
    print("left:", rest.exceptions)
print(isinstance(eg, Exception), eg.subgroup(ValueError).exceptions)
# handled ValueError
# left: (KeyError('T5'),)
# True (ValueError('T2'),)

except* tiene dos reglas. En 3.14, las dos se detectan como errores de sintaxis antes de ejecutar nada.

src1 = '''
def f():
    try:
        pass
    except* ValueError:
        return 1
'''
src2 = '''
try:
    pass
except ValueError:
    pass
except* KeyError:
    pass
'''
for src in (src1, src2):
    try:
        compile(src, "<menu>", "exec")
    except SyntaxError as e:
        print("SyntaxError:", e.msg)
# SyntaxError: 'break', 'continue' and 'return' cannot appear in an except* block
# SyntaxError: cannot have both 'except' and 'except*' on the same 'try'

• Dentro de un bloque except* no se puede usar return, break ni continue. Varios bloques tienen que ejecutarse uno tras otro, y no se puede salir a mitad. Guarda el resultado en una variable y devuélvelo fuera del try.
• En un mismo try no se pueden mezclar except y except*. Si necesitas los dos, anida dos try. Así lo hace también el motor de los bancos de pruebas de este curso.

10-4. Límites de tiempo: asyncio.timeout y wait_for (8 min)

La segunda regla de esta noche es «ningún plato pasa del tiempo fijado». asyncio.timeout() (3.11) pone un límite de tiempo a todo un bloque.

import asyncio, time

async def stew(sec):
    await asyncio.sleep(sec)
    return "stew"

async def main():
    start = time.perf_counter()
    try:
        async with asyncio.timeout(0.2):
            await stew(1.0)
    except TimeoutError:
        print(f"timeout after {time.perf_counter() - start:.2f}s")

    try:
        print(await asyncio.wait_for(stew(0.1), timeout=0.2))
        print(await asyncio.wait_for(stew(1.0), timeout=0.2))
    except TimeoutError:
        print("wait_for: timeout")

    print(asyncio.TimeoutError is TimeoutError)

asyncio.run(main())
# timeout after 0.20s
# stew
# wait_for: timeout
# True

• Si se pasa el tiempo, el trabajo de dentro del bloque se cancela y fuera del bloque sale un TimeoutError integrado. Desde 3.11, asyncio.TimeoutError es lo mismo que el TimeoutError integrado.
• asyncio.wait_for(corrutina, timeout=) es la forma de siempre de poner un límite a una sola corrutina. Es cómoda cuando quieres el resultado directamente.
• asyncio.timeout puede agrupar varios await en un solo bloque. Algo como «20 minutos en total, desde que se toma el pedido hasta que se sirve».

El límite también se puede fijar como «a tal hora» en vez de «dentro de tantos segundos», y se puede cambiar después de entrar en el bloque.

import asyncio, time

async def main():
    loop = asyncio.get_running_loop()
    start = time.perf_counter()
    deadline = loop.time() + 0.3            # el «instante» que será dentro de 0.3 s
    try:
        async with asyncio.timeout_at(deadline):
            await asyncio.sleep(1)
    except TimeoutError:
        print(f"timeout_at: {time.perf_counter() - start:.1f}s")

    start = time.perf_counter()
    try:
        async with asyncio.timeout(None) as cm:   # al principio, sin límite
            cm.reschedule(loop.time() + 0.2)      # más tarde se fija en 0.2 s
            print(cm.when() is not None)
            await asyncio.sleep(1)
    except TimeoutError:
        print(f"reschedule: {time.perf_counter() - start:.1f}s, expired={cm.expired()}")

asyncio.run(main())
# timeout_at: 0.3s
# True
# reschedule: 0.2s, expired=True

• timeout_at(instante) recibe un instante absoluto medido con loop.time(). Viene bien cuando varias fases comparten una misma «hora límite».
• timeout(None) empieza sin límite. Con cm.reschedule(instante) puedes fijar el límite más tarde o aplazarlo.

10-5. Cancelar: cancel, CancelledError y finally (6 min)

TaskGroup y los límites de tiempo, al final, hacen lo mismo: enviar una cancelación a una tarea. Para enviarla tú directamente, usa task.cancel().

import asyncio

async def braise():
    print("stove on")
    try:
        await asyncio.sleep(10)
    finally:
        print("stove off")          # aunque se cancele, se apaga el fuego

async def main():
    task = asyncio.create_task(braise())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancelled:", task.cancelled())

asyncio.run(main())
# stove on
# stove off
# cancelled: True

• cancel() no mata la tarea en el acto. Solo hace que salte CancelledError en el próximo await en el que la tarea se detenga.
• Lo de recoger, como apagar el fuego o cerrar conexiones, va en un try/finally. Se ejecuta siempre: con cancelación, con éxito o con fallo.
• CancelledError no hereda de Exception, sino de BaseException. Está diseñado así para que un except Exception: no se trague una cancelación por descuido.

¿Qué pasa si te tragas la cancelación?

import asyncio, time

async def stubborn():
    try:
        await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        pass                          # se ha tragado la cancelación
    await asyncio.sleep(0.5)          # y sigue cocinando
    return "late"

async def main():
    start = time.perf_counter()
    try:
        async with asyncio.timeout(0.1):
            print(await stubborn())
    except TimeoutError:
        print("timeout")
    print(f"{time.perf_counter() - start:.1f}s")

asyncio.run(main())
# late
# 0.6s

Pusimos un límite de 0,1 segundos, pero tardó 0,6 y ni siquiera saltó TimeoutError. El límite de tiempo funciona mediante una cancelación, y la corrutina se la comió. La «cancelación del resto» de TaskGroup se viene abajo exactamente igual. Por eso, si recibes un CancelledError, recoge lo que tengas que recoger y vuelve a lanzarlo (raise).

10-6. shield para frenar cancelaciones y cancelling para contarlas (2 min)

Hay trabajos que no pueden quedarse a medias, como guardar un tique. asyncio.shield() impide que una cancelación que llega de fuera baje hasta la tarea de dentro.

import asyncio

async def save_receipt():
    await asyncio.sleep(0.3)
    print("receipt saved")
    return "ok"

async def main():
    inner = asyncio.create_task(save_receipt())
    try:
        async with asyncio.timeout(0.1):
            await asyncio.shield(inner)
    except TimeoutError:
        print("waiter gave up")
    print(await inner)              # la tarea de dentro siguió en marcha

asyncio.run(main())
# waiter gave up
# receipt saved
# ok

Quien esperaba se rindió a los 0,1 segundos, pero la tarea de dentro llegó hasta el final. Para recoger su resultado más tarde, tienes que guardar una referencia a esa tarea.

Task.cancelling() devuelve cuántas peticiones de cancelación ha recibido la tarea, y Task.uncancel() resta una. Son herramientas que asyncio.timeout y TaskGroup usan por dentro para distinguir «¿esta cancelación la envié yo?». Rara vez tendrás que usarlas directamente.

import asyncio

async def worker():
    try:
        await asyncio.sleep(1)
    except asyncio.CancelledError:
        me = asyncio.current_task()
        print("cancelling:", me.cancelling())
        me.uncancel()
        print("after uncancel:", me.cancelling())
        return "rescued"

async def main():
    t = asyncio.create_task(worker())
    await asyncio.sleep(0.05)
    t.cancel()
    print(await t)

asyncio.run(main())
# cancelling: 1
# after uncancel: 0
# rescued

Cierre (10 min)

Aplicamos las dos reglas de esta noche en una misma cocina: un límite de 0,6 segundos por mesa, y las seis mesas agrupadas en un solo TaskGroup.

import asyncio, time

async def table(n, sec, allergy, stopped):
    try:
        async with asyncio.timeout(0.6):     # ningún plato pasa de 0.6 s
            await asyncio.sleep(sec)
            if n == allergy:
                raise ValueError(f"T{n} allergy")
            return f"T{n}"
    except asyncio.CancelledError:
        stopped.append(f"T{n}")             # apaga el fuego y levanta la mano
        raise

async def service(menu, allergy=None):
    start, stopped = time.perf_counter(), []
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(table(n, s, allergy, stopped)) for n, s in menu.items()]
        print("served:", [t.result() for t in tasks])
    except* ValueError as eg:
        print("cancelled by:", [str(e) for e in eg.exceptions], "stopped:", sorted(stopped))
    except* TimeoutError as eg:
        print("too slow:", len(eg.exceptions))
    print(f"{time.perf_counter() - start:.1f}s")

MENU = {1: 0.30, 2: 0.35, 3: 0.20, 4: 0.40, 5: 0.45, 6: 0.50}
asyncio.run(service(MENU))
asyncio.run(service(MENU, allergy=3))
asyncio.run(service({**MENU, 6: 0.9}))
# served: ['T1', 'T2', 'T3', 'T4', 'T5', 'T6']
# 0.5s
# cancelled by: ['T3 allergy'] stopped: ['T1', 'T2', 'T4', 'T5', 'T6']
# 0.2s
# too slow: 1
# 0.6s

• En una noche normal, las seis mesas hierven juntas y todo termina hacia los 0,5 segundos, al ritmo de la más larga, T6.
• Si la mesa 3 falla a los 0,2 segundos, las otras cinco apagan el fuego en ese mismo momento.
• Si un plato pasa de 0,6 segundos, el TimeoutError de esa mesa sube dentro del paquete y, otra vez, el resto se para.

Cerca del cierre, la clienta de la ventana termina su bol de fideos y cierra su libretita. Al pagar, le dice a la chef en pocas palabras: «Vuelvo el viernes que viene».

La chef asiente y se gira hacia su segundo de cocina. «Tanto la cancelación como el límite de tiempo pasan, al final, en un await. Pero el horno viejo del rincón no sabe lo que es un await. Si Lou se queda plantada delante esperándolo, ¿qué le pasa a la sala?».

Ejercicios

1. En la línea de tiempo, cambia la mesa que falla de la 1 a la 6. Explica por qué solo sobreviven las mesas con un número menor que la que falla. Después, pon la mesa que falla en «Ninguno» y sube el límite de tiempo de 0,3 a 0,55 segundos, observando cómo aumentan las mesas que llegan hasta el final.
2. Cambia el ejemplo de gather de 10-1 por un TaskGroup y haz que, cuando falle la mesa 2, las mesas 1 y 3 impriman cancelled.
3. En table del código del cierre, borra el raise y pon en su lugar await asyncio.sleep(0.3) (ignorar la cancelación y seguir hirviendo), y ejecútalo con allergy=3. Mira cómo cambia el tiempo y apunta qué ha ido mal.