Concurrencia en Python: Ejecuta Tareas en Paralelo con `threading` y `multiprocessing`

La concurrencia es un concepto fundamental en la programación que permite que un programa maneje múltiples tareas aparentemente al mismo tiempo. En Python, esto se puede lograr principalmente a través de hilos (threads) y procesos (processes), cada uno con sus propias ventajas y desventajas. Entender cuándo y cómo usar cada uno es crucial para escribir aplicaciones eficientes y escalables.

Este tutorial te guiará a través de los módulos threading y multiprocessing de Python, desglosando sus mecanismos internos, el famoso Global Interpreter Lock (GIL), y cómo aplicarlos en escenarios prácticos.

🚀 ¿Qué es la Concurrencia y por qué es Importante?

La concurrencia se refiere a la capacidad de un sistema para manejar múltiples tareas en progreso al mismo tiempo. No significa necesariamente que se estén ejecutando exactamente al mismo instante (paralelismo), sino que el sistema puede avanzar en varias tareas de forma intercalada, dando la impresión de simultaneidad.

Concurrencia vs. Paralelismo

Es importante diferenciar estos dos términos:

• Concurrencia: Tratar con muchas cosas a la vez. Un sistema concurrente puede tener múltiples tareas en curso, pero solo una se está ejecutando en un momento dado (en el caso de un solo núcleo de CPU, como los hilos en Python debido al GIL).
• Paralelismo: Hacer muchas cosas a la vez. Un sistema paralelo ejecuta múltiples tareas simultáneamente, típicamente usando múltiples núcleos de CPU o múltiples máquinas.

¿Por qué necesitamos concurrencia?

La concurrencia es vital para:

• Mejorar la capacidad de respuesta: Las aplicaciones pueden seguir siendo interactivas mientras realizan tareas pesadas en segundo plano (ej. UI, servidores web).
• Optimizar el rendimiento: Aprovechar mejor los recursos del sistema, especialmente en tareas de E/S intensivas.
• Gestión de tareas asíncronas: Manejar múltiples solicitudes o eventos sin bloquear la aplicación principal.

🧵 Hilos (Threads) con el Módulo threading

Los hilos son unidades de ejecución más ligeras que los procesos. Comparten el mismo espacio de memoria dentro de un único proceso. En Python, el módulo threading nos permite trabajar con hilos.

El Global Interpreter Lock (GIL) en Python

Antes de sumergirnos en los hilos, es crucial entender el Global Interpreter Lock (GIL). El GIL es un mecanismo que garantiza que solo un hilo de Python pueda ejecutar bytecode de Python a la vez dentro del mismo intérprete. Esto significa que, incluso en un sistema con múltiples núcleos, los hilos de Python no pueden lograr paralelismo real para tareas que consumen intensivamente la CPU.

Creación y Ejecución Básica de Hilos

El módulo threading proporciona una forma sencilla de crear y gestionar hilos. La forma más común es crear una subclase de threading.Thread o pasar una función a su constructor.

Ejemplo 1: Creando hilos con una función

import threading
import time

def tarea_sencilla(nombre):
    print(f"Hilo {nombre}: Iniciado")
    time.sleep(2) # Simula una tarea que toma tiempo (ej. E/S)
    print(f"Hilo {nombre}: Finalizado")

print("Programa principal: Iniciando hilos...")

hilo1 = threading.Thread(target=tarea_sencilla, args=("uno",))
hilo2 = threading.Thread(target=tarea_sencilla, args=("dos",))

hilo1.start() # Inicia la ejecución del hilo1
hilo2.start() # Inicia la ejecución del hilo2

# Esperar a que los hilos terminen
hilo1.join()
hilo2.join()

print("Programa principal: Todos los hilos han terminado.")

Explicación:

1. Definimos tarea_sencilla que simula un trabajo. time.sleep() libera el GIL, permitiendo que otros hilos se ejecuten.
2. Creamos dos objetos threading.Thread, pasando la función tarea_sencilla al parámetro target y una tupla de argumentos al parámetro args.
3. hilo.start() inicia la ejecución del hilo. El sistema operativo asigna un hilo nativo y Python empieza a ejecutar la función target en ese hilo.
4. hilo.join() bloquea el hilo principal hasta que el hilo en cuestión (hilo1 o hilo2) haya terminado su ejecución. Es fundamental para asegurar que el programa principal no termine antes que sus hilos.

Ejemplo 2: Creando hilos con una clase

import threading
import time

class MiHilo(threading.Thread):
    def __init__(self, nombre):
        super().__init__()
        self.nombre = nombre

    def run(self):
        print(f"Hilo {self.nombre}: Iniciado")
        time.sleep(3)
        print(f"Hilo {self.nombre}: Finalizado")

print("Programa principal: Iniciando hilos con clases...")

hilo_clase1 = MiHilo("clase-uno")
hilo_clase2 = MiHilo("clase-dos")

hilo_clase1.start()
hilo_clase2.start()

hilo_clase1.join()

hilo_clase2.join()

print("Programa principal: Todos los hilos con clases han terminado.")

Explicación:

1. Creamos una clase MiHilo que hereda de threading.Thread.
2. Sobrescribimos el método run(), que es el que será ejecutado cuando el hilo se inicie.
3. El resto del proceso es similar: se instancian los hilos y se llaman a start() y join().

Sincronización de Hilos: Evitando Condiciones de Carrera

Cuando múltiples hilos acceden y modifican los mismos datos compartidos, pueden surgir problemas como las condiciones de carrera, donde el resultado final depende del orden impredecible de ejecución de los hilos. Para evitar esto, necesitamos mecanismos de sincronización.

Locks (Cerraduras)

Un Lock es el mecanismo de sincronización más básico. Un hilo adquiere el lock antes de acceder a un recurso compartido y lo libera después. Si otro hilo intenta adquirir un lock que ya está en uso, se bloquea hasta que el lock sea liberado.

import threading
import time

# Recurso compartido
contador = 0
# Objeto Lock
lock = threading.Lock()

def incrementar(nombre):
    global contador
    for _ in range(100000):
        lock.acquire() # Adquiere el lock
        try:
            contador += 1
        finally:
            lock.release() # Libera el lock (siempre en un finally)
    print(f"Hilo {nombre}: Contador finalizado en {contador}")

print("Programa principal: Iniciando hilos con locks...")

hilos_incremento = []
for i in range(5):
    hilo = threading.Thread(target=incrementar, args=(f"H{i}",))
    hilos_incremento.append(hilo)
    hilo.start()

for hilo in hilos_incremento:
    hilo.join()

print(f"Programa principal: Contador final: {contador}")

Sin el lock, el valor final de contador sería inconsistente y menor a 500,000 debido a las condiciones de carrera.

RLock (Reentrant Lock)

Un RLock es similar a un Lock, pero puede ser adquirido múltiples veces por el mismo hilo. Esto es útil en situaciones donde una función que ya ha adquirido un lock necesita llamar a otra función que también intenta adquirir el mismo lock.

Semáforos (Semaphore)

Un Semaphore es un contador interno que puede ser decrementado o incrementado. Permite que un número limitado de hilos accedan a un recurso simultáneamente, a diferencia de un Lock que solo permite uno.

import threading
import time

# Permitir solo 3 hilos a la vez en la sección crítica
semaforo = threading.Semaphore(3)

def trabajar_con_recurso(nombre):
    print(f"Hilo {nombre}: Intentando adquirir el semáforo...")
    semaforo.acquire()
    try:
        print(f"Hilo {nombre}: Adquirió el semáforo y está trabajando...")
        time.sleep(2) # Simula trabajo en recurso
        print(f"Hilo {nombre}: Terminó de trabajar.")
    finally:
        semaforo.release()

print("Programa principal: Iniciando hilos con semáforo...")

hilos_semaforo = []
for i in range(10):
    hilo = threading.Thread(target=trabajar_con_recurso, args=(f"S{i}",))
    hilos_semaforo.append(hilo)
    hilo.start()

for hilo in hilos_semaforo:
    hilo.join()

print("Programa principal: Todos los hilos con semáforo han terminado.")

Otras Primitivas de Sincronización

• Eventos (Event): Permiten la comunicación entre hilos. Un hilo puede esperar a que un evento se set() y otro puede clear()lo.
• Condiciones (Condition): Más avanzadas que los eventos, permiten a los hilos esperar hasta que se cumpla una condición específica y ser notificados cuando cambie.
• Barreras (Barrier): Permiten que un grupo de hilos se esperen mutuamente hasta que todos hayan alcanzado un punto común.

🏋️ Procesos (Processes) con el Módulo multiprocessing

Para superar las limitaciones del GIL y lograr un paralelismo real en tareas intensivas de CPU, Python ofrece el módulo multiprocessing. Los procesos son mucho más pesados que los hilos; cada proceso tiene su propio espacio de memoria y su propio intérprete de Python, lo que significa que el GIL no es una limitación entre procesos.

Creación y Ejecución Básica de Procesos

El módulo multiprocessing tiene una API muy similar a threading, lo que facilita su uso si ya estás familiarizado con hilos.

Ejemplo 3: Creando procesos con una función

import multiprocessing
import time
import os

def tarea_pesada(nombre):
    print(f"Proceso {nombre} (PID: {os.getpid()}): Iniciado")
    suma = 0
    for i in range(10**7): # Tarea intensiva de CPU
        suma += i
    print(f"Proceso {nombre}: Suma calculada ({suma}). Finalizado")

print("Programa principal (PID: {os.getpid()}): Iniciando procesos...")

procesos = []
for i in range(2):
    p = multiprocessing.Process(target=tarea_pesada, args=(f"P{i}",))
    procesos.append(p)
    p.start()

for p in procesos:
    p.join()

print("Programa principal: Todos los procesos han terminado.")

Explicación:

1. Importamos multiprocessing y os para obtener el ID del proceso (PID).
2. La tarea_pesada ahora es una tarea intensiva de CPU para demostrar el paralelismo.
3. Creamos objetos multiprocessing.Process de manera similar a threading.Thread.
4. p.start() inicia un nuevo proceso del sistema operativo. Cada proceso tiene su propio GIL, permitiendo que la tarea_pesada se ejecute en paralelo real en diferentes núcleos de CPU.
5. p.join() espera a que el proceso termine.

Comunicación entre Procesos

Dado que los procesos tienen espacios de memoria separados, la comunicación y el intercambio de datos entre ellos son más complejos que con los hilos. multiprocessing ofrece varias herramientas para esto.

Colas (Queue)

Las colas son una forma segura de pasar mensajes y objetos entre procesos. Son ideales para productores/consumidores.

import multiprocessing
import time

def productor(cola):
    for i in range(5):
        mensaje = f"Mensaje {i}"
        print(f"Productor: Enviando {mensaje}")
        cola.put(mensaje)
        time.sleep(1)
    cola.put("FIN") # Señal para el consumidor

def consumidor(cola):
    while True:
        mensaje = cola.get()
        if mensaje == "FIN":
            break
        print(f"Consumidor: Recibido {mensaje}")
        time.sleep(0.5)

print("Programa principal: Iniciando comunicación con cola...")

cola = multiprocessing.Queue()

p_productor = multiprocessing.Process(target=productor, args=(cola,))
p_consumidor = multiprocessing.Process(target=consumidor, args=(cola,))

p_productor.start()
p_consumidor.start()

p_productor.join()
p_consumidor.join()

print("Programa principal: Comunicación con cola finalizada.")

Pipes (Pipe)

Un Pipe es un canal bidireccional de comunicación entre dos procesos. Puedes enviar y recibir objetos Python a través de sus extremos.

import multiprocessing
import time

def remitente(conn):
    for i in range(3):
        msg = f"Dato {i}"
        print(f"Remitente: Enviando {msg}")
        conn.send(msg)
        time.sleep(1)
    conn.close()

def receptor(conn):
    while True:
        try:
            msg = conn.recv()
            print(f"Receptor: Recibido {msg}")
        except EOFError:
            # El otro extremo se cerró
            break
        time.sleep(0.5)
    conn.close()

print("Programa principal: Iniciando comunicación con pipe...")

parent_conn, child_conn = multiprocessing.Pipe()

p_remitente = multiprocessing.Process(target=remitente, args=(child_conn,))
p_receptor = multiprocessing.Process(target=receptor, args=(parent_conn,))

p_remitente.start()
p_receptor.start()

p_remitente.join()
p_receptor.join()

print("Programa principal: Comunicación con pipe finalizada.")

Sincronización entre Procesos

Al igual que con los hilos, los procesos también necesitan mecanismos de sincronización para evitar condiciones de carrera al acceder a recursos compartidos. multiprocessing proporciona sus propias versiones de Lock, Semaphore, Event, etc., que funcionan entre procesos.

Locks para Procesos

import multiprocessing
import time

# Recurso compartido (Manager para datos compartidos entre procesos)
# Es crucial usar Manager para compartir objetos complejos o mutables
# de forma segura entre procesos.
manager = multiprocessing.Manager()
contador_compartido = manager.Value('i', 0) # 'i' para entero
lock_proceso = multiprocessing.Lock()

def incrementar_proceso(nombre, contador, lock):
    for _ in range(100000):
        lock.acquire()
        try:
            contador.value += 1
        finally:
            lock.release()
    print(f"Proceso {nombre}: Contador finalizado en {contador.value}")

print("Programa principal: Iniciando procesos con locks...")

procesos_incremento = []
for i in range(4):
    p = multiprocessing.Process(target=incrementar_proceso, args=(f"P{i}", contador_compartido, lock_proceso))
    procesos_incremento.append(p)
    p.start()

for p in procesos_incremento:
    p.join()

print(f"Programa principal: Contador final (compartido): {contador_compartido.value}")

Pools de Procesos (Pool)

El objeto Pool es una de las características más potentes de multiprocessing. Permite enviar un conjunto de tareas a un grupo de procesos trabajadores, distribuyendo la carga de trabajo automáticamente. Esto es ideal para aplicar una función a muchos elementos de forma paralela.

import multiprocessing
import time

def funcion_cuadrado(numero):
    time.sleep(0.1) # Simula un cálculo
    return numero * numero

if __name__ == '__main__': # Esencial para Windows y algunos sistemas Unix
    print("Programa principal: Iniciando pool de procesos...")
    numeros = range(10)

    # Crear un pool con el número de procesos igual al número de CPU o menos
    # (por defecto, usa os.cpu_count())
    with multiprocessing.Pool(processes=4) as pool:
        # map() aplica la función a cada elemento de la iterable y devuelve los resultados
        resultados = pool.map(funcion_cuadrado, numeros)

    print(f"Números originales: {list(numeros)}")
    print(f"Resultados cuadrados: {resultados}")
    print("Programa principal: Pool de procesos finalizado.")

apply_async y map_async

Además de map, Pool ofrece métodos asíncronos como apply_async (para una sola función) y map_async (para iterables) que devuelven un objeto AsyncResult. Puedes usar get() en este objeto para recuperar el resultado cuando esté disponible, o ready()/successful() para verificar el estado.

⚖️ Hilos vs. Procesos: ¿Cuándo usar cada uno?

La elección entre hilos y procesos depende en gran gran medida del tipo de tarea que deseas paralelizar.

✨ Ejemplos de Uso en la Vida Real

🌐 Servidor Web Simple con Hilos

Un servidor web es un caso de uso clásico para hilos. Cada solicitud de cliente puede ser manejada por un hilo separado, permitiendo que el servidor siga aceptando nuevas conexiones sin bloquearse.

import threading
import socket

def manejar_cliente(conn, addr):
    print(f"Conexión desde {addr} manejada por hilo {threading.current_thread().name}")
    try:
        data = conn.recv(1024).decode()
        print(f"Recibido de {addr}: {data.strip()}")
        respuesta = "HTTP/1.1 200 OK\nContent-Type: text/plain\n\nHola desde el servidor concurrente!\n"
        conn.sendall(respuesta.encode())
    except Exception as e:
        print(f"Error al manejar cliente {addr}: {e}")
    finally:
        conn.close()
        print(f"Conexión con {addr} cerrada por hilo {threading.current_thread().name}")

def iniciar_servidor(host, port):
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind((host, port))
    server_socket.listen(5) # Aceptar hasta 5 conexiones en cola
    print(f"Servidor escuchando en {host}:{port}...")

    while True:
        conn, addr = server_socket.accept() # Bloquea hasta que llega una conexión
        print(f"Nueva conexión de {addr}")
        # Crear un nuevo hilo para manejar al cliente
        client_thread = threading.Thread(target=manejar_cliente, args=(conn, addr))
        client_thread.start()

if __name__ == '__main__':
    HOST = '127.0.0.1'
    PORT = 8080
    # Para probar:
    # Abre un navegador y ve a http://127.0.0.1:8080
    # O usa curl en la terminal: curl http://127.0.0.1:8080
    # Prueba con múltiples requests rápidos para ver la concurrencia.
    iniciar_servidor(HOST, PORT)

📊 Procesamiento de Datos con Procesos

Imagina que tienes una lista enorme de números y necesitas realizar una operación costosa en cada uno. multiprocessing.Pool es perfecto para esto.

import multiprocessing
import time
import random

def calcular_operacion_compleja(numero):
    # Simula una operación intensiva de CPU
    resultado = 0
    for _ in range(1000000):
        resultado += numero * random.random()
    return resultado

if __name__ == '__main__':
    datos = [random.randint(1, 100) for _ in range(20)] # 20 números aleatorios

    print(f"Datos a procesar: {datos[:5]}...{datos[-5:]}")

    start_time_sync = time.time()
    resultados_sync = [calcular_operacion_compleja(d) for d in datos]
    end_time_sync = time.time()
    print(f"\nProcesamiento Sincrónico tomó {end_time_sync - start_time_sync:.2f} segundos.")

    start_time_pool = time.time()
    with multiprocessing.Pool(processes=None) as pool: # processes=None usa os.cpu_count()
        resultados_pool = pool.map(calcular_operacion_compleja, datos)
    end_time_pool = time.time()
    print(f"Procesamiento con Pool tomó {end_time_pool - start_time_pool:.2f} segundos.")

    # Puedes verificar que los resultados son los mismos (ordenados)
    # print(f"Resultados sincrónicos (primeros 5): {resultados_sync[:5]}")
    # print(f"Resultados con Pool (primeros 5): {resultados_pool[:5]}")

    print("\nComparación de tiempos:")
    print(f"Mejora de rendimiento: {((end_time_sync - start_time_pool) / (end_time_pool - start_time_pool)):.2f}x")

Observa la significativa mejora en tiempo al usar multiprocessing.Pool para tareas CPU-bound.

🛠️ Buenas Prácticas y Consideraciones

• Evita el uso excesivo de hilos/procesos: Crear demasiados hilos o procesos puede generar más overhead que beneficios, ya que el sistema operativo pasa más tiempo gestionando el cambio de contexto que ejecutando el trabajo real.
• Diseño defensivo: Siempre usa mecanismos de sincronización (locks, semáforos) cuando accedas a recursos compartidos para evitar condiciones de carrera y deadlocks.
• Manejo de excepciones: Asegúrate de manejar las excepciones dentro de tus funciones de hilo/proceso. Un error no capturado en un hilo puede terminar silenciosamente, mientras que en un proceso puede causar su terminación.
• Datos compartidos entre procesos: Utiliza multiprocessing.Manager o Queue/Pipe para intercambiar datos de forma segura entre procesos. ¡Nunca intentes pasar objetos complejos directamente a un nuevo proceso a través de argumentos si estos objetos no son serializables o no se manejan específicamente para el intercambio entre procesos!
• if __name__ == '__main__':: Es crucial para asegurar que tus scripts de multiprocessing funcionen correctamente, especialmente en Windows.
• Depuración: Depurar aplicaciones concurrentes puede ser un desafío. Usa logging extensivamente para entender el flujo de ejecución y los valores de las variables en cada hilo/proceso.
• Alternativas asíncronas (asyncio): Para tareas I/O-bound y con una estructura de eventos bien definida, asyncio es una alternativa moderna y muy eficiente en Python 3.5+. Permite concurrencia en un solo hilo sin el overhead de la creación de hilos del sistema operativo.

❓ Preguntas Frecuentes (FAQ)

🏁 Conclusión

La concurrencia en Python es una herramienta poderosa que te permite escribir aplicaciones más eficientes y responsivas. Comprender las diferencias entre hilos y procesos, así como cuándo aplicar cada uno, es fundamental. Los hilos (threading) son excelentes para tareas de E/S intensivas, mientras que los procesos (multiprocessing) son la solución para exprimir todo el potencial de tu CPU en tareas de cálculo intensivo, superando la limitación del GIL. Siempre considera las buenas prácticas y los mecanismos de sincronización para construir sistemas robustos y libres de errores.

¡Ahora tienes las bases para empezar a construir aplicaciones concurrentes en Python! Experimenta con los ejemplos y adapta estos conceptos a tus propios proyectos.