Producer-Consumer

Um lado gera itens de trabalho, o outro os processa, e entre eles há um buffer de tamanho fixo. Um servidor web coloca requisições numa fila e um pool de threads as atende. Um decodificador de vídeo enche um buffer de quadros que a placa de vídeo esvazia. O padrão é sempre o mesmo, e traz junto sempre os mesmos dois riscos: o produtor não pode escrever num buffer cheio, e o consumidor não pode ler de um buffer vazio. Se qualquer um dos dois ignorar isso, o programa corrompe dados ou lê lixo.

Contar quantas threads podem prosseguir é exatamente o que um semáforo faz, e aqui a gente precisa de dois. Um conta quantas posições livres o buffer ainda tem; o produtor faz P nele antes de escrever, e bloqueia se o buffer está cheio. O outro conta quantos itens prontos existem; o consumidor faz P nele antes de ler, e bloqueia se o buffer está vazio. Cada sem_post de um lado é o que eventualmente libera o outro.

Dois semáforos e um mutex

Se houvesse um só produtor e um só consumidor, os dois semáforos bastariam. Mas com vários produtores, dois deles podem passar pelo semáforo de posições livres ao mesmo tempo, e aí os dois vão escrever na mesma posição do buffer e avançar o mesmo índice, um por cima do outro. O semáforo garante que existe espaço; ele não garante que dois produtores não vão brigar pelo mesmo espaço. Essa briga pelo índice é uma seção crítica comum, e quem resolve seção crítica é o mutex.

Daí a divisão de trabalho que é o coração da solução: os semáforos coordenam (quantos podem entrar), o mutex protege (a manipulação dos índices). O Código 10-1 tem dois produtores e dois consumidores sobre um buffer de quatro posições:

#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>

#define TAM 4                 // tamanho do buffer
#define NP 2                  // produtores
#define NC 2                  // consumidores
#define POR_PROD 5            // itens que cada produtor gera
#define POR_CONS 5            // itens que cada consumidor consome

int buffer[TAM];
int in = 0, out = 0;

sem_t vazias;                 // posições livres
sem_t cheias;                 // itens prontos
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void *produtor(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < POR_PROD; i++) {
        int item = id * 100 + i;
        sem_wait(&vazias);            // espera uma posição livre
        pthread_mutex_lock(&mutex);   // protege os índices
        buffer[in] = item;
        in = (in + 1) % TAM;
        pthread_mutex_unlock(&mutex);
        sem_post(&cheias);            // agora tem mais um item pronto
    }
    return NULL;
}

void *consumidor(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < POR_CONS; i++) {
        sem_wait(&cheias);            // espera um item pronto
        pthread_mutex_lock(&mutex);
        int item = buffer[out];
        out = (out + 1) % TAM;
        pthread_mutex_unlock(&mutex);
        sem_post(&vazias);            // liberou uma posição
        printf("consumidor %d pegou %d\n", id, item);
    }
    return NULL;
}

int main(void) {
    sem_init(&vazias, 0, TAM);        // começa com TAM posições livres
    sem_init(&cheias, 0, 0);          // e zero itens prontos
    pthread_t p[NP], c[NC];
    int pid[NP], cid[NC];
    for (int i = 0; i < NP; i++) { pid[i]=i; pthread_create(&p[i],NULL,produtor,&pid[i]); }
    for (int i = 0; i < NC; i++) { cid[i]=i; pthread_create(&c[i],NULL,consumidor,&cid[i]); }
    for (int i = 0; i < NP; i++) pthread_join(p[i],NULL);
    for (int i = 0; i < NC; i++) pthread_join(c[i],NULL);
    printf("pronto\n");
    return 0;
}

$ cc main.c -o main -pthread
$ ./main
consumidor 0 pegou 0
consumidor 1 pegou 1
consumidor 1 pegou 3
consumidor 1 pegou 4
consumidor 1 pegou 100
consumidor 0 pegou 2
consumidor 1 pegou 101
consumidor 0 pegou 102
consumidor 0 pegou 103
consumidor 0 pegou 104
pronto

Os dez itens saem: os do produtor 0 (0 a 4) e os do produtor 1 (100 a 104), cada um consumido uma única vez, sem nenhum sumir nem aparecer duplicado. A ordem de quem pega o quê muda a cada execução, porque o escalonador decide quando cada thread roda, mas as duas invariantes se mantêm sempre: nunca um produtor escreve num buffer cheio, nunca um consumidor lê de um vazio.

Repare como as três primitivas se encaixam. Tira o semáforo vazias e um produtor rápido passa por cima de itens que ninguém consumiu ainda. Tira o cheias e um consumidor lê posições que nunca foram escritas. Tira o mutex e dois produtores corrompem o índice in. Cada uma cobre um buraco que as outras duas não cobrem, e é essa complementaridade que faz a solução funcionar.

A questão do término

Uma sutileza fácil de passar batido: como o programa sabe parar? Aqui, cada produtor gera um número fixo de itens e cada consumidor consome um número fixo, com as contas batendo (dois produtores de cinco, dois consumidores de cinco). Cada thread sabe exatamente quantas voltas dar e termina sozinha.

Em sistemas de verdade, quem produz e quem consome raramente sabe o total de antemão. O consumidor fica num loop infinito esperando trabalho que pode nunca mais vir, e desligar isso com elegância, sem deixar consumidor preso pra sempre num sem_wait, é um problema à parte, com soluções próprias como enviar um valor especial que sinaliza “acabou”. O núcleo da coordenação, os dois semáforos contra o buffer e o mutex sobre os índices, é o que não muda.