Atlas

Python Asynchronous Programming: asyncio, event loop

asyncio es concurrencia cooperativa de un solo hilo: un event loop corre muchas coroutines que ceden el control voluntariamente en cada await, y mientras una espera I/O el loop hace avanzar a otra. No hay paralelismo ni preempción — hay interleaving determinista gobernado por los puntos de await. Dominar async es entender exactamente dónde tu código cede el control, qué pasa cuando no lo cede (bloqueas el loop), y cuándo el problema real no era I/O-bound sino CPU-bound (y entonces async es la herramienta equivocada).

Teoría

Qué es async y cuándo usarlo (I/O-bound)

Una tarea es I/O-bound cuando pasa la mayoría de su tiempo esperando algo externo: una respuesta HTTP, una query a la base de datos, una lectura de disco, un mensaje de una cola. Durante esa espera el CPU está ocioso. Async explota exactamente ese hueco: mientras una coroutine espera la respuesta de la red, el event loop corre otra. El resultado es que miles de operaciones I/O “en vuelo” se manejan con un solo hilo y un solo proceso, sin el overhead de memoria de miles de threads.

Casos típicos donde async brilla:

Cuándo no usar async: trabajo CPU-bound (hashing, compresión, parseo pesado, cómputo numérico). Ahí no hay espera que aprovechar; el trabajo es el CPU ocupado, y una coroutine que calcula sin await bloquea el loop entero. Eso va a multiprocessing o a un ProcessPoolExecutor.

async vs parallel (concurrencia ≠ paralelismo)

Son ejes distintos y confundirlos es el error conceptual #1 en entrevistas.

asyncio da concurrencia sin paralelismo: un solo hilo, un solo core, el GIL ni compite. Nunca hay dos líneas de bytecode Python corriendo simultáneamente. Lo que ganas es que el tiempo total tiende al de la operación más lenta, no a la suma, porque las esperas se solapan.

import asyncio, time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)   # simula I/O: aquí se cede el control
    return name

async def sequential() -> None:
    t0 = time.perf_counter()
    a = await fetch("a", 1)      # espera 1s...
    b = await fetch("b", 1)      # ...LUEGO espera otro 1s
    print("sequential", time.perf_counter() - t0)   # ~2.0s

async def concurrent() -> None:
    t0 = time.perf_counter()
    a, b = await asyncio.gather(fetch("a", 1), fetch("b", 1))
    print("concurrent", time.perf_counter() - t0)   # ~1.0s

sequential no es “lento por ser async” — es lento porque await uno-por-uno serializa las esperas. La concurrencia solo aparece cuando lanzas varias coroutines antes de esperar sus resultados (gather, create_task, TaskGroup).

El event loop: cómo funciona

El event loop es un bucle infinito que administra la ejecución cooperativa. En cada iteración (“tick”):

  1. Ejecuta los callbacks listos (los que call_soon encoló) y las tasks que ya pueden avanzar, corriendo cada coroutine hasta su siguiente await.
  2. Cuando una coroutine hace await sobre algo que aún no está listo (I/O pendiente, timer), suspende esa coroutine y registra una continuación: “cuando este socket sea legible / cuando venza este timer, reanúdame”.
  3. Si no queda nada listo, el loop duerme en un syscall de multiplexado de I/O (select/poll/epoll según el OS), que despierta cuando algún file descriptor está listo o vence el timer más próximo. Esto es clave: el loop no hace busy-wait, cede el CPU al kernel.
  4. Al despertar, marca como listas las coroutines cuyo I/O completó y vuelve al paso 1.

La propiedad esencial es que es cooperativo, no preemptivo: el loop solo recupera el control en los await. No hay un timer que interrumpa una coroutine a la fuerza (como sí hace el OS scheduler con threads). Corolario directo: cualquier código síncrono que no ceda —un time.sleep, un loop de cómputo, un driver de DB bloqueante— congela TODAS las demás coroutines hasta que retorne.

import asyncio

async def main() -> None:
    print("start")
    await asyncio.sleep(0)   # yield al loop sin dormir: da un tick a otras tasks
    print("resumed")

asyncio.run(main())   # crea el loop, corre main() hasta completar, cierra el loop

asyncio.run() es el entrypoint de alto nivel: crea un event loop nuevo, corre la coroutine hasta completarla, cancela tasks pendientes y cierra el loop. No debes anidar asyncio.run() dentro de otro (ya hay un loop corriendo) — desde dentro de una coroutine se usa await o create_task.

async / await y coroutines

async def define una coroutine function. Llamarla no ejecuta nada: devuelve un objeto coroutine “frío” que solo avanza cuando el loop lo mueve (vía await, create_task o gather).

async def greet() -> str:
    return "hi"

c = greet()          # NO imprime, NO corre: c es un coroutine object
# print(c)           -> <coroutine object greet at 0x...>
# olvidarlo aquí     -> RuntimeWarning: coroutine 'greet' was never awaited
result = await greet()   # ahora sí corre y devuelve "hi"  (dentro de otra coroutine)

await X significa “suspende esta coroutine hasta que X produzca su resultado, cediendo el control al loop mientras tanto”. X debe ser un awaitable: una coroutine, una Task, un Future, o un objeto con __await__. await es el único punto donde una coroutine puede ceder — entre dos await el código es efectivamente atómico respecto al loop (no hay context switch en medio), lo que elimina buena parte de las race conditions clásicas de threads.

Coroutines, Tasks y Futures

Tres niveles de abstracción, fáciles de confundir:

import asyncio

async def work(n: int) -> int:
    await asyncio.sleep(0.1)
    return n * 2

async def main() -> None:
    # create_task AGENDA la coroutine ya: empieza a correr en el próximo tick,
    # sin necesidad de await inmediato.
    t1 = asyncio.create_task(work(10))
    t2 = asyncio.create_task(work(20))
    # ...aquí ambas ya están progresando concurrentemente...
    r1 = await t1     # espero su resultado
    r2 = await t2
    print(r1, r2)     # 20 40

asyncio.run(main())

Diferencia práctica: await coro corre la coroutine ahora y en línea (no hay concurrencia con lo que sigue). create_task(coro) la lanza en background y sigues; concurrencia real aparece cuando hay varias tasks vivas antes de esperarlas.

Trap de fire-and-forget: el loop guarda solo una referencia débil a las tasks. Si creas una task y no guardas la referencia, el garbage collector puede recolectarla a mitad de camino y la task “desaparece” silenciosamente. Guarda las tasks en un set y remuévelas en un done_callback:

_bg: set[asyncio.Task] = set()

def spawn(coro) -> None:
    t = asyncio.create_task(coro)
    _bg.add(t)
    t.add_done_callback(_bg.discard)

Obtener y controlar el loop (API de bajo nivel)

Dentro de una coroutine, la forma correcta de obtener el loop es asyncio.get_running_loop() (falla explícito si no hay loop corriendo). Evita asyncio.get_event_loop() en código nuevo: su semántica sin loop activo está deprecada.

Programar callbacks (no coroutines, funciones normales) en el loop:

import asyncio

def ping(msg: str) -> None:
    print("callback:", msg)

async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.call_soon(ping, "now")
    loop.call_later(0.5, ping, "in 0.5s")
    await asyncio.sleep(1)   # dejar tiempo a que corran los callbacks

asyncio.run(main())

Correr/parar el loop manualmente (útil para entender qué hace asyncio.run por dentro, o para embeber asyncio en frameworks que gestionan su propio loop):

import asyncio

async def main() -> None:
    await asyncio.sleep(0.1)
    return "done"

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
    result = loop.run_until_complete(main())   # corre hasta que main() termina
finally:
    loop.close()

loop.stop() detiene un loop.run_forever() tras el tick actual; se usa con loops de larga vida manejados a mano. En aplicaciones normales no toques esto — asyncio.run() (o el runner de tu framework: uvicorn, etc.) lo maneja por ti.

Librerías async: el ecosistema aio*

asyncio solo aporta el motor; para no bloquear el loop necesitas drivers async nativos en cada borde de I/O. Usar un cliente síncrono (requests, psycopg2 clásico, time.sleep) dentro de una coroutine bloquea el loop y anula toda la ventaja.

import aiohttp, asyncio

async def fetch_json(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as resp:      # async context manager (ver abajo)
        resp.raise_for_status()
        return await resp.json()               # await: el body llega por red

async def main() -> None:
    async with aiohttp.ClientSession() as session:
        data = await fetch_json(session, "https://example.com/api")
        print(data)

asyncio.run(main())

Chaining/fan-out: gather

asyncio.gather(*aws) corre múltiples awaitables concurrentemente y devuelve sus resultados en el orden de los argumentos (no en orden de finalización). Es la herramienta de fan-out por excelencia.

import asyncio

async def get_user(uid): ...
async def get_orders(uid): ...
async def get_prefs(uid): ...

async def profile(uid: int) -> dict:
    user, orders, prefs = await asyncio.gather(
        get_user(uid), get_orders(uid), get_prefs(uid),
    )
    return {"user": user, "orders": orders, "prefs": prefs}

Semántica de errores de gather, que es donde está la trampa senior:

Por eso en Python 3.11+ se prefiere asyncio.TaskGroup para fan-out con structured concurrency: si una task falla, el grupo cancela todas las demás y re-lanza al salir del async with. No hay trabajo huérfano ni fallo parcial silencioso.

import asyncio

async def profile(uid: int) -> dict:
    async with asyncio.TaskGroup() as tg:          # 3.11+
        t_user   = tg.create_task(get_user(uid))
        t_orders = tg.create_task(get_orders(uid))
        t_prefs  = tg.create_task(get_prefs(uid))
    # al salir del 'async with' todas terminaron; si alguna falló,
    # el resto fue cancelado y aquí se re-lanza (como ExceptionGroup).
    return {"user": t_user.result(), "orders": t_orders.result(), "prefs": t_prefs.result()}

Regla práctica: gather cuando el fallo parcial es aceptable (o con return_exceptions=True y manejo explícito); TaskGroup cuando “todo o nada” y no quieres tasks huérfanas.

Timeouts y cancellation

En sistemas de producción toda espera de I/O debe tener timeout. Sin él, un upstream colgado deja la coroutine —y el recurso que sostiene— pendiente para siempre.

import asyncio

async def main() -> None:
    try:
        # wait_for: si excede el timeout, CANCELA la coroutine interna y
        # lanza TimeoutError.
        result = await asyncio.wait_for(slow_call(), timeout=2.0)
    except asyncio.TimeoutError:
        result = fallback()

    # 3.11+: asyncio.timeout() como context manager, envuelve un bloque entero.
    try:
        async with asyncio.timeout(2.0):
            a = await step_one()
            b = await step_two()   # el timeout cubre TODO el bloque
    except asyncio.TimeoutError:
        ...

Cancellation es el mecanismo por el que asyncio detiene una task en curso. task.cancel() programa que, en el próximo await de esa task, se lance asyncio.CancelledError dentro de la coroutine. Puntos clave, muy preguntados:

import asyncio

async def worker() -> None:
    try:
        while True:
            await do_chunk()
    except asyncio.CancelledError:
        await flush_partial()      # cleanup permitido...
        raise                      # ...pero SIEMPRE re-lanzar
    finally:
        await release_resources()  # cleanup garantizado

Para proteger una sección de la cancelación (raro, pero existe): asyncio.shield(coro). Para esperar a un set de tasks con política: asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED).

Async generators y async for

Un async generator es una async def que usa yield y además puede await entre yields. Sirve para producir un stream cuyos elementos llegan de forma asíncrona (paginación de una API, lectura de una cola, streaming de filas de la DB) sin materializar todo en memoria.

import asyncio

async def paginate(client, url: str):
    next_url = url
    while next_url:
        page = await client.get_json(next_url)   # await dentro del generator
        for item in page["items"]:
            yield item                            # emite item a item
        next_url = page.get("next")

async def main(client) -> None:
    async for item in paginate(client, "/api/v1/things"):   # consumo con async for
        process(item)

async for es la contraparte de consumo: en cada iteración hace await sobre __anext__() del async iterator, cediendo al loop mientras el próximo elemento se produce. Existe también la async comprehension: [x async for x in agen()].

Async context managers y async with

Un async context manager implementa __aenter__/__aexit__ (versiones awaitables de __enter__/__exit__) y se usa con async with. Necesario cuando la adquisición o liberación del recurso implican I/O: abrir/cerrar una sesión HTTP, tomar una conexión de un pool, adquirir un asyncio.Lock, iniciar/commit-ear una transacción.

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def db_transaction(pool):
    conn = await pool.acquire()          # I/O al adquirir
    tx = conn.transaction()
    await tx.start()
    try:
        yield conn
    except Exception:
        await tx.rollback()              # I/O al liberar
        raise
    else:
        await tx.commit()
    finally:
        await pool.release(conn)

async def main(pool) -> None:
    async with db_transaction(pool) as conn:
        await conn.execute("INSERT ...")

Async Queue: patrón producer/consumer

asyncio.Queue coordina tasks dentro del mismo loop sin locks manuales (recuerda: single-thread, no hay race entre awaits). Es la base del patrón worker-pool async: productores encolan trabajo, un número acotado de consumidores lo drena, y el backpressure sale gratis si le pones maxsize.

import asyncio

async def producer(q: asyncio.Queue, items: list[int]) -> None:
    for it in items:
        await q.put(it)          # bloquea (cede) si la cola está llena -> backpressure

async def consumer(q: asyncio.Queue, name: str) -> None:
    while True:
        item = await q.get()
        try:
            await handle(item)
        finally:
            q.task_done()        # marca el item como procesado

async def main() -> None:
    q: asyncio.Queue = asyncio.Queue(maxsize=100)
    workers = [asyncio.create_task(consumer(q, f"w{i}")) for i in range(5)]
    await producer(q, list(range(1000)))
    await q.join()               # espera a que todos los items se procesen
    for w in workers:            # los consumidores corren en while True: cancelarlos
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

Cuándo async, cuándo threads, cuándo processes

El árbol de decisión senior parte de la naturaleza del trabajo:

SituaciónHerramientaPor qué
I/O-bound + drivers async disponiblesasynciomáxima concurrencia, un hilo, mínimo overhead
I/O-bound pero la librería es síncrona/bloqueanteThreadPoolExecutor (o run_in_executor)el GIL se libera durante el syscall de I/O, así que threads sí escalan aquí
CPU-bound (cómputo puro)ProcessPoolExecutor / multiprocessingcada proceso tiene su propio GIL → paralelismo real en varios cores

concurrent.futures da una API uniforme (Executor) para las dos últimas: ThreadPoolExecutor y ProcessPoolExecutor comparten interfaz (submit, map), así que cambiar de threads a procesos es cambiar una línea. Y asyncio se integra con ellos vía run_in_executor, que es el puente para no bloquear el loop con código legacy:

import asyncio, functools
from concurrent.futures import ProcessPoolExecutor

def heavy_cpu(data: bytes) -> str:      # función SÍNCRONA, CPU-bound
    return expensive_hash(data)

async def handler(data: bytes) -> str:
    loop = asyncio.get_running_loop()
    # offload al pool de PROCESOS: no bloquea el loop, corre en otro core.
    return await loop.run_in_executor(
        _process_pool, functools.partial(heavy_cpu, data)
    )

_process_pool = ProcessPoolExecutor(max_workers=4)

Con None como primer argumento, run_in_executor usa el ThreadPoolExecutor por defecto del loop — perfecto para envolver una librería síncrona de I/O (un requests.get, un driver de DB legacy) sin bloquear. Para CPU-bound, pásale explícitamente un ProcessPoolExecutor.

Custom event loops: uvloop

El event loop por defecto de CPython está escrito en Python. uvloop es una implementación alternativa basada en libuv (el motor de Node.js), escrita en Cython, que suele dar 2–4x más throughput en cargas de red. Es un drop-in: no cambias tu código async, solo la policy del loop.

import asyncio
import uvloop

# Forma clásica (instala uvloop como loop por defecto):
uvloop.install()          # antes de asyncio.run(...)
asyncio.run(main())

# Python 3.12+ / uvloop reciente, con runner explícito:
# uvloop.run(main())

En producción, servidores como uvicorn ya seleccionan uvloop automáticamente si está instalado (--loop uvloop). No está disponible en Windows (libuv/uvloop apuntan a plataformas Unix); ahí se cae al loop estándar o a ProactorEventLoop. La lección conceptual: el event loop es pluggable vía asyncio.AbstractEventLoop / event loop policy, y uvloop demuestra que la semántica async del lenguaje es independiente de la implementación del motor.

Errores comunes (los que se ven en code review)

Ejercicios

Ejercicio 1 — Endpoint I/O-bound con fan-out concurrente

Tienes un endpoint que arma el dashboard de un usuario consultando 3 servicios upstream (fetch_profile, fetch_stats, fetch_notifications), cada uno ~200 ms. La versión actual los espera en serie (~600 ms). Reescríbela para que corra en ~200 ms, y haz que si cualquiera falla se cancele el resto y se propague el error (nada de dashboards a medias).

Solución 
import asyncio

async def build_dashboard(uid: int) -> dict:
    async with asyncio.TaskGroup() as tg:      # 3.11+: structured concurrency
        t_profile = tg.create_task(fetch_profile(uid))
        t_stats   = tg.create_task(fetch_stats(uid))
        t_notifs  = tg.create_task(fetch_notifications(uid))
    # Al salir del 'async with' las 3 terminaron. Si una lanzó, el grupo canceló
    # las otras y re-lanza aquí (ExceptionGroup). Nunca hay resultado parcial.
    return {
        "profile": t_profile.result(),
        "stats":   t_stats.result(),
        "notifs":  t_notifs.result(),
    }

# Variante pre-3.11 con gather. OJO: return_exceptions=False NO cancela las otras
# si una falla. Para "todo o nada" real hay que cancelarlas a mano:
async def build_dashboard_gather(uid: int) -> dict:
    tasks = [
        asyncio.create_task(fetch_profile(uid)),
        asyncio.create_task(fetch_stats(uid)),
        asyncio.create_task(fetch_notifications(uid)),
    ]
    try:
        profile, stats, notifs = await asyncio.gather(*tasks)
    except Exception:
        for t in tasks:
            t.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)  # drenar cancelaciones
        raise
    return {"profile": profile, "stats": stats, "notifs": notifs}

Clave: el tiempo total pasa de suma (~600 ms) al máximo (~200 ms) porque las 3 esperas de red se solapan. TaskGroup da el “todo o nada” sin código extra; con gather hay que cancelar manualmente.

Ejercicio 2 — Timeout + cancelación con cleanup correcto

Escribe charge_with_timeout(gateway, order_id, timeout=3.0) que llame a un gateway de pago async. Si excede el timeout, debe cancelar la llamada, hacer rollback de la reserva y lanzar un error de dominio. La coroutine del gateway debe hacer su propio cleanup al ser cancelada sin volverse no-cancelable.

Solución 
import asyncio

class PaymentTimeout(Exception): ...

async def call_gateway(gateway, order_id: str) -> dict:
    try:
        return await gateway.charge(order_id)
    except asyncio.CancelledError:
        await gateway.abort(order_id)   # cleanup del recurso remoto...
        raise                           # ...pero SIEMPRE re-lanzar CancelledError

async def charge_with_timeout(gateway, order_id: str, timeout: float = 3.0) -> dict:
    try:
        async with asyncio.timeout(timeout):     # 3.11+
            return await call_gateway(gateway, order_id)
    except asyncio.TimeoutError as err:
        await rollback_reservation(order_id)
        raise PaymentTimeout(f"gateway timed out for {order_id}") from err

Puntos evaluados: (1) asyncio.timeout cancela la coroutine interna al vencer; (2) call_gateway captura CancelledError solo para hacer cleanup y re-lanza (si no, la task queda no-cancelable y el timeout no surtiría efecto); (3) el error de timeout se traduce a excepción de dominio con from para preservar el root cause. Nota: except Exception NO habría atrapado CancelledError (hereda de BaseException), que es justo lo que queremos.

Ejercicio 3 — Detectar dónde se bloquea el loop

Este handler “async” tiene p99 de latencia horrible bajo carga y CPU al 100% en un solo core. Encuentra las tres cosas que bloquean el event loop y arréglalas.

import time, requests, hashlib
import asyncio

async def handle(url: str, payload: bytes) -> dict:
    time.sleep(0.1)                                  # (1)
    resp = requests.get(url)                         # (2)
    digest = hashlib.pbkdf2_hmac("sha256", payload, b"salt", 1_000_000)  # (3)
    return {"status": resp.status_code, "digest": digest.hex()}
Solución

Las tres líneas son síncronas y bloquean el único hilo del loop, congelando todas las demás coroutines mientras corren:

  1. time.sleep(0.1) → bloqueo puro. Fix: await asyncio.sleep(0.1) (o eliminarlo).
  2. requests.get(url) → cliente HTTP síncrono. Fix: un driver async (aiohttp/httpx async).
  3. hashlib.pbkdf2_hmac(..., 1_000_000) → CPU-bound puro (1M iteraciones). No hay I/O que aprovechar: await no ayuda. Fix: offload a un ProcessPoolExecutor.
import asyncio, functools, hashlib
import aiohttp
from concurrent.futures import ProcessPoolExecutor

_cpu_pool = ProcessPoolExecutor(max_workers=4)

def _derive(payload: bytes) -> bytes:                # síncrona, CPU-bound
    return hashlib.pbkdf2_hmac("sha256", payload, b"salt", 1_000_000)

async def handle(session: aiohttp.ClientSession, url: str, payload: bytes) -> dict:
    await asyncio.sleep(0.1)                          # (1) no bloquea
    loop = asyncio.get_running_loop()
    async with session.get(url) as resp:             # (2) I/O async real
        status = resp.status
    digest = await loop.run_in_executor(             # (3) CPU-bound a otro proceso
        _cpu_pool, functools.partial(_derive, payload)
    )
    return {"status": status, "digest": digest.hex()}

El insight: (2) es I/O-bound → driver async; (3) es CPU-bound → proceso aparte. Confundir ambos (meter el hash en un ThreadPool) no ayudaría, porque el GIL serializa el cómputo Python entre threads.

Ejercicio 4 — Worker pool con asyncio.Queue y backpressure

Implementa un crawler que procese una lista de URLs con como máximo 10 requests concurrentes (no 10.000 a la vez), reintentando fallos una vez, y que termine limpio cuando todas se procesaron.

Solución 
import asyncio

async def worker(q: asyncio.Queue, session, results: list) -> None:
    while True:
        url, attempts = await q.get()
        try:
            results.append(await fetch(session, url))
        except Exception:
            if attempts < 1:
                await q.put((url, attempts + 1))   # reintento: re-encola
            # si ya reintentó, se descarta (o loguear)
        finally:
            q.task_done()                          # marca ESTE item, siempre

async def crawl(urls: list[str], concurrency: int = 10) -> list:
    q: asyncio.Queue = asyncio.Queue()
    results: list = []
    for u in urls:
        q.put_nowait((u, 0))

    async with make_session() as session:
        workers = [
            asyncio.create_task(worker(q, session, results))
            for _ in range(concurrency)
        ]
        await q.join()                             # espera a que se drene la cola
        for w in workers:                          # while True -> cancelar
            w.cancel()
        await asyncio.gather(*workers, return_exceptions=True)
    return results

La concurrencia queda acotada por el número de workers (10), no por el número de URLs — eso es el control de concurrencia/backpressure. q.join() desbloquea cuando cada get() tuvo su task_done() correspondiente (incluidos los reintentos, porque re-encolar suma un item que también deberá marcarse). Alternativa más simple para acotar concurrencia sin cola: un asyncio.Semaphore(10) alrededor de cada fetch, con asyncio.gather sobre todas las URLs.

Preguntas tipo entrevista (EN)

Q1 — What’s the difference between concurrency and parallelism, and which one does asyncio give you?

Q2 — What exactly happens when you await a coroutine? And what does calling an async def without awaiting do?

Q3 — You call time.sleep(5) inside an async request handler. What happens, and how do you fix it?

Q4 — asyncio.gather vs asyncio.TaskGroup: when and why?

Q5 — Explain cancellation. Why shouldn’t you catch CancelledError with except Exception, and what must you do if you catch it?

Q6 — Coroutine vs Task vs Future — what’s the difference?

Q7 — When would you reach for run_in_executor, and how do you choose ThreadPool vs ProcessPool?

Referencias