Proyecto: un pool de hilos - Concurrencia y Sincronización

1. Objetivo del proyecto

El objetivo es construir un pool de hilos que:

Crear un número fijo de hilos trabajadores al inicio.
Recibir tareas (funciones) de los clientes.
Los trabajadores procesan las tareas de la cola.
Apagado ordenado: esperar a que todas las tareas terminen antes de destruir el pool.

El pool de hilos es un patrón fundamental en la programación concurrente: se usa en servidores web, bases de datos, y cualquier sistema que procese muchas tareas pequeñas.

2. Arquitectura del pool

El pool tiene estas partes:

Cola de tareas: una cola acotada (o no acotada) de funciones a ejecutar.
Trabajadores: hilos que toman tareas de la cola y las ejecutan.
Mutex: protege la cola de tareas.
Variable de condición: los trabajadores esperan a que haya tareas.
Flag de apagado: indica que el pool se está cerrando.

3. La cola de tareas

La cola de tareas es una cola acotada (o no acotada) de funciones. Cada tarea es una función con un argumento:

tarea

typedef void (*tarea_func_t)(void *arg);

typedef struct tarea {
    tarea_func_t func;
    void *arg;
    struct tarea *siguiente;
} tarea_t;

4. Los trabajadores

Cada trabajador es un hilo que ejecuta un bucle: toma una tarea de la cola, la ejecuta y repite. Si no hay tareas, espera en la variable de condición:

trabajador

void *trabajador(void *arg) {
    pool_t *pool = (pool_t *)arg;
    while (1) {
        pthread_mutex_lock(&pool->mutex);
        while (pool->cola == NULL && !pool->apagado) {
            pthread_cond_wait(&pool->cond, &pool->mutex);
        }
        if (pool->apagado && pool->cola == NULL) {
            pthread_mutex_unlock(&pool->mutex);
            break;  /* salir */
        }
        tarea_t *tarea = pool->cola;
        pool->cola = tarea->siguiente;
        pthread_mutex_unlock(&pool->mutex);

        /* Ejecutar la tarea (fuera del mutex). */
        tarea->func(tarea->arg);
        free(tarea);
    }
    return NULL;
}

5. Enviar tareas

Para enviar una tarea al pool, se añade a la cola y se señala a los trabajadores:

submit

void pool_submit(pool_t *pool, tarea_func_t func, void *arg) {
    tarea_t *tarea = malloc(sizeof(tarea_t));
    tarea->func = func;
    tarea->arg = arg;
    tarea->siguiente = NULL;

    pthread_mutex_lock(&pool->mutex);
    /* Añadir al final de la cola. */
    if (pool->cola == NULL) {
        pool->cola = tarea;
    } else {
        tarea_t *actual = pool->cola;
        while (actual->siguiente) actual = actual->siguiente;
        actual->siguiente = tarea;
    }
    pthread_cond_signal(&pool->cond);  /* despertar a un trabajador */
    pthread_mutex_unlock(&pool->mutex);
}

6. Apagado ordenado

El apagado ordenado consiste en:

Marcar el pool como apagado: pool->apagado = 1. marcar
Despertar a todos los trabajadores: pthread_cond_broadcast. despertar
Esperar a que terminen: pthread_join para cada trabajador. esperar
Liberar recursos: destruir el mutex y la condición. liberar

Los trabajadores que están procesando una tarea la terminan antes de salir. Los que están esperando salen inmediatamente.

7. Código completo

thread_pool.c

#define _POSIX_C_SOURCE 200809L
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

typedef void (*tarea_func_t)(void *arg);

typedef struct tarea {
    tarea_func_t func;
    void *arg;
    struct tarea *siguiente;
} tarea_t;

typedef struct {
    tarea_t *cola;
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    int num_trabajadores;
    pthread_t *trabajadores;
    int apagado;
} pool_t;

/* Trabajador. */
static void *trabajador(void *arg) {
    pool_t *pool = (pool_t *)arg;
    while (1) {
        pthread_mutex_lock(&pool->mutex);
        while (pool->cola == NULL && !pool->apagado) {
            pthread_cond_wait(&pool->cond, &pool->mutex);
        }
        if (pool->apagado && pool->cola == NULL) {
            pthread_mutex_unlock(&pool->mutex);
            break;
        }
        tarea_t *tarea = pool->cola;
        pool->cola = tarea->siguiente;
        pthread_mutex_unlock(&pool->mutex);

        tarea->func(tarea->arg);
        free(tarea);
    }
    return NULL;
}

/* Crear el pool. */
void pool_init(pool_t *pool, int num_trabajadores) {
    pool->cola = NULL;
    pool->num_trabajadores = num_trabajadores;
    pool->apagado = 0;
    pthread_mutex_init(&pool->mutex, NULL);
    pthread_cond_init(&pool->cond, NULL);
    pool->trabajadores = malloc(num_trabajadores * sizeof(pthread_t));
    for (int i = 0; i < num_trabajadores; i++) {
        pthread_create(&pool->trabajadores[i], NULL, trabajador, pool);
    }
}

/* Enviar una tarea. */
void pool_submit(pool_t *pool, tarea_func_t func, void *arg) {
    tarea_t *tarea = malloc(sizeof(tarea_t));
    tarea->func = func;
    tarea->arg = arg;
    tarea->siguiente = NULL;

    pthread_mutex_lock(&pool->mutex);
    if (pool->cola == NULL) {
        pool->cola = tarea;
    } else {
        tarea_t *actual = pool->cola;
        while (actual->siguiente) actual = actual->siguiente;
        actual->siguiente = tarea;
    }
    pthread_cond_signal(&pool->cond);
    pthread_mutex_unlock(&pool->mutex);
}

/* Apagar el pool. */
void pool_shutdown(pool_t *pool) {
    pthread_mutex_lock(&pool->mutex);
    pool->apagado = 1;
    pthread_cond_broadcast(&pool->cond);
    pthread_mutex_unlock(&pool->mutex);

    for (int i = 0; i < pool->num_trabajadores; i++) {
        pthread_join(pool->trabajadores[i], NULL);
    }

    /* Liberar tareas pendientes. */
    tarea_t *actual = pool->cola;
    while (actual) {
        tarea_t *sig = actual->siguiente;
        free(actual);
        actual = sig;
    }

    pthread_mutex_destroy(&pool->mutex);
    pthread_cond_destroy(&pool->cond);
    free(pool->trabajadores);
}

8. Propiedades

Propiedad	¿La cumple?	Comentario
Thread-safe	Sí	El mutex protege la cola.
Apagado ordenado	Sí	Espera a que todas las tareas terminen.
Reutilizable	Sí	Los trabajadores se reutilizan para múltiples tareas.
Escalable	Sí	Se puede ajustar el número de trabajadores.
Sin deadlock	Sí	No hay ciclos de espera.

9. Simulador interactivo

Este simulador muestra cómo un pool de hilos procesa tareas:

Simulador · pool de hilos

Trabajadores Tareas

Paso	Trabajador	Tarea	Estado

10. Resumen

Un pool de hilos tiene una cola de tareas y trabajadores reutilizables.
Los trabajadores toman tareas de la cola y las ejecutan.
El mutex protege la cola; la condición despierta a los trabajadores.
El apagado ordenado espera a que todas las tareas terminen.
Es un patrón fundamental en la programación concurrente.

11. Preguntas de repaso

Intenta responder cada una antes de abrir la respuesta.

1. ¿Qué es un pool de hilos?

Un conjunto de hilos trabajadores que procesan tareas de una cola compartida.

2. ¿Para qué se usa?

Para procesar muchas tareas pequeñas eficientemente, reutilizando hilos.

3. ¿Qué protege el mutex?

La cola de tareas.

4. ¿Qué despierta a los trabajadores?

pthread_cond_signal cuando se añade una tarea.

5. ¿Qué ocurre en el apagado ordenado?

Se marcan como apagado, se despiertan todos, se espera a que terminen.

6. ¿Por qué se ejecuta la tarea fuera del mutex?

Para no bloquear la cola mientras se ejecuta la tarea.

7. ¿Qué ocurre si se envía una tarea después del apagado?

No se procesa (el pool está cerrado).

8. ¿Es thread-safe el pool?

Sí, el mutex protege la cola.

9. ¿Qué patrón de sincronización usa el pool?

Productor-consumidor (la cola es el buffer, los clientes son productores, los trabajadores son consumidores).

10. ¿Dónde se usa el pool de hilos en la práctica?

En servidores web, bases de datos y cualquier sistema que procese muchas tareas.

Para practicar

Implementa el pool de hilos y pruébalo con varias tareas.
Añade un contador de tareas completadas.
Implementa un pool con cola acotada (bloquear si está llena).
Usa ThreadSanitizer para verificar que no hay carreras.
Compara el pool con crear un hilo por tarea.
Investiga cómo se implementa un thread pool en Java (ExecutorService).