7. Concurrencia

Un aviso antes de empezar: la mayoría de los scripts no necesitan concurrencia, y meterla sin pensar añade complejidad y errores difíciles de reproducir. Pero llega un punto (leer cien boyas por red, procesar mil ficheros) en que hacerlo todo en fila es demasiado lento. Esta lección te da el criterio para saber cuándo tirar de ella y qué herramienta usar.

Concurrencia, paralelismo y el GIL

Dos palabras que se confunden. Concurrencia es gestionar varias tareas que avanzan solapadas, aprovechando los ratos en que una espera. Paralelismo es ejecutarlas literalmente a la vez en varios núcleos del procesador.

La distinción que de verdad importa en la práctica es otra: ¿tu trabajo es de espera o de cálculo?

• Ligado a I/O (I/O-bound): el programa pasa el tiempo esperando algo externo, la red, el disco, un sensor. El procesador está de brazos cruzados.
• Ligado a CPU (CPU-bound): el programa quema procesador haciendo cuentas, un análisis espectral del oleaje, por ejemplo.

Y aquí entra el GIL (Global Interpreter Lock): en CPython, el intérprete estándar, se ejecuta un solo hilo de Python a la vez. Traducido: los hilos no te dan paralelismo real de cálculo, pero sí son perfectos para I/O, porque mientras un hilo espera la red, otro avanza. De esa distinción sale toda la regla de decisión.

concurrent.futures: hilos para la espera

Para trabajo ligado a I/O, la puerta de entrada es ThreadPoolExecutor. Reparte el trabajo entre varios hilos y, mientras unos esperan, otros progresan. La API es la misma que ya conoces de map:

from concurrent.futures import ThreadPoolExecutor

buoy_ids = ["north-sea-1", "north-sea-2", "med-1"]

with ThreadPoolExecutor(max_workers=8) as executor:
    weathers = list(executor.map(fetch_weather, buoy_ids))

Si fetch_weather tarda un segundo por boya esperando la red, hacerlo en fila son tres segundos; con el pool, poco más de uno. Fíjate en el with: el executor es un context manager, justo lo que acabas de aprender, y al salir espera a que terminen todas las tareas y se cierra solo.

Procesos para el cálculo

Cuando el cuello de botella es el procesador, los hilos no ayudan por culpa del GIL. Ahí necesitas procesos, cada uno con su propio intérprete y su propio GIL. El cambio es de una sola palabra, ProcessPoolExecutor, porque comparte API:

from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor() as executor:
    results = list(executor.map(spectral_analysis, big_files))

El precio a pagar: los datos viajan entre procesos serializados (con pickle), así que conviene pasar cosas simples y funciones definidas a nivel de módulo. Las funciones puras, sin estado compartido, son las que mejor se reparten.

asyncio: miles de esperas a la vez

Cuando no son diez esperas sino diez mil (un servicio que consulta muchas APIs), crear un hilo por cada una no escala. Para eso está asyncio: un solo hilo que hace malabares con miles de tareas mediante async y await.

import asyncio


async def fetch_weather(buoy_id: str) -> float:
    await asyncio.sleep(0.1)  # simula la espera de la red
    return 2.3


async def fetch_all(buoy_ids: list[str]) -> list[float]:
    tasks = [fetch_weather(buoy_id) for buoy_id in buoy_ids]
    return await asyncio.gather(*tasks)


results = asyncio.run(fetch_all(["north-sea-1", "med-1"]))

Una función async es una corrutina: await marca los puntos donde cede el turno mientras espera, para que otra tarea avance. asyncio.gather las lanza todas a la vez y recoge los resultados. Para hablar con APIs de verdad usarías una librería async como httpx o aiohttp, no requests, que es síncrona y bloquearía el hilo.

Qué elegir

No hay que decidir a ciegas, hay una regla clara:

Y una idea que cierra otro círculo del curso: el estilo funcional es tu mejor aliado en la concurrencia. Las funciones puras no comparten estado mutable, así que se pueden repartir entre hilos, procesos o corrutinas sin condiciones de carrera ni cerrojos. El estado compartido y mutable, ese que llevamos todo el curso evitando, es precisamente lo que convierte la concurrencia en un infierno de bugs imposibles de reproducir. Aíslas los efectos, mantienes puro el núcleo y paralelizarlo se vuelve casi gratis.

Para testear, el consejo de siempre: separa la lógica pura de la capa concurrente. Prueba el cálculo con valores, síncrono y sin sorpresas; la capa de hilos o corrutinas queda fina y se comprueba aparte.

Ejemplo guía: promediar una carpeta de registros en paralelo

Retomamos la carpeta de CSV de la lección anterior. Leer ficheros es trabajo de I/O, así que los hilos encajan. Separamos el cálculo puro (una función por fichero) de la orquestación concurrente. En src/metocean_tools/batch.py:

from concurrent.futures import ThreadPoolExecutor
from pathlib import Path


def average_of_file(path: Path) -> float:
    """Altura media de ola de un único fichero, en metros."""
    with path.open(encoding="utf-8") as file:
        next(file)  # saltar la cabecera
        heights = [float(line.strip().split(",")[1]) for line in file]
    return sum(heights) / len(heights) if heights else 0.0


def averages_for_folder(folder: Path) -> dict[str, float]:
    """Promedia todos los CSV de la carpeta a la vez y devuelve nombre -> media."""
    files = sorted(folder.glob("*.csv"))
    with ThreadPoolExecutor() as executor:
        averages = executor.map(average_of_file, files)
    return {path.name: average for path, average in zip(files, averages)}

Cada fichero se procesa en un hilo distinto y, mientras uno espera al disco, otro calcula. Si la carpeta tiene cien registros, la diferencia con hacerlo en fila es enorme. Y como average_of_file no comparte estado con las demás llamadas, no hemos necesitado un solo cerrojo.