La sala se congela: llamadas bloqueantes y combinar herramientas

Once de la noche. La mesa 3 pide un gratinado. El gratinado solo se puede hacer en el horno viejo que la chef compró de segunda mano. Una vez cerrada la puerta, el asa se bloquea hasta que termina de hornear, y alguien tiene que quedarse de pie sujetándola.

Lou se pone delante del horno. En ese instante, la sala se congela. La mesa 1, que pide agua, y la mesa 5, que quiere pagar, se quedan esperando a que vuelva Lou.

El señor Tak chasquea la lengua. «Estar plantado delante del horno no es trabajar. Es estar plantado». La chef asiente. «Lou no puede esperar. Lo de esperar, que lo haga otro».

11-1. Si Lou se para, se para la sala (7 min)

El bucle de eventos de asyncio funciona en un solo hilo. Solo cuando una corrutina cede el paso en un await puede el bucle pasar a otra tarea. Pero si llamas a una función que no sabe nada de await, por ejemplo time.sleep o una vieja biblioteca síncrona, el bucle no puede moverse hasta que esa función termine. A estas llamadas se les llama llamadas bloqueantes.

import asyncio, time

start = time.perf_counter()

async def lu_rounds():                 # Lou da una vuelta a la sala cada 0.1 s
    for _ in range(5):
        print(f"round {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.1)

def old_oven():                        # una función vieja que no conoce await
    time.sleep(0.35)
    return "bread"

async def bake():
    await asyncio.sleep(0.15)
    print("oven", old_oven())          # aquí se para todo el bucle durante 0.35 s

async def main():
    await asyncio.gather(lu_rounds(), bake())

asyncio.run(main())
# round 0.0s
# round 0.1s
# oven bread
# round 0.5s
# round 0.6s
# round 0.7s

• Las vueltas de los 0,2, 0,3 y 0,4 segundos han desaparecido por completo. Entre los 0,15 y los 0,5 segundos, mientras Lou estaba delante del horno, no pudo avanzar ninguna tarea.
• Una llamada bloqueante no retrasa solo a su corrutina. Para todas las tareas del bucle a la vez. Esa es la gran diferencia con tener varios hilos.
• Los culpables habituales: time.sleep, bibliotecas de red síncronas como requests.get, leer archivos grandes, cálculos pesados.

En la línea de tiempo de abajo, una mesa lenta usa el horno viejo (time.sleep, 0,6 segundos). Las demás mesas esperan con asyncio.sleep, en trozos de 0,1 segundos.

• Con el interruptor apagado, tarda unos 0,9 segundos. T1 y T2 son de 0,3 segundos, pero se alargan hasta cerca de 0,76; T4–T6 no empiezan hasta los 0,6 segundos, cuando termina el horno. Lou ni siquiera había llegado aún a esas mesas.
• Si cambias la mesa lenta a la 1, las otras cinco no empiezan hasta pasados 0,6 segundos.
• Con el interruptor encendido, el horno pasa a un hilo: las demás mesas terminan hacia los 0,3 segundos y el total baja a unos 0,6, que es lo que tarda el horno.

11-2. Encontrar al culpable con el modo de depuración (6 min)

En un programa grande, cuesta saber qué corrutina tuvo secuestrado el bucle. El modo de depuración de asyncio avisa en el log de los callbacks y las tareas que se ejecutaron demasiado tiempo de un tirón.

import asyncio, logging, time

logging.basicConfig(level=logging.WARNING, format="%(name)s %(levelname)s: %(message)s")

def old_oven():
    time.sleep(0.3)

async def bake():
    old_oven()

async def main():
    await asyncio.create_task(bake(), name="T3-oven")

asyncio.run(main(), debug=True)
# p. ej.: asyncio WARNING: Executing <Task finished name='T3-oven' coro=<bake() done, defined at ...oven.py:8>
#     result=None created at ...\asyncio\tasks.py:395> took 0.301 seconds

• Se activa con asyncio.run(main(), debug=True). También con la variable de entorno PYTHONASYNCIODEBUG=1 o con python -X dev.
• En una sola línea de aviso aparecen el nombre de la tarea (T3-oven), el archivo y la línea donde está definida la corrutina, y el tiempo que tuvo el bucle (took 0.301 seconds). Si pones name= a tus tareas, encontrar al culpable es mucho más fácil.
• El mensaje sale como WARNING del logger asyncio. Las rutas cambian de un PC a otro.

El umbral es loop.slow_callback_duration, que por defecto vale 0,1 segundos. Si quieres que salte antes, bájalo.

import asyncio, time

async def main():
    loop = asyncio.get_running_loop()
    print(loop.get_debug(), loop.slow_callback_duration)
    loop.slow_callback_duration = 0.05     # avísame si pasa de 0.05 s
    time.sleep(0.07)

asyncio.run(main(), debug=True)
# True 0.1
# p. ej.: Executing <Task finished name='Task-1' coro=<main() done, defined at ...> result=None created at ...> took 0.070 seconds

Además, el modo de depuración te dice dónde se creó una corrutina que se abandonó sin await, y da error si otro hilo toca el bucle de forma indebida. Actívalo mientras desarrollas y desactívalo en producción (lo hace todo más lento).

11-3. Lo de esperar, para un hilo: to_thread y run_in_executor (8 min)

Si no puedes arreglar el horno viejo, pon a otra persona delante de él. asyncio.to_thread() (3.9) ejecuta una función bloqueante en otro hilo y le devuelve al bucle un objeto al que se puede esperar.

import asyncio, time

start = time.perf_counter()

async def lu_rounds():
    for _ in range(5):
        print(f"round {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.1)

def old_oven(item, *, minutes):
    time.sleep(0.35)
    return f"{item} {minutes}m"

async def bake():
    await asyncio.sleep(0.15)
    bread = await asyncio.to_thread(old_oven, "bread", minutes=40)
    print("oven", bread)

async def main():
    await asyncio.gather(lu_rounds(), bake())

asyncio.run(main())
# round 0.0s
# round 0.1s
# round 0.2s
# round 0.3s
# round 0.4s
# oven bread 40m

• Lou recorre la sala cada 0,1 segundos sin saltarse ninguna vuelta. Del horno se ocupa un hilo, y Lou solo recoge el resultado con await.
• to_thread(función, *args, **kwargs) pasa los argumentos tal cual. Se ejecuta en el grupo de hilos de la lección 7 (el executor por defecto del bucle).
• Como es un hilo, siguen valiendo las reglas del GIL (lección 8). Funciona muy bien con código bloqueante que espera, pero no acelera el trabajo de cálculo pesado.

El trabajo de cálculo pesado se manda a la cocina sucursal, es decir, a un grupo de procesos. Obtienes el bucle en marcha con asyncio.get_running_loop() y usas loop.run_in_executor(executor, función, *args). Como usa procesos, guárdalo en un archivo y ejecútalo.

import asyncio
from concurrent.futures import ProcessPoolExecutor

def knead(n):                          # trabajo de manos (CPU)
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=2) as pool:
        doughs = await asyncio.gather(
            loop.run_in_executor(pool, knead, 2_000_000),
            loop.run_in_executor(pool, knead, 3_000_000),
        )
    print(doughs)

if __name__ == "__main__":
    asyncio.run(main())
# [2666664666667000000, 8999995500000500000]

• Si pasas None como primer argumento de run_in_executor, usa el grupo de hilos por defecto. to_thread es justo la función que hace eso más cómodo.
• run_in_executor no acepta argumentos con nombre. Si los necesitas, envuélvelo con functools.partial(knead, n=...).
• Al ser un grupo de procesos, se aplican las mismas reglas de la lección 8: hace falta if __name__ == "__main__":, y la función y los valores que pasas tienen que poder serializarse con pickle.

11-4. Las herramientas de sincronización de asyncio (8 min)

Las herramientas de las lecciones 4 y 6 tienen su propia versión en asyncio. Se llaman igual, pero al esperar, en lugar de parar el hilo, ceden el paso con await.

El semáforo se usa sobre todo para «limitar las conexiones simultáneas». Hay que hacer cinco llamadas a un proveedor, pero el proveedor solo atiende dos a la vez.

import asyncio, time

start = time.perf_counter()
supplier = asyncio.Semaphore(2)        # al proveedor, como mucho dos llamadas a la vez

async def call(n):
    async with supplier:
        print(f"call #{n} at {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.2)

async def main():
    await asyncio.gather(*(call(n) for n in range(1, 6)))

asyncio.run(main())
# call #1 at 0.0s
# call #2 at 0.0s
# call #3 at 0.2s
# call #4 at 0.2s
# call #5 at 0.4s

asyncio.Queue tiene la misma forma que el riel de comandas de la lección 6. También tiene maxsize, task_done() y join().

import asyncio

async def hall(q):
    for n in range(1, 6):
        await q.put(f"#{n}")           # si el riel está lleno, espera aquí
    print("hall done")

async def cook(name, q, done):
    while True:
        ticket = await q.get()
        await asyncio.sleep(0.1)
        done.append((name, ticket))
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=2)
    done = []
    cooks = [asyncio.create_task(cook(c, q, done)) for c in ("A", "B")]
    await hall(q)
    await q.join()                     # hasta que todas las comandas colgadas tengan su task_done
    for c in cooks:
        c.cancel()
    print(len(done), sorted(t for _, t in done))

asyncio.run(main())
# hall done
# 5 ['#1', '#2', '#3', '#4', '#5']

Como el bucle es un solo hilo, parece que no puede haber condiciones de carrera, pero si hay un await entre la lectura y la escritura, se repite tal cual el accidente de la pizarra de la lección 3.

import asyncio

async def main():
    board = {"count": 0}

    async def cook():
        for _ in range(100):
            v = board["count"]
            await asyncio.sleep(0)     # aquí se pasa a otra tarea
            board["count"] = v + 1

    await asyncio.gather(*(cook() for _ in range(4)))
    print(board["count"])

asyncio.run(main())
# 100

La pizarra debería decir 400 y dice 100. Las cuatro tareas leen el mismo valor, ceden el paso y escriben el mismo valor. Si encierras ese tramo con asyncio.Lock y haces que esperen la señal de «¡abrimos!» con asyncio.Event, queda así.

import asyncio

async def main():
    lock = asyncio.Lock()
    opened = asyncio.Event()
    board = {"count": 0}

    async def cook():
        await opened.wait()            # espera al «¡abrimos!»
        for _ in range(100):
            async with lock:           # si hay un await entre leer y escribir, hace falta un cerrojo
                v = board["count"]
                await asyncio.sleep(0)
                board["count"] = v + 1

    tasks = [asyncio.create_task(cook()) for _ in range(4)]
    await asyncio.sleep(0.1)
    opened.set()
    await asyncio.gather(*tasks)
    print(board["count"])

asyncio.run(main())
# 400

No mezcles las herramientas de los dos mundos. Si una corrutina espera un threading.Lock, se para el bucle entero. Y la tarea que tiene el cerrojo pierde la ocasión de volver a soltarlo.

import asyncio, threading, time

lock = threading.Lock()                # ¿y si usas en una corrutina un cerrojo para hilos?
start = time.perf_counter()

async def cook(n):
    if not lock.acquire(timeout=0.2):  # mientras espera, se para todo el bucle
        print(f"#{n} gave up at {time.perf_counter() - start:.1f}s")
        return
    try:
        await asyncio.sleep(0.1)       # cede el paso con el cerrojo cerrado
        print(f"#{n} done")
    finally:
        lock.release()

async def main():
    await asyncio.gather(cook(1), cook(2))

asyncio.run(main())
# #2 gave up at 0.2s
# #1 done

Sin el timeout, #2 esperaría para siempre y #1 no volvería a despertar nunca. Y al revés: las herramientas de asyncio no son seguras entre hilos. No metas cosas en un asyncio.Queue desde otro hilo. Cómo pasarle trabajo al bucle desde un hilo lo veremos en 11-6.

11-5. async with, async for y generadores asíncronos (7 min)

with y for también tienen su versión que espera.

• async with: recursos que necesitan un await al entrar y al salir (__aenter__, __aexit__). Cerrojos, semáforos, conexiones de red.
• async for: bucles en los que hay que esperar para recibir el siguiente valor (__aiter__, __anext__). Comandas que llegan de una en una, respuestas en streaming.
• Generadores asíncronos: si pones yield dentro de un async def, obtienes un iterador que se recorre con async for.

import asyncio
from contextlib import asynccontextmanager

async def tickets(n):                  # generador asíncrono: async def + yield
    for i in range(1, n + 1):
        await asyncio.sleep(0.05)      # espera a que llegue la siguiente comanda
        yield f"#{i}"

@asynccontextmanager
async def stove(name):                 # un fogón que se usa con async with
    await asyncio.sleep(0.01)
    print(f"{name} on")
    try:
        yield name
    finally:
        print(f"{name} off")

async def main():
    async with stove("S1"):
        async for t in tickets(3):
            print("cook", t)
    print([t async for t in tickets(2)])

asyncio.run(main())
# S1 on
# cook #1
# cook #2
# cook #3
# S1 off
# ['#1', '#2']

contextlib.asynccontextmanager te crea un recurso para async with con un solo yield. También puedes usar async for en una comprensión de listas (solo dentro de una corrutina).

El as_completed de la lección 7 también tiene versión asyncio. Desde 3.13 se puede recorrer con async for, y entonces devuelve, en el orden en que terminan, exactamente las mismas tareas que le pasaste.

import asyncio

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

async def main():
    tasks = [asyncio.create_task(dish(n, s)) for n, s in [(1, 0.3), (2, 0.1), (3, 0.2)]]
    async for t in asyncio.as_completed(tasks):      # desde 3.13
        print(t in tasks, await t)

asyncio.run(main())
# True #2
# True #3
# True #1

Puedes seguir escribiéndolo a la antigua, for f in asyncio.as_completed(tasks): await f, pero entonces el f que recibes no es la tarea original, sino un objeto nuevo. Si quieres saber qué pedido ha terminado, async for es más cómodo.

11-6. Pasar trabajo desde fuera del bucle y ver las tareas en marcha (4 min)

Supongamos que la persona que atiende el teléfono de reservas trabaja en otro hilo. Para que ese hilo meta trabajo en la agenda de Lou, se usa asyncio.run_coroutine_threadsafe(corrutina, loop). Lo que devuelve es el concurrent.futures.Future de la lección 7, así que desde el hilo puedes esperarlo con result(timeout=).

import asyncio

async def seat(table):
    await asyncio.sleep(0.05)
    return f"{table} seated"

def phone_line(loop):                  # un hilo fuera del bucle (el teléfono de reservas)
    fut = asyncio.run_coroutine_threadsafe(seat("T7"), loop)
    print(fut.result(timeout=1))       # concurrent.futures.Future

async def main():
    loop = asyncio.get_running_loop()
    await asyncio.to_thread(phone_line, loop)

asyncio.run(main())
# T7 seated

Tareas eager (3.12): normalmente, create_task solo pone la tarea en la agenda y no la ejecuta hasta que le llega el turno. Con asyncio.eager_task_factory, se ejecuta nada más crearla, hasta su primer await. Si es una corrutina que termina sin ningún await, ni siquiera entra en la agenda: termina ahí mismo.

import asyncio

async def greet(n):
    print(f"greet T{n}")               # hasta el primer await
    await asyncio.sleep(0)
    return n

async def main(eager):
    if eager:
        asyncio.get_running_loop().set_task_factory(asyncio.eager_task_factory)
    t = asyncio.create_task(greet(1))
    print("created, done =", t.done())
    await t

asyncio.run(main(False))
asyncio.run(main(True))
# created, done = False
# greet T1
# greet T1
# created, done = False

Ver las tareas en marcha (3.14): si conoces el número de proceso (PID) de un programa asyncio en ejecución, desde otra terminal puedes ver la lista de tareas con python -m asyncio ps <PID>, y con python -m asyncio pstree <PID>, en forma de árbol, quién espera a quién. Por ejemplo, si la tarea hall está esperando a T1 y T2 mediante un TaskGroup, sale algo así (con algunas columnas y rutas omitidas).

task name   coroutine stack                                   awaiter name
Task-1      main
hall        TaskGroup._aexit -> TaskGroup.__aexit__ -> hall   Task-1
T1          sleep -> table                                    hall
T2          sleep -> table                                    hall

Es la herramienta para el momento de «la sala se ha congelado y no sé dónde está plantada Lou». Si pones name= a tus tareas, la lista se lee mucho mejor.

Cierre (10 min)

Solo el gratinado de la mesa 3 va al horno viejo, y delante del horno ponemos un hilo. Lo comparamos con un único interruptor (fix).

import asyncio, time

def old_oven(sec):                     # una biblioteca vieja que no podemos tocar
    time.sleep(sec)
    return "gratin"

async def table(n, start, fix):
    if n == 3:
        dish = await asyncio.to_thread(old_oven, 0.6) if fix else old_oven(0.6)
    else:
        await asyncio.sleep(0.3)
        dish = "noodle"
    return f"T{n} {dish} {time.perf_counter() - start:.1f}s"

async def service(fix):
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(table(n, start, fix)) for n in range(1, 7)]
    print([t.result() for t in tasks])

asyncio.run(service(fix=False))
asyncio.run(service(fix=True))
# ['T1 noodle 0.6s', 'T2 noodle 0.6s', 'T3 gratin 0.6s', 'T4 noodle 0.9s', 'T5 noodle 0.9s', 'T6 noodle 0.9s']
# ['T1 noodle 0.3s', 'T2 noodle 0.3s', 'T3 gratin 0.6s', 'T4 noodle 0.3s', 'T5 noodle 0.3s', 'T6 noodle 0.3s']

• Si Lou se planta ella misma delante del horno, ni uno solo de los fideos de 0,3 segundos sale a los 0,3 segundos. T1 y T2 ya tenían la olla al fuego antes del horno, pero no se sirven hasta los 0,6 segundos; T4–T6 no ponen la olla al fuego hasta que termina el horno, y salen a los 0,9. Es exactamente lo que viste en la línea de tiempo.
• Con to_thread, los fideos salen a los 0,3 segundos y el gratinado a los 0,6, lo que tarda el horno. Objetivo de esta noche cumplido.

Después del cierre, la chef va señalando con el dedo, una a una, las tarjetas de recetas de la pared de la cocina. Hilos, cerrojos, colas, grupos, procesos, el bucle, TaskGroup, to_thread. «Ya tenemos todas estas herramientas. Pero cuando un cliente abre la puerta y entra, ¿cómo decides cuál sacar primero?».

Ejercicios

1. En la línea de tiempo, cambia la mesa lenta de la 1 a la 6. Explica en qué se diferencian las barras de las mesas con número menor y con número mayor que la lenta, y por qué.
2. Ejecuta el primer ejemplo de 11-1 con asyncio.run(main(), debug=True) y comprueba qué tarea dice el aviso que tuvo el bucle, y durante cuántos segundos.
3. Si cambias el ejemplo del semáforo de 11-4 a asyncio.Semaphore(3), calcula primero en qué segundo empezará call #5, y luego ejecútalo.