53 · concurrent.futures: ThreadPool y ProcessPool a la manera moderna
concurrent.futures envuelve hilos y procesos en una API limpia basada en Future. Cambia ThreadPoolExecutor por ProcessPoolExecutor cambiando una palabra, y usa as_completed() para manejar resultados cuando terminan.
`concurrent.futures` es la API de concurrencia de alto nivel de Python. `ThreadPoolExecutor` y `ProcessPoolExecutor` son intercambiables — la misma interfaz `executor.submit()`, `executor.map()` y `as_completed()` funciona para ambos. Esta simetría significa que puedes prototipar con hilos (sin restricciones de Pyodide, inicio instantáneo) y cambiar a procesos cambiando un nombre de clase cuando el profiling revela que el cuello de botella es la CPU. `as_completed(futures)` te da resultados cuando terminan — no en el orden de envío — que es exactamente lo que quieres para tareas I/O con latencia variable.
Sin esto:
Sin `concurrent.futures` gestionas hilos o procesos en bruto manualmente — rastreando cuáles hilos siguen corriendo, recopilando sus resultados, manejando excepciones por hilo. Este boilerplate es propenso a errores y hace que cambiar entre pools de hilos y procesos sea una refactorización de varias horas. `concurrent.futures` lo reduce a un cambio de una sola palabra.
El diseño de concurrent.futures
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
Tres conceptos centrales:
-
Executor — un gestor de pool.
ThreadPoolExecutor(max_workers=N)crea N hilos;ProcessPoolExecutor(max_workers=N)crea N procesos. Ambos son context managers (withcierra el pool de forma elegante). -
Future — un handle a un resultado que aún no está listo.
future.result()se bloquea hasta que el valor esté disponible;future.exception()devuelve cualquier excepción que se haya lanzado;future.done()no es bloqueante. -
Dos métodos de envío:
executor.submit(fn, *args)— envía una tarea, devuelve unFutureinmediatamente.executor.map(fn, iterable)— envía todas las tareas, devuelve resultados en orden de envío (iterador perezoso; cada.next()puede bloquearse).
as_completed vs map
| | executor.map(fn, items) | as_completed(futures) |
|---|---|---|
| Orden | Orden de envío | Orden de finalización |
| Mejor para | Pipelines ordenadas | I/O con latencia variable |
| Manejo de excepciones | Se lanza al iterar | Llama a future.exception() o future.result() |
La regla de oro
Comienza con secuencial. Si es I/O-bound →
ThreadPoolExecutor. Si es CPU-bound →ProcessPoolExecutor. Usathreading.Threaden bruto solo cuando necesites control fino (hilos daemon personalizados, orden de join manual, hooks de ciclo de vida por hilo).
Python (in browser)
`executor.map(fn, iterable)` es un reemplazo directo del `map(fn, iterable)` incorporado de Python — con el executor distribuyendo tareas en el pool. El context manager (`with`) llama a `executor.shutdown(wait=True)` automáticamente: todas las tareas enviadas se completan antes de que salga el bloque.
Python runs entirely in your browser via Pyodide (~6 MB on first Run, cached after).
Python (in browser)
Localmente, las tareas 3 (1 ms) y 1 (2 ms) terminarían e imprimirían primero aunque las tareas 0–2 se enviaron primero. En Pyodide terminan en orden de envío porque los hilos son secuenciales. El dict `future_to_id` es el patrón estándar para mapear un `Future` de vuelta a su entrada.
Python runs entirely in your browser via Pyodide (~6 MB on first Run, cached after).
Este patrón — `submit` todas las tareas primero, luego `as_completed` — es sobre el que están construidos los frameworks de scraping, los clientes API paralelos y los ejecutores de pruebas paralelas. El thread pool maneja la planificación; tú solo procesas los resultados cuando llegan.
El cambio de una sola palabra de `ThreadPoolExecutor` a `ProcessPoolExecutor` es el punto central de `concurrent.futures`. Prototipa con hilos (inicio instantáneo, sin restricciones de Pyodide), mide, luego cambia a procesos si el profiler muestra saturación de CPU.
¿Qué devuelve `as_completed(futures)`?
- `concurrent.futures` proporciona una API unificada basada en `Future` para tanto pools de hilos (`ThreadPoolExecutor`) como pools de procesos (`ProcessPoolExecutor`). Cambiar entre ellos requiere cambiar un nombre de clase.
- `executor.submit(fn, *args)` devuelve un `Future` inmediatamente. `executor.map(fn, iterable)` es un wrapper de conveniencia que produce resultados en orden de envío.
- `as_completed(futures)` produce objetos `Future` en **orden de finalización** — el más rápido primero. Ideal para tareas I/O con latencia variable donde quieres procesar resultados cuando llegan.
- El context manager (`with executor:`) llama a `executor.shutdown(wait=True)` automáticamente — todas las tareas enviadas se completan antes de que salga el bloque.
- Usa `concurrent.futures` primero. Usa `threading.Thread` en bruto solo cuando necesites control fino del ciclo de vida. `joblib.Parallel` es un wrapper de nivel superior con mejores defaults para numpy/ML.
El preprocesamiento de datos en paralelo en una pipeline de entrenamiento — redimensionar y normalizar imágenes antes de que entren a un DataLoader — es una carga de trabajo clásica de `ThreadPoolExecutor` cuando el cuello de botella son las lecturas de disco (I/O). La inferencia en batch sobre múltiples entradas independientes (puntuar N documentos simultáneamente) se mapea naturalmente a `executor.map(score, documents)`. La búsqueda de hiperparámetros en paralelo (`GridSearchCV(n_jobs=-1)`) y `joblib.Parallel` en scikit-learn son thin wrappers sobre `ProcessPoolExecutor`. Ejecutar múltiples configuraciones de experimentos simultáneamente en una máquina multi-GPU usa aislamiento a nivel de proceso para evitar el estado de GPU compartido.
Si lo quitas: Sin `concurrent.futures`, cambiar una pipeline de preprocesamiento de hilos a procesos para encontrar cuál es más rápida requiere una refactorización significativa — reescribir la lógica de `Thread` + join manual en lógica de `Process` + join manual. Con executors, el mismo código sirve para ambos, y puedes hacer benchmark del cambio en 30 segundos. Esta es la razón principal por la que existe `concurrent.futures`: hacer que el modo de concurrencia sea un detalle de configuración, no una decisión de arquitectura.