Multiprocessing y Paralelismo CPU en Python

Multiprocessing y Paralelismo CPU en Python

El GIL (Global Interpreter Lock) de CPython impide que múltiples hilos ejecuten código Python puro en paralelo. Para código CPU-intensivo, la solución es multiprocessing: procesos separados, sin GIL, con toda la capacidad de los núcleos del procesador.

Cuándo usar qué

Process básico

import multiprocessing as mp
import os
import time

def tarea_cpu(nombre: str, n: int) -> int:
    """Calcula la suma de cuadrados — tarea CPU-intensiva de ejemplo."""
    pid = os.getpid()
    print(f"[PID {pid}] Iniciando {nombre} con n={n}")
    resultado = sum(i * i for i in range(n))
    print(f"[PID {pid}] {nombre} terminó: {resultado}")
    return resultado

if __name__ == '__main__':   # ← OBLIGATORIO en Windows/macOS con spawn
    # Crear y lanzar procesos manualmente
    p1 = mp.Process(target=tarea_cpu, args=('A', 1_000_000))
    p2 = mp.Process(target=tarea_cpu, args=('B', 2_000_000))

    t0 = time.perf_counter()
    p1.start()
    p2.start()
    p1.join()    # esperar a que termine p1
    p2.join()    # esperar a que termine p2
    elapsed = time.perf_counter() - t0

    print(f"Tiempo total (paralelo): {elapsed:.2f}s")
    print(f"Proceso padre PID: {os.getpid()}")

Pool.map — paralelismo sobre colección

import multiprocessing as mp
import math
import time

def factorizar(n: int) -> list[int]:
    """Factorización por división de prueba."""
    factores = []
    d = 2
    while d * d <= n:
        while n % d == 0:
            factores.append(d)
            n //= d
        d += 1
    if n > 1:
        factores.append(n)
    return factores

if __name__ == '__main__':
    numeros = [10**12 + i for i in range(100)]

    # Sin paralelismo
    t0 = time.perf_counter()
    resultados_seq = [factorizar(n) for n in numeros]
    t_seq = time.perf_counter() - t0

    # Con Pool (usa todos los núcleos disponibles)
    ncpus = mp.cpu_count()
    t0 = time.perf_counter()
    with mp.Pool(processes=ncpus) as pool:
        resultados_par = pool.map(factorizar, numeros)
    t_par = time.perf_counter() - t0

    print(f"Secuencial: {t_seq:.2f}s")
    print(f"Paralelo ({ncpus} núcleos): {t_par:.2f}s")
    print(f"Speedup: {t_seq/t_par:.1f}x")

Pool con función de múltiples argumentos

import multiprocessing as mp
from functools import partial

def procesar_imagen(ruta: str, calidad: int, formato: str) -> dict:
    """Convierte una imagen — intensivo en CPU."""
    import time, pathlib
    time.sleep(0.1)   # simular conversión
    return {
        'entrada': ruta,
        'salida':  pathlib.Path(ruta).with_suffix(f'.{formato}').name,
        'calidad': calidad,
    }

if __name__ == '__main__':
    rutas = [f'foto_{i:03d}.jpg' for i in range(20)]

    # starmap — múltiples argumentos como tuplas
    argumentos = [(r, 85, 'webp') for r in rutas]
    with mp.Pool() as pool:
        resultados = pool.starmap(procesar_imagen, argumentos)

    # partial — fijar argumentos comunes
    convertir_webp = partial(procesar_imagen, calidad=85, formato='webp')
    with mp.Pool() as pool:
        resultados2 = pool.map(convertir_webp, rutas)

    print(f"Convertidas: {len(resultados)} imágenes")

Pool.imap — procesamiento en streaming

import multiprocessing as mp

def calcular_pesado(n: int) -> float:
    return sum(i**0.5 for i in range(n))

if __name__ == '__main__':
    numeros = range(0, 10_000_000, 100_000)

    with mp.Pool() as pool:
        # imap devuelve un iterador — procesa sin esperar a todos
        # chunksize: agrupar items para reducir overhead de comunicación
        for i, resultado in enumerate(pool.imap(calcular_pesado, numeros, chunksize=4)):
            print(f"  [{i+1}] {resultado:.2f}")

        # imap_unordered — resultados en orden de llegada (más rápido si el orden no importa)
        for resultado in pool.imap_unordered(calcular_pesado, list(numeros), chunksize=4):
            pass   # procesar cada resultado conforme llega

Comunicación entre procesos — Queue y Pipe

import multiprocessing as mp
import time

# ── Queue — múltiples productores/consumidores ────────────────────────────────
def productor(cola: mp.Queue, pid: int, n: int):
    for i in range(n):
        item = {'pid': pid, 'item': i, 'datos': i ** 2}
        cola.put(item)
        time.sleep(0.01)
    cola.put(None)   # señal de fin

def consumidor(cola: mp.Queue, cid: int, resultados: mp.Queue):
    total = 0
    while True:
        item = cola.get()
        if item is None:
            cola.put(None)   # propagar señal para otros consumidores
            break
        total += item['datos']
    resultados.put({'consumidor': cid, 'total': total})

if __name__ == '__main__':
    cola_trabajo   = mp.Queue(maxsize=50)
    cola_resultados = mp.Queue()

    productores  = [mp.Process(target=productor,  args=(cola_trabajo, i, 10)) for i in range(3)]
    consumidores = [mp.Process(target=consumidor, args=(cola_trabajo, j, cola_resultados)) for j in range(2)]

    for p in productores + consumidores:
        p.start()
    for p in productores:
        p.join()
    for p in consumidores:
        p.join()

    while not cola_resultados.empty():
        r = cola_resultados.get()
        print(f"Consumidor {r['consumidor']}: total={r['total']}")


# ── Pipe — comunicación bidireccional entre 2 procesos ────────────────────────
def trabajador_pipe(conn_hijo):
    while True:
        datos = conn_hijo.recv()
        if datos == 'STOP':
            conn_hijo.close()
            break
        conn_hijo.send(datos ** 2)

if __name__ == '__main__':
    conn_padre, conn_hijo = mp.Pipe(duplex=True)
    p = mp.Process(target=trabajador_pipe, args=(conn_hijo,))
    p.start()

    for n in [5, 10, 15]:
        conn_padre.send(n)
        resultado = conn_padre.recv()
        print(f"{n}² = {resultado}")

    conn_padre.send('STOP')
    p.join()

Memoria compartida

import multiprocessing as mp
import numpy as np
import ctypes

def incrementar(array_compartido, inicio: int, fin: int):
    """Incrementa una sección del array en un proceso hijo."""
    for i in range(inicio, fin):
        array_compartido[i] += 1

if __name__ == '__main__':
    # Value — un único valor compartido
    contador = mp.Value(ctypes.c_int, 0)
    lock     = mp.Lock()

    def incrementar_contador(v, lck):
        for _ in range(1000):
            with lck:   # necesario para evitar condición de carrera
                v.value += 1

    procesos = [mp.Process(target=incrementar_contador, args=(contador, lock))
                for _ in range(4)]
    for p in procesos: p.start()
    for p in procesos: p.join()
    print(f"Contador final: {contador.value}")   # debe ser 4000

    # Array — array de C tipado compartido
    arr_compartido = mp.Array(ctypes.c_double, 1000)
    for i in range(1000):
        arr_compartido[i] = float(i)

    ps = [
        mp.Process(target=incrementar, args=(arr_compartido, i*200, (i+1)*200))
        for i in range(5)
    ]
    for p in ps: p.start()
    for p in ps: p.join()
    print(f"arr[500] = {arr_compartido[500]}")   # 501.0

    # shared_memory (Python 3.8+) — NumPy sobre memoria compartida
    from multiprocessing.shared_memory import SharedMemory

    datos = np.arange(1000, dtype=np.float64)
    shm   = SharedMemory(create=True, size=datos.nbytes)
    arr_np = np.ndarray(datos.shape, dtype=datos.dtype, buffer=shm.buf)
    arr_np[:] = datos[:]

    # Pasar nombre de la memoria compartida a procesos hijos
    def procesar_shared(nombre: str, shape, dtype_str: str):
        from multiprocessing.shared_memory import SharedMemory
        import numpy as np
        shm2  = SharedMemory(name=nombre)
        arr2  = np.ndarray(shape, dtype=np.dtype(dtype_str), buffer=shm2.buf)
        arr2 *= 2
        shm2.close()

    p = mp.Process(target=procesar_shared, args=(shm.name, datos.shape, str(datos.dtype)))
    p.start()
    p.join()
    print(f"arr_np[0] = {arr_np[0]}")   # 0.0 * 2 = 0.0
    shm.close()
    shm.unlink()   # liberar memoria compartida

ProcessPoolExecutor — la API más limpia

from concurrent.futures import ProcessPoolExecutor, as_completed
import math
import time

def es_primo(n: int) -> bool:
    if n < 2: return False
    if n == 2: return True
    if n % 2 == 0: return False
    return all(n % i != 0 for i in range(3, int(math.isqrt(n)) + 1, 2))

if __name__ == '__main__':
    numeros = range(10**7, 10**7 + 10_000)

    # map — bloquea hasta que todos terminan, conserva orden
    with ProcessPoolExecutor() as ex:
        t0 = time.perf_counter()
        primos = list(filter(None, ex.map(es_primo, numeros, chunksize=500)))
        print(f"Primos encontrados: {len(primos)} en {time.perf_counter()-t0:.2f}s")

    # submit + as_completed — resultados conforme llegan
    with ProcessPoolExecutor(max_workers=4) as ex:
        futuros = {ex.submit(es_primo, n): n for n in range(10**7, 10**7 + 100)}
        for futuro in as_completed(futuros):
            n = futuros[futuro]
            if futuro.result():
                print(f"  Primo: {n}")

Inicializador de pool — recursos compartidos por hijo

import multiprocessing as mp
import sqlite3

conexion_bd = None   # global en cada proceso hijo

def inicializar_worker(ruta_bd: str):
    """Se ejecuta una vez al crear cada proceso hijo."""
    global conexion_bd
    conexion_bd = sqlite3.connect(ruta_bd)
    print(f"[PID {mp.current_process().pid}] BD conectada")

def procesar_registro(registro_id: int) -> dict:
    """Usa la conexión ya abierta por el inicializador."""
    global conexion_bd
    cur = conexion_bd.execute("SELECT * FROM registros WHERE id=?", (registro_id,))
    fila = cur.fetchone()
    return {'id': registro_id, 'datos': fila}

if __name__ == '__main__':
    ids = list(range(1, 101))
    with mp.Pool(
        processes=4,
        initializer=inicializar_worker,
        initargs=('datos.db',),
    ) as pool:
        resultados = pool.map(procesar_registro, ids)
    print(f"Procesados: {len(resultados)} registros")

Buenas prácticas

• if __name__ == '__main__': es obligatorio en Windows y macOS (donde el método de inicio es spawn, no fork). Sin esto, cada proceso hijo intentará importar el script y ejecutar el código de nivel superior de nuevo.
• Usar Pool.map en lugar de Process manual para paralelizar sobre colecciones — gestiona automáticamente el pool de trabajadores y la recopilación de resultados.
• chunksize en pool.map/imap para reducir el overhead de comunicación IPC: grupos más grandes = menos mensajes, pero peor balance de carga.
• ProcessPoolExecutor de concurrent.futures es la API más moderna y pythónica; úsala salvo que necesites funciones específicas de Pool (como initializer).
• No compartir objetos no serializables (sockets, locks de threading, conexiones de BD) entre procesos — usar el patrón initializer del Pool.
• mp.Value con Lock para contadores atómicos; mp.Array o SharedMemory para arrays grandes que necesitan zero-copy.

Conversiones relacionadas

Conversiones frecuentes del catálogo:

• JPG a PNG
• MP4 a MP3
• PDF a DOCX
• PNG a JPG
• MP3 a WAV
• DOCX a PDF