Productor-consumidor con semáforos

1. El problema

El problema del productor-consumidor (también llamado problema del búfer acotado) consiste en coordinar dos tipos de procesos: los productores que generan datos y los consumidores que los procesan. Los datos se almacenan en un buffer acotado (de capacidad fija). El productor espera si el buffer está lleno; el consumidor espera si está vacío.

El problema tiene dos desafíos:

• Exclusión mutua: el acceso al buffer (insertar/extraer) debe ser mutuamente excluyente.
• Sincronización: el productor debe esperar si el buffer está lleno; el consumidor debe esperar si está vacío.

El problema fue propuesto por Dijkstra en 1965 y es uno de los problemas clásicos de sincronización.

2. El buffer acotado

El buffer acotado es un array circular de capacidad fija. Los productores insertan al final; los consumidores extraen del principio.

#define CAPACIDAD 10

typedef struct {
    int buffer[CAPACIDAD];
    int frente;
    int fondo;
    int tamano;
} buffer_t;

El buffer es circular: cuando se llega al final, se vuelve al principio. El tamaño indica cuántos elementos hay actualmente.

3. Los tres semáforos

La solución con semáforos usa tres semáforos:

Los semáforos vacios y llenos son la clave: vacios empieza en la capacidad (todos los espacios vacíos) y llenos empieza en 0 (ningún elemento). A medida que se producen y consumen, los contadores se ajustan.

La relación invariante es: vacios + llenos = CAPACIDAD. Esta invariante se mantiene siempre.

4. El productor

El productor genera un dato, espera a que haya un espacio vacío, adquiere el mutex, inserta el dato, libera el mutex y señala que hay un elemento más:

void *productor(void *arg) {
    for (int i = 0; i < N; i++) {
        int dato = producir();  /* generar dato */

        sem_wait(&vacios);       /* esperar espacio vacío */
        pthread_mutex_lock(&mutex);
        insertar(buffer, dato);  /* insertar en el buffer */
        pthread_mutex_unlock(&mutex);
        sem_post(&llenos);       /* señalar que hay un elemento */
    }
    return NULL;
}

El productor espera en sem_wait(&vacios) si el buffer está lleno. Cuando un consumidor extrae un elemento, hace sem_post(&vacios), despertando al productor.

5. El consumidor

El consumidor espera a que haya un elemento, adquiere el mutex, extrae el dato, libera el mutex y señala que hay un espacio vacío:

void *consumidor(void *arg) {
    for (int i = 0; i < N; i++) {
        sem_wait(&llenos);       /* esperar elemento */
        pthread_mutex_lock(&mutex);
        int dato = extraer(buffer);  /* extraer del buffer */
        pthread_mutex_unlock(&mutex);
        sem_post(&vacios);       /* señalar que hay espacio */
        consumir(dato);          /* procesar dato */
    }
    return NULL;
}

El consumidor espera en sem_wait(&llenos) si el buffer está vacío. Cuando un productor inserta un elemento, hace sem_post(&llenos), despertando al consumidor.

6. Por qué el orden de los wait importa

El orden de los sem_wait es crítico. El productor debe hacer sem_wait(&vacios) antes de pthread_mutex_lock, y el consumidor debe hacer sem_wait(&llenos) antes de pthread_mutex_lock.

La razón es: si se bloquea el mutex primero y luego se espera en el semáforo, se puede producir un deadlock.

7. El deadlock por orden incorrecto

Veamos qué ocurre si el orden es incorrecto:

/* Productor INCORRECTO. */
void *productor_malo(void *arg) {
    for (int i = 0; i < N; i++) {
        int dato = producir();
        pthread_mutex_lock(&mutex);      /* bloquea el mutex primero */
        sem_wait(&vacios);              /* espera espacio (puede bloquearse) */
        insertar(buffer, dato);
        pthread_mutex_unlock(&mutex);
        sem_post(&llenos);
    }
    return NULL;
}

/* Consumidor INCORRECTO. */
void *consumidor_malo(void *arg) {
    for (int i = 0; i < N; i++) {
        pthread_mutex_lock(&mutex);      /* bloquea el mutex primero */
        sem_wait(&llenos);              /* espera elemento (puede bloquearse) */
        int dato = extraer(buffer);
        pthread_mutex_unlock(&mutex);
        sem_post(&vacios);
        consumir(dato);
    }
    return NULL;
}

Escenario de deadlock:

1. El buffer está lleno. vacios = 0, llenos = CAPACIDAD. estado
2. El productor bloquea el mutex y luego hace sem_wait(&vacios). Se bloquea (vacios = 0). Tiene el mutex. bloqueo
3. El consumidor intenta bloquear el mutex. Se bloquea (el mutex está ocupado). bloqueo
4. Deadlock: el productor espera a que vacios > 0, pero el consumidor no puede hacer post(vacios) porque no tiene el mutex. El consumidor espera el mutex, pero el productor no lo libera. deadlock

Este deadlock ocurre porque el productor bloquea el mutex y luego se bloquea en el semáforo, reteniendo el mutex mientras espera. El consumidor no puede liberar el mutex porque no puede adquirirlo.

8. Código completo

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

#define CAPACIDAD 5
#define NUM_PRODUCTORES 2
#define NUM_CONSUMIDORES 2
#define ITEMS_POR_PRODUCTOR 10

int buffer[CAPACIDAD];
int frente = 0;
int fondo = 0;

sem_t mutex;
sem_t vacios;
sem_t llenos;

void insertar(int dato) {
    buffer[fondo] = dato;
    fondo = (fondo + 1) % CAPACIDAD;
}

int extraer(void) {
    int dato = buffer[frente];
    frente = (frente + 1) % CAPACIDAD;
    return dato;
}

void *productor(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < ITEMS_POR_PRODUCTOR; i++) {
        int dato = id * 100 + i;  /* generar dato */
        sem_wait(&vacios);          /* esperar espacio (correcto: antes del mutex) */
        pthread_mutex_lock(&mutex);
        insertar(dato);
        printf("Productor %d: produjo %d\n", id, dato);
        pthread_mutex_unlock(&mutex);
        sem_post(&llenos);          /* señalar elemento */
    }
    return NULL;
}

void *consumidor(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < (NUM_PRODUCTORES * ITEMS_POR_PRODUCTOR) / NUM_CONSUMIDORES; i++) {
        sem_wait(&llenos);          /* esperar elemento (correcto: antes del mutex) */
        pthread_mutex_lock(&mutex);
        int dato = extraer();
        printf("Consumidor %d: consumió %d\n", id, dato);
        pthread_mutex_unlock(&mutex);
        sem_post(&vacios);          /* señalar espacio */
    }
    return NULL;
}

int main(void) {
    sem_init(&mutex, 0, 1);
    sem_init(&vacios, 0, CAPACIDAD);
    sem_init(&llenos, 0, 0);

    pthread_t prods[NUM_PRODUCTORES];
    pthread_t cons[NUM_CONSUMIDORES];
    int ids_prod[NUM_PRODUCTORES];
    int ids_cons[NUM_CONSUMIDORES];

    for (int i = 0; i < NUM_PRODUCTORES; i++) {
        ids_prod[i] = i;
        pthread_create(&prods[i], NULL, productor, &ids_prod[i]);
    }
    for (int i = 0; i < NUM_CONSUMIDORES; i++) {
        ids_cons[i] = i;
        pthread_create(&cons[i], NULL, consumidor, &ids_cons[i]);
    }

    for (int i = 0; i < NUM_PRODUCTORES; i++) pthread_join(prods[i], NULL);
    for (int i = 0; i < NUM_CONSUMIDORES; i++) pthread_join(cons[i], NULL);

    sem_destroy(&mutex);
    sem_destroy(&vacios);
    sem_destroy(&llenos);
    return 0;
}

9. Propiedades

10. Simulador interactivo

Este simulador muestra el buffer acotado con productores y consumidores:

11. Resumen

• El productor-consumidor usa tres semáforos: mutex, vacíos y llenos.
• El productor espera espacio (vacios) y luego inserta (mutex).
• El consumidor espera elemento (llenos) y luego extrae (mutex).
• El orden de los wait es crítico: sem_wait antes de mutex_lock.
• El orden incorrecto causa deadlock.
• La invariante es: vacios + llenos = CAPACIDAD.
• Los semáforos vacíos y llenos cuentan recursos; el mutex protege el buffer.

12. Preguntas de repaso

Intenta responder cada una antes de abrir la respuesta.

Para practicar

1. Implementa el productor-consumidor y ejecuta con 2 productores y 2 consumidores.
2. Intercambia el orden de los wait (mutex antes de semáforo) y observa el deadlock.
3. Modifica la capacidad del buffer y observa el efecto.
4. Añade más productores que consumidores y observa.
5. Usa sem_trywait para verificar si hay espacio/elementos sin bloquear.
6. Investiga cómo se implementa el productor-consumidor en Java (BlockingQueue).