Integrar código bloqueante y CPU

Una biblioteca síncrona de red bloquea al hilo que la llama mientras espera. Si la ejecutas directamente en el event loop, bloqueas todas sus tareas. Un cálculo largo también ocupa ese hilo, aunque no tenga ninguna operación de red.

Para I/O síncrono que no puedes sustituir, delegar a un hilo permite mantener el bucle disponible. Para trabajo intenso de Python puro, un pool de procesos puede ejecutar cálculos en otros núcleos. Son adaptaciones con costes y ciclos de vida propios, no conversiones automáticas de código síncrono en corrutinas.

asyncio.to_thread(funcion, argumentos...) ejecuta la llamada en un hilo y ofrece una corrutina para esperar su resultado. La función debe pasarse sin ejecutarla en el bucle.

import asyncio
import tempfile
from pathlib import Path


async def main():
    with tempfile.TemporaryDirectory() as directorio:
        ruta = Path(directorio) / "nota.txt"
        await asyncio.to_thread(ruta.write_text, "hola", encoding="utf-8")
        texto = await asyncio.to_thread(ruta.read_text, encoding="utf-8")
        assert texto == "hola"
        print(texto)


asyncio.run(main())

Path representa una ruta y ofrece métodos de lectura/escritura. TemporaryDirectory crea un espacio temporal y lo elimina al salir. Pasar ruta.read_text() sería un error: realizaría la lectura antes de llamar a to_thread. Al esperar la operación delegada recibes su resultado o su excepción.

Un archivo minúsculo no necesita hilos por rendimiento. El ejemplo muestra el contrato. La decisión real depende de cuánto puede bloquear la llamada y de la latencia que deben mantener las otras tareas.

No existe una forma general segura de matar un hilo en medio de una función arbitraria. Si el trabajo ya empezó, cancelar su espera asíncrona no impide que siga escribiendo archivos, llamando a una API o consumiendo CPU.

Este experimento evita depender de la velocidad del equipo: el hilo avisa cuando empezó y espera una señal explícita antes de producir el efecto.

import asyncio
import threading


async def main():
    loop = asyncio.get_running_loop()
    iniciada = asyncio.Event()
    terminada = asyncio.Event()
    liberar = threading.Event()
    efectos = []

    def trabajo():
        loop.call_soon_threadsafe(iniciada.set)
        liberar.wait()
        efectos.append("efecto realizado")
        loop.call_soon_threadsafe(terminada.set)

    tarea = asyncio.create_task(asyncio.to_thread(trabajo))
    try:
        await iniciada.wait()
        tarea.cancel()
        try:
            await tarea
        except asyncio.CancelledError:
            print("la espera fue cancelada")
        assert efectos == []
    finally:
        liberar.set()
        await terminada.wait()
    assert efectos == ["efecto realizado"]
    print(efectos)


asyncio.run(main())

threading.Event se utiliza en el hilo; asyncio.Event, en el bucle. loop.call_soon_threadsafe envía al bucle una función para que la ejecute con seguridad desde su hilo. No llamamos directamente a asyncio.Event.set() desde el worker.

La prueba termina esperando la señal del hilo: no abandona trabajo indefinido. En un servicio real, la función síncrona debería tener timeouts propios y, si necesita detenerse, aceptar una señal cooperativa que consulte entre pasos. Una señal no puede interrumpir una llamada nativa bloqueada que nunca vuelve a comprobarla.

Un error sutil es adquirir un semáforo, esperar to_thread, cancelar esa espera y liberar el permiso mientras el hilo sigue trabajando. Una nueva tarea puede entrar y superar el número de operaciones reales que pretendías limitar.

Para integrar una biblioteca bloqueante de forma sostenida, utiliza workers dedicados que posean sus llamadas hasta que terminen, un pool explícito con capacidad conocida y una cola de entrada acotada. Los timeouts internos del cliente síncrono deben impedir que un worker quede ocupado para siempre.

El pool limita los hilos ejecutando, pero su cola interna puede aceptar muchos envíos. No hagas cien mil llamadas a run_in_executor por adelantado. Aplica también contrapresión antes de enviar trabajo.

En CPython 3.12 habitual, el GIL limita la ejecución simultánea de bytecode Python entre hilos. Dos hilos de Python puro no equivalen a dos núcleos calculando independientemente. Algunas bibliotecas nativas liberan el GIL; en ellas los hilos pueden comportarse de otro modo. Mide la biblioteca concreta.

Un proceso tiene su propio intérprete y memoria. Puede calcular en paralelo, a cambio de arrancar otro proceso y serializar entradas/salidas. Si cada cálculo tarda menos que enviarlo y reconstruirlo, el pool puede ser más lento que un bucle sencillo.

Guarda el siguiente programa en un archivo .py y ejecútalo como script. El modo spawn inicia intérpretes nuevos; necesitan importar la función por su nombre. Por eso la función está en el nivel superior y el arranque está protegido por if __name__ == "__main__".

import asyncio
import multiprocessing
from concurrent.futures import ProcessPoolExecutor


def suma_cuadrados(limite):
    return sum(numero * numero for numero in range(limite))


async def main(pool):
    loop = asyncio.get_running_loop()
    resultados = await asyncio.gather(
        loop.run_in_executor(pool, suma_cuadrados, 100_000),
        loop.run_in_executor(pool, suma_cuadrados, 200_000),
    )
    assert resultados[0] == 333328333350000
    print(resultados)


if __name__ == "__main__":
    contexto = multiprocessing.get_context("spawn")
    with ProcessPoolExecutor(max_workers=2, mp_context=contexto) as pool:
        asyncio.run(main(pool))

run_in_executor devuelve un objeto esperable que conecta el resultado del executor con el bucle. El with del pool está fuera de asyncio.run: su cierre síncrono puede esperar trabajo pendiente sin bloquear un bucle que todavía atienda solicitudes.

No pases sockets, clientes HTTP ni locks de asyncio a los procesos. Envía datos serializables y devuelve resultados pequeños. Las lambdas o funciones locales complican la importación bajo spawn; una función de módulo deja claro qué código debe ejecutar cada proceso.

Son tres estados distintos. Cancelar un future puede impedir un trabajo todavía no iniciado. Una vez que está ejecutándose, Future.cancel() no lo mata. shutdown(cancel_futures=True) cancela pendientes que no han empezado; los que ya ejecutan siguen su política de finalización.

Si necesitas detener forzosamente una herramienta externa, puedes ejecutarla como subproceso independiente y diseñar una secuencia de señal, plazo y terminación. Eso exige tratar recursos y efectos parciales; no equivale a cancelar una corrutina. En un pool compartido, matar un proceso también puede afectar a otros trabajos y dejar el pool inutilizable.

Para una aplicación pequeña es razonable empezar con cálculos acotados, un pool pequeño y un plazo de parada que admita su peor duración. Añade aislamiento más fuerte cuando exista una necesidad concreta.

La prueba del hilo debe imprimir el efecto después de cancelar la espera. Si eliminases liberar.set(), el programa podría no cerrar porque el trabajo delegado nunca terminó. Ese es precisamente el riesgo que una aplicación debe controlar.

En el ejemplo de procesos, compara entradas pequeñas y grandes sin convertir el cronómetro en una garantía universal. Lo que queda demostrado es separación de ejecución y resultado correcto; la ganancia depende del tamaño de trabajo, el número de núcleos y la carga de la máquina.

• to_thread, Python 3.12.
• Executors y cancelación, Python 3.12.
• run_in_executor y comunicación entre hilos.