Una sola encargada de sala: primeros pasos con asyncio

Desde esta noche, la sala cambia. Solo hay una persona atendiendo: Lou. Y diez mesas. La chef no ha contratado a nadie más.

Lou recorre la sala sin parar, vuelta tras vuelta. Toma un pedido, lo pasa a cocina y, en lugar de esperar el plato, se va directa a la mesa siguiente. Cuando suena el timbre de la cocina, sale con ese plato. Una vuelta, y otra, y otra. Por esa forma de recorrer siempre el mismo camino respondiendo a los timbres (los eventos), desde el primer día los cocineros le pusieron un apodo: el bucle de eventos.

Lou se ata el delantal y dice: «Lo que más odio es quedarme plantada delante de una mesa esperando. Mientras espero, las otras mesas me están esperando a mí.»

La chef asiente. «Hasta ayer añadíamos gente. Hoy toca aprender a aprovechar el tiempo de espera de una sola persona. En Python se llama asyncio.»

9-1. async def crea corrutinas (7 min)

Si pones async delante de una función, se convierte en una función corrutina. Empecemos por lo raro: la llamas y no se ejecuta.

import asyncio

async def take_order(table):
    print("order", table)
    return f"T{table}"

c = take_order(1)
print(type(c).__name__)
print(c)
# coroutine
# <coroutine object take_order at 0x...>
# <sys>:0: RuntimeWarning: coroutine 'take_order' was never awaited

• take_order(1) no imprimió "order 1". No se ejecutó ni una línea del cuerpo de la función.
• Lo que devuelve es un objeto corrutina. Es como una nota que dice «tomar el pedido de la mesa 1».
• Si tiras la nota sin ejecutarla nunca, al final Python lanza el aviso was never awaited. Si ves este aviso, casi siempre es que se te olvidó un await.

Hay dos maneras de ejecutar la nota: en el nivel superior, pasándosela a asyncio.run(); dentro de otra corrutina, con await.

print(asyncio.run(take_order(1)))
# order 1
# T1

c = take_order(2)
asyncio.run(c)
asyncio.run(c)
# order 2
# RuntimeError: cannot reuse already awaited coroutine

Un objeto corrutina solo se puede usar una vez. Para repetir el mismo pedido, vuelve a llamar a take_order(2) y obtendrás una nota nueva.

9-2. await y asyncio.run (7 min)

await significa «yo espero aquí hasta que esto termine; mientras tanto, el bucle puede hacer otras cosas». await solo se puede usar dentro de async def. Si lo pones dentro de un def normal, salta SyntaxError: 'await' outside async function.

import asyncio
import time

async def table(n, wait):
    print(f"T{n} order")
    await asyncio.sleep(wait)   # mientras espera a la cocina, Lou queda libre
    print(f"T{n} served")
    return f"T{n}"

async def one_by_one():
    a = await table(1, 0.3)
    b = await table(2, 0.3)
    return [a, b]

start = time.perf_counter()
print(asyncio.run(one_by_one()), f"{time.perf_counter() - start:.2f}s")
# T1 order
# T1 served
# T2 order
# T2 served
# ['T1', 'T2'] 0.62s

• asyncio.sleep(0.3) es la versión asyncio de time.sleep(0.3). Mientras espera, cede el turno al bucle.
• Pero el resultado es 0,3 + 0,3 = unos 0,6 segundos. Hasta que no termina await table(1, ...), no se pasa a la línea siguiente. Es como si Lou se quedara delante de la mesa 1 hasta que saliera el plato.
• await no significa «a la vez», sino «esperar». Para esperar varias cosas a la vez, hace falta la herramienta del apartado siguiente.

asyncio.run(main()) es la puerta de entrada del programa.

1. Crea un bucle de eventos nuevo (Lou entra a trabajar).
2. Ejecuta la corrutina main() hasta que termina.
3. Cancela las tareas que queden y cierra el bucle (Lou se va a casa).

Normalmente se llama una sola vez en el programa, y solo en el nivel superior. Si lo llamas donde ya hay un bucle en marcha, salta RuntimeError: asyncio.run() cannot be called from a running event loop.

9-3. Qué hace el bucle de eventos (6 min)

El bucle de eventos es un bucle que corre en un solo hilo. Lo que hace es lo mismo que Lou recorriendo la sala.

1. Elige una corrutina que pueda avanzar ahora.
2. La ejecuta hasta el siguiente await.
3. Apunta qué está esperando esa corrutina (un temporizador, una respuesta de red) y pasa a la siguiente.
4. Si no hay ninguna que pueda avanzar, espera el timbre que vaya a sonar primero y vuelve a despertar a esa corrutina.

Si le das tres mesas a la vez (con gather, del siguiente apartado), el recorrido de Lou es este. La mesa 1 tiene un plato de 0,3 segundos; la 2, de 0,2; la 3, de 0,1.

Recuerda dos cosas.

• Hay una sola persona. Hay un único hilo, y en cada instante solo se está ejecutando una corrutina. Se gana velocidad porque los tiempos de espera se solapan, no porque haya más manos.
• El turno solo cambia en un await. A los hilos el sistema operativo los corta en cualquier momento (lección 3), pero una corrutina solo deja su sitio cuando ella misma hace await. Por eso el código que hay entre dos await nunca se interrumpe a medias.

9-4. gather: esperar a la vez (8 min)

asyncio.gather(corrutina, corrutina, ...) sube varias corrutinas al bucle de una vez y espera hasta que terminan todas.

async def together():
    return await asyncio.gather(table(1, 0.3), table(2, 0.2), table(3, 0.1))

start = time.perf_counter()
print(asyncio.run(together()), f"{time.perf_counter() - start:.2f}s")
# T1 order
# T2 order
# T3 order
# T3 served
# T2 served
# T1 served
# ['T1', 'T2', 'T3'] 0.31s

• El tiempo no es la suma (0,6 segundos), sino lo que dura la más larga (0,3 segundos).
• Terminaron en el orden T3 → T2 → T1, pero la lista de resultados va en el orden en que se pasaron. Igual que executor.map en la lección 7.
• Si pasas varias corrutinas en una lista, despliégalas con un asterisco: gather(*coros).

En el banco de pruebas de abajo, alterna entre «await mesa por mesa» y «Todo a la vez con gather». Cada fila es una mesa, y todas las barras tienen el color de «Esperando». Lou casi no usa las manos: casi todo es tiempo esperando a la cocina.

• Mesa por mesa: el tiempo crece con el número de mesas. Con 10, unos 2,6 segundos.
• gather: sea 1 mesa o 12, unos 0,25 segundos. El efecto que en la lección 1 conseguimos contratando a ocho cocineros, ahora lo conseguimos con una sola persona.
• Este banco de pruebas también ejecuta un bucle asyncio de verdad en el servidor y dibuja el instante en que cada mesa empezó y terminó.

Si una mesa sale mal: return_exceptions

Si una de las corrutinas que le pasaste a gather lanza una excepción, gather la propaga tal cual. Si quieres recibir todos los resultados, pasa return_exceptions=True.

async def table(n):
    await asyncio.sleep(0.1 * n)
    if n == 2:
        raise ValueError(f"T{n} wrong dish")
    return f"T{n}"

async def main():
    results = await asyncio.gather(table(1), table(2), table(3), return_exceptions=True)
    print(results)
    try:
        await asyncio.gather(table(1), table(2), table(3))
    except ValueError as e:
        print("ValueError:", e)

asyncio.run(main())
# ['T1', ValueError('T2 wrong dish'), 'T3']
# ValueError: T2 wrong dish

• Con return_exceptions=True, la excepción también ocupa su casilla en la lista de resultados. Revisa cada resultado con isinstance(r, Exception).
• Con el valor por defecto (False), aunque se propague la excepción, el resto de mesas no se cancela y siguen en marcha. «Si una falla, para las demás» lo hace el TaskGroup de la lección 10.

9-5. create_task: encargarlo antes (6 min)

asyncio.create_task(corrutina) envuelve la corrutina en una tarea y la programa en el bucle enseguida. La tarea empieza cuando Lou tiene las manos libres (cuando la corrutina actual hace await). Como no espera a que termine y pasa a la línea siguiente, puedes dejar el pedido encargado y ponerte con otra cosa.

async def table(n, wait):
    print(f"T{n} order")
    await asyncio.sleep(wait)
    print(f"T{n} served")
    return f"T{n}"

async def main():
    start = time.perf_counter()
    t1 = asyncio.create_task(table(1, 0.3))
    t2 = asyncio.create_task(table(2, 0.2))
    print("Lou: 2 tasks created")
    await asyncio.sleep(0.1)          # Lou hace otra cosa un momento
    print("Lou: back", t1.done(), t2.done())
    print(await t1, await t2, f"{time.perf_counter() - start:.2f}s")

asyncio.run(main())
# Lou: 2 tasks created
# T1 order
# T2 order
# Lou: back False False
# T2 served
# T1 served
# T1 T2 0.31s

• "Lou: 2 tasks created" sale antes que T1 order. La tarea queda programada nada más crearla, pero se ejecuta por primera vez cuando main deja su sitio con un await.
• await t1 espera a que termine una tarea que ya está en marcha. Como las dos tareas esperaron a la vez, no tarda 0,5 segundos sino unos 0,3.
• gather envuelve por dentro en tareas las corrutinas que recibe. Si solo necesitas juntar los resultados, usa gather; si quieres guardarlas aparte para ver su estado o esperarlas más tarde, usa create_task.

Hay que guardar una referencia a la tarea

El bucle de eventos solo guarda las tareas mediante referencias débiles. Una tarea que no guardas en una variable o en una colección puede desaparecer con la recolección de basura antes de terminar. Aunque encargues algo cuyo resultado no necesitas, como rellenar el agua, tienes que quedarte con la referencia.

background = set()

async def refill_water(table):
    await asyncio.sleep(0.1)
    print(f"T{table} water")

async def main():
    for n in (1, 2, 3):
        task = asyncio.create_task(refill_water(n))
        background.add(task)                         # quedarse con la referencia
        task.add_done_callback(background.discard)   # al terminar, sale sola
    print("pending", len(background))
    await asyncio.sleep(0.2)
    print("pending", len(background))

asyncio.run(main())
# pending 3
# T1 water
# T2 water
# T3 water
# pending 0

Una línea que tira el resultado, como asyncio.create_task(refill_water(n)), normalmente funciona bien, pero el problema es ese «normalmente». Una tarea que desaparece de vez en cuando es el error más difícil de encontrar.

9-6. asyncio.sleep frente a time.sleep (3 min)

Los dos significan «esperar un momento», pero para Lou son cosas completamente distintas.

async def polite(n):
    await asyncio.sleep(0.3)   # mientras espera, cede el turno al bucle

async def stubborn(n):
    time.sleep(0.3)            # detiene el bucle entero

async def main(fn):
    start = time.perf_counter()
    await asyncio.gather(*(fn(n) for n in range(3)))
    print(fn.__name__, f"{time.perf_counter() - start:.2f}s")

asyncio.run(main(polite))
asyncio.run(main(stubborn))
# polite 0.32s
# stubborn 0.90s

• asyncio.sleep, en el momento del await, le dice al bucle «despiértame dentro de 0,3 segundos» y deja su sitio.
• time.sleep detiene el propio hilo. Como el bucle corre en un solo hilo, es como si Lou se quedara plantada delante de una mesa esperando. Mientras tanto, nadie atiende las demás mesas.
• Por eso las tres mesas fueron una tras otra, 0,3 segundos cada una: unos 0,9 segundos. Aunque uses gather, no sirve de nada.

No solo time.sleep: un cálculo largo o una biblioteca antigua que no cede el turno mientras espera también detienen el bucle. Ese problema y su solución son el tema de la lección 11.

9-7. Corrutinas, tareas, futuros y lo que cuestan (3 min)

En asyncio aparecen tres nombres parecidos.

print(issubclass(asyncio.Task, asyncio.Future))   # True

async def main():
    coro = take_order(1)
    task = asyncio.create_task(coro)
    print(type(coro).__name__, type(task).__name__)   # coroutine Task

La idea es la misma que la de concurrent.futures.Future de la lección 7. La diferencia es que los futuros de asyncio se pueden esperar con await y solo se usan dentro del mismo bucle. Rara vez tendrás que crear un futuro a mano; normalmente bastan corrutinas y tareas.

Las tareas son ligeras

Los hilos los crea el sistema operativo, así que cada uno lleva su memoria de pila y el coste de gestión del sistema. Una tarea no es más que un objeto de Python.

async def table(n):
    await asyncio.sleep(0.5)
    return n

async def main():
    return await asyncio.gather(*(table(n) for n in range(10_000)))

# 10,000 tasks: unos 0.57 s   (diez mil mesas que esperan 0,5 s)
# 1,000 threads: unos 0.62 s  (mil hilos que esperan 0,5 s)
# valores medidos en este PC

Lanzar diez mil tareas cuesta menos que mil hilos. Por eso asyncio encaja tan bien cuando hay miles o decenas de miles de esperas (un servidor de chat, enviar peticiones a muchas direcciones).

Cierre (10 min)

El objetivo de esta noche era sacar los platos de las diez mesas en menos de 0,5 segundos sin contratar a nadie. Basta con Lou, un bucle de eventos y un gather.

import asyncio
import time

async def table(n):
    await asyncio.sleep(0.3)   # esperar a que salga el plato de cocina
    return f"T{n}"

async def hall():
    return await asyncio.gather(*(table(n) for n in range(1, 11)))

start = time.perf_counter()
served = asyncio.run(hall())
print(served)
print(f"{time.perf_counter() - start:.2f}s")
# ['T1', 'T2', 'T3', 'T4', 'T5', 'T6', 'T7', 'T8', 'T9', 'T10']
# 0.30s

No hemos creado ni un solo hilo. Diez mesas de 0,3 segundos terminaron en unos 0,3 segundos.

Lou deja la bandeja vacía. «Diez mesas, en una sola vuelta.» La chef mira el temporizador y pregunta: «Bien. Pero mañana hay una reserva de grupo. Si de seis mesas se cancela una, ¿cómo vas a parar los platos de las otras cinco?»

Ejercicios

1. Ejecuta una tras otra una línea que llame al take_order de 9-1 sin await y otra que lo llame con asyncio.run(), y compara la salida y los avisos.
2. Si quitas el await asyncio.sleep(0.1) del main de 9-5, ¿cómo cambia el orden de la salida? Escribe primero tu predicción y luego compruébalo ejecutándolo.
3. Sube las mesas del código del cierre a 100 y a 1000, y mide el tiempo. Compáralo también con hacer lo mismo con 1000 hilos.