multiprocessing en Python para tareas intensivas en CPU

1. Descripción general

Cuando tu código Python debe hacer muchos cálculos, por ejemplo procesar imágenes, calcular checksums o simular números, verás que usar threads casi no acelera nada. La razón es el GIL (Global Interpreter Lock). Este artículo explica brevemente ese problema. Luego te guía sobre cómo usar el módulo multiprocessing y ProcessPoolExecutor para dividir el trabajo entre múltiples procesos. Al terminar, sabrás cómo paralelizar una tarea CPU-bound, escribir pruebas para ella y evitar errores comunes.

Todos los ejemplos usan Python 3.12+ y solo necesitan la librería estándar, sin instalaciones adicionales.

2. Vía saque threading no ayuda

2.1. Una tarea intensiva en CPU

Usamos una función que cuenta números primos escrita en Python puro. Este es un trabajo típico que consume CPU:

# cpu_task.py
import math

def count_primes(start: int, end: int) -> int:
    count = 0
    for n in range(max(start, 2), end):
        if all(n % d for d in range(2, math.isqrt(n) + 1)):
            count += 1
    return count

La función recorre cada número en [start, end) e intenta dividirlo por números desde 2 hasta su raíz cuadrada. La función no lee archivos ni hace llamadas de red. Solo calcula.

2.2. Medir tiempo: secuencial y thread

# bench_threads.py
import time
from concurrent.futures import ThreadPoolExecutor
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

def run_sequential() -> int:
    return sum(count_primes(a, b) for a, b in RANGES)

def run_threads() -> int:
    starts, ends = zip(*RANGES)
    with ThreadPoolExecutor(max_workers=8) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    for fn in (run_sequential, run_threads):
        t0 = time.perf_counter()
        result = fn()
        print(f"{fn.__name__}: {result} primes, {time.perf_counter() - t0:.2f}s")

RANGES divide el intervalo [0, 1_000_000) en 8 partes. run_threads asigna cada parte a un thread. El resultado en una máquina de 8 núcleos (los números en tu máquina serán diferentes):

run_sequential: 78498 primes, 6.41s
run_threads: 78498 primes, 6.58s

La versión con threads es incluso un poco más lenta. En CPython estándar, en cada momento solo un thread ejecuta bytecode de Python, porque debe mantener el GIL. Ocho threads solo se turnan usando un núcleo, y además hay un costo de cambio de contexto entre ellos.

Nota: Desde Python 3.13 hay una construcción free-threaded (python3.13t) que elimina el GIL. Esta construcción aún está en fase experimental, y muchas extensiones C no la soportan. Para la mayoría de proyectos hoy en día, usar procesos sigue siendo la forma más segura.

3. Conceptos básicos de multiprocessing

Cada proceso tiene su propio intérprete y su propio GIL. Por lo tanto, N procesos pueden ejecutarse realmente en paralelo en N núcleos.

3.1. Crear Process manualmente

# basic_process.py
from multiprocessing import Process, Queue
from cpu_task import count_primes

def worker(start: int, end: int, results: Queue) -> None:
    results.put(count_primes(start, end))

if __name__ == "__main__":
    results = Queue()
    ranges = [(0, 500_000), (500_000, 1_000_000)]
    processes = [Process(target=worker, args=(a, b, results)) for a, b in ranges]

    for p in processes:
        p.start()
    total = sum(results.get() for _ in processes)
    for p in processes:
        p.join()

    print("Total primes:", total)

Explicación de cada parte:

• Process(target=..., args=...) describe un proceso hijo. No se ejecuta hasta que llames a start().
• El proceso hijo no comparte memoria con el proceso padre. Si quieres enviar un resultado, debes usar un canal de comunicación. Aquí es Queue.
• El programa llama a results.get() antes de join(). Si llamas a join() primero mientras la cola tiene datos grandes sin leer, el programa puede deadlock.

Total primes: 78498

Este enfoque es flexible pero verboso. En la práctica, rara vez administrarás Process manualmente.

3.2. Usar Pool

Pool crea un grupo de workers preconfigurados y distribuye el trabajo entre ellos automáticamente:

# pool_example.py
from multiprocessing import Pool
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

if __name__ == "__main__":
    with Pool(processes=8) as pool:
        counts = pool.starmap(count_primes, RANGES)
    print(counts)
    print("Total:", sum(counts))

starmap desempaqueta cada tupla en RANGES como parámetros start, end. Los resultados se devuelven en el mismo orden que la entrada:

[10746, 9895, 9622, 9480, 9637, 9554, 9454, 9610]
Total: 78498

Si no pasas processes, Pool usa os.cpu_count() workers. El bloque with limpiará los workers cuando termine.

4. ProcessPoolExecutor: un API más moderno

concurrent.futures.ProcessPoolExecutor tiene la misma interfaz que ThreadPoolExecutor. Gracias a esto, puedes cambiar de threads a procesos casi solo reemplazando el nombre de la clase:

# parallel.py
import os
from concurrent.futures import ProcessPoolExecutor
from cpu_task import count_primes

def split_range(start: int, end: int, parts: int) -> list[tuple[int, int]]:
    step = (end - start + parts - 1) // parts
    return [(s, min(s + step, end)) for s in range(start, end, step)]

def parallel_count_primes(start: int, end: int, workers: int | None = None) -> int:
    workers = workers or os.cpu_count() or 1
    chunks = split_range(start, end, workers)
    starts, ends = zip(*chunks)
    with ProcessPoolExecutor(max_workers=workers) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    import time
    t0 = time.perf_counter()
    print(parallel_count_primes(0, 1_000_000, workers=8))
    print(f"{time.perf_counter() - t0:.2f}s")

78498
1.27s

El tiempo se reduce de aproximadamente 6,4 segundos a 1,3 segundos. La velocidad no aumenta exactamente 8 veces por dos razones. Primero, los segmentos no tienen un volumen de trabajo uniforme: los números más grandes toman más tiempo para verificar. Segundo, iniciar procesos y transferir datos también cuesta tiempo.

Si quieres procesar cada resultado tan pronto como esté listo, usa submit junto con as_completed:

from concurrent.futures import ProcessPoolExecutor, as_completed
from cpu_task import count_primes

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(count_primes, a, a + 250_000): a
                   for a in range(0, 1_000_000, 250_000)}
        for future in as_completed(futures):
            print(f"Rango comenzando en {futures[future]}: {future.result()} primos")

Si un worker lanza una excepción, future.result() la relanzará en el proceso padre. Por lo tanto, puedes capturar errores con try/except como de costumbre.

5. Muchas tareas pequeñas: chunksize e imap_unordered

Cada vez que envías una tarea a un worker, Python debe hacer pickle de los parámetros, enviarlos por pipe, e inmediatamente unpickle el resultado. Si tienes 100.000 tareas muy pequeñas, este costo puede ser mayor que la propia computación. Lo resuelves agrupando tareas en lotes usando chunksize:

# chunks.py
from multiprocessing import Pool

def is_prime(n: int) -> bool:
    if n < 2:
        return False
    return all(n % d for d in range(2, int(n ** 0.5) + 1))

if __name__ == "__main__":
    numbers = range(1_000_000)
    with Pool() as pool:
        total = sum(pool.imap_unordered(is_prime, numbers, chunksize=10_000))
    print(total)

• chunksize=10_000 hace que cada envío contenga 10.000 números en lugar de uno. El número de comunicaciones entre procesos se reduce mucho.
• imap_unordered devuelve resultados de forma lazy y no mantiene el orden. Esto es útil cuando solo necesitas agregar resultados, como sumarlos, y quieres ahorrar memoria.
• ProcessPoolExecutor.map también acepta el parámetro chunksize. Su valor por defecto es 1, así que recuerda ajustarlo cuando tengas muchas tareas pequeñas.

6. Inicializar recursos una vez por worker

Supongamos que cada tarea necesita una tabla de búsqueda que cuesta tiempo crear. Si creas la tabla nuevamente en cada invocación de función, es muy derrochador. El parámetro initializer te permite ejecutar una función exactamente una vez cuando el worker se inicia:

# initializer_example.py
from concurrent.futures import ProcessPoolExecutor

_price_table: dict[str, float] = {}

def load_prices(currency: str) -> None:
    global _price_table
    rate = 25_000 if currency == "VND" else 1
    _price_table = {f"SKU-{i}": i * 1.5 * rate for i in range(100_000)}

def order_total(skus: list[str]) -> float:
    return sum(_price_table[s] for s in skus)

if __name__ == "__main__":
    orders = [["SKU-1", "SKU-2"], ["SKU-10", "SKU-20", "SKU-30"]]
    with ProcessPoolExecutor(max_workers=2, initializer=load_prices,
                             initargs=("VND",)) as executor:
        print(list(executor.map(order_total, orders)))

[112500.0, 2250000.0]

La variable _price_table es una variable global en cada worker. El proceso padre no ve esos cambios. Esta es la forma segura de que cada worker tenga sus propios recursos sin tener que enviar la tabla de búsqueda con cada tarea.

7. Escribir pruebas para código paralelo

Puedes probar código que usa procesos con pytest como código normal. La condición es que la función worker debe estar definida a nivel de módulo:

# test_parallel.py
from cpu_task import count_primes
from parallel import parallel_count_primes, split_range

def test_count_primes_known_value():
    assert count_primes(0, 100) == 25

def test_split_range_covers_whole_interval():
    chunks = split_range(0, 10, 3)
    assert chunks == [(0, 4), (4, 8), (8, 10)]

def test_parallel_matches_sequential():
    expected = count_primes(0, 50_000)
    assert parallel_count_primes(0, 50_000, workers=4) == expected

pip install pytest
pytest -q test_parallel.py

...                                                    [100%]
3 passed in 0.62s

La prueba más importante es test_parallel_matches_sequential. Asegura que la división de rangos no pierde ni duplica elementos. Este es un error muy común al escribir split_range manualmente.

8. Errores comunes y cuándo no usar multiprocessing

Olvidar if __name__ == "__main__":. Con el método de inicio spawn (por defecto en Windows y macOS) o forkserver (por defecto en Linux desde Python 3.14), el proceso hijo reimportará el módulo principal. Si el código que crea el pool está a nivel de módulo, cada proceso hijo creará un nuevo pool. Entonces obtendrás RuntimeError o procesos que se generan indefinidamente.

Pasar lambdas o funciones anidadas. Todo lo que se envía a un worker debe ser serializable con pickle:

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(lambda x: x * x, range(5))))

_pickle.PicklingError: Can't pickle <function <lambda> at 0x...>: attribute lookup <lambda> on __main__ failed

Define la función usando def a nivel de módulo. Si necesitas fijar algunos parámetros, usa functools.partial.

Pasar datos muy grandes. Una lista de millones de elementos pasada a cada tarea se hará pickle múltiples veces. El tiempo de copia puede dominar el cálculo. Hay varios arreglos: enviar solo índices o rutas de archivo, usar initializer, o usar multiprocessing.shared_memory con datos tipo array.

Depender del comportamiento de fork. Código que funciona en Linux antiguo gracias a fork (el proceso hijo hereda variables globales) puede fallar en macOS, Windows o Python 3.14. Para tener un comportamiento consistente, elige explícitamente el método de inicio:

import multiprocessing as mp

if __name__ == "__main__":
    mp.set_start_method("spawn")

Nota: No uses multiprocessing para tareas I/O-bound como llamadas HTTP o consultas a bases de datos. Mientras esperas I/O, el thread sigue liberando el GIL, así que ThreadPoolExecutor o asyncio son mucho más ligeros. Tampoco uses multiprocessing cuando cada tarea toma solo unos pocos microsegundos sin agrupar, o cuando el cálculo ya está en una librería C que libera el GIL (NumPy, funciones hash en datos grandes…). En esos casos, threads o vectorización suelen ser suficientes.

Crear demasiados workers. Establecer el número de workers mayor que el número de núcleos de CPU no acelera el código CPU-bound. Solo desperdicia RAM, porque cada proceso es un intérprete completo. Comenzar con os.cpu_count() es razonable.

9. Conclusión

El GIL hace que threads no aceleren el código Python puro intensivo en CPU. Cada proceso tiene su propio GIL, así que dividir el trabajo entre múltiples procesos es la forma simple de usar todos los núcleos de CPU. En la mayoría de casos, ProcessPoolExecutor es la opción más limpia. Pool es útil cuando necesitas imap_unordered o starmap. Los Process manuales solo son para situaciones especiales.

Cuando uses multiprocessing, recuerda algunos puntos:

• Siempre incluye if __name__ == "__main__":.
• Solo pasa funciones y datos que se puedan serializar con pickle.
• Aumenta chunksize cuando tengas muchas tareas pequeñas.
• Usa initializer para recursos pesados.
• Escribe una prueba que compare el resultado paralelo con el resultado secuencial para asegurar que la división de rangos sea correcta.