Wiki CC3301

Un semáforo es un contenedor de tickets. En él se pueden depositar un ticket con sem_post y extraer un ticket con sem_wait. La garantía es que si sem_wait no encuentra tickets entonces no retorna hasta que algún thread invoque sem_post para depositar un ticket.

Un semáforo se manipula con las siguientes funciones:

  #include <semaphore.h>
  sem_t sem;
  sem_init(&sem, 0, tickets_iniciales);
  sem_post(&sem);
  sem_wait(&sem);

La siguiente es una implementación correcta de la estructura Buffer del productor/consumidor.

  /* Versión correcta */
  typedef struct {
    int size;
    Cuadro **array;
    int in, out;
    sem_t empty, full;
  } Buffer;
  
  Buffer *nuevoBuffer(int size) {
    Buffer *buf= (Buffer*)malloc(sizeof(Buffer));
    buf->size= size;
    buf->array= (Cuadro**)malloc(size*sizeof(Cuadro*));
    buf->in= buf->out= 0;
    sem_init(&buf->empty, 0, size);
    sem_init(&buf->full, 0, 0);
    return buf;
  }
  
  void put(Buffer *buf, Cuadro *cuadro) {
    sem_wait(&buf->empty);
    buf->array[buf->in]= cuadro;
    buf->in= (buf->in+1) % buf->size;
    sem_post(&buf->full);
  }
  
  Cuadro *get(Buffer *buf) {
    Cuadro *cuadro;
    sem_wait(&buf->full);
    cuadro= buf->array[buf->out];
    buf->out= (buf->out+1) % buf->size;
    sem_post(&buf->empty);
    return cuadro;
  }

Hay muchos problemas que se pueden paralelizar usando la abstracción productor/consumidor. En ellos la solución es la misma de arriba. Solo cambia el código del productor, del consumidor y los ítemes que se depositan en el Buffer. Pero el código del buffer es el mismo.

El problema del productor/consumidor se puede generalizar a múltiples productores y múltiples consumidores que comparten el mismo buffer. Muestre que en este caso el código para depositar y extraer elementos de buffer puede sufrir de data-races. Corrija el código usando un semáforo o mutex adicional.

La segunda solución se basa en que el deadlock solo se produce si los 5 filósofos llegan a cenar. Si se limita a 4 los filósofos que pueden cenar, ya no habrá deadlock. Entonces se usa un semáforo con 4 tickets iniciales:

  pthread_mutex_t palitos[5];
  sem_t portero; /* con 4 tickets iniciales */
  
  void filosofo(int i) {
    for (;;) {
      sem_wait(&portero);
      pthread_mutex_lock(&palitos[i]);
      pthread_mutex_lock(&palitos[(i+1)%5]);
      comer(i, (i+1)%5);
      pthread_mutex_unlock(&palitos[i]);
      pthread_mutex_unlock(&palitos[(i+1)%5]);
      sem_post(&portero);
      pensar();
    }
  }

El problema de esta solución es que solo se aplica a 5 filósofos y 5 palitos. No se puede generalizar a un número variable de threads accediendo a múltiples estructuras de datos.

Los semáforos y monitores (mutex + condición) son equivalentes, en el sentido que si un problema se puede resolver con una herramienta, entonces también se puede resolver con la otra herramienta. La única excepción la constituye el timeout de pthread_cond_timedwait que no se puede implementar con las semáforos por sí solos.

Probaremos la equivalencia implementando un semáforo a partir de un monitor y un monitor a partir de un semáforo. Esto le ayudará a comprender el significado de ambas herramientas.

  #include <pthread.h>

  typedef struct {
    int tickets;
    pthread_cond_t vacio; /* Condicion para esperar entrar al semáforo */
    pthread_mutex_t mutex;
  } Semaphore;

  void sema_init(Semaphore *sema, int ini) {
    sema->tickets = ini;
    pthread_cond_init(&sema->vacio, NULL);
    pthread_mutex_init(&sema->mutex, NULL);
  }
 
  void sema_wait(Semaphore *sema) {
    pthread_mutex_lock(&sema->mutex);
    while (sema->tickets == 0) /* Nunca un if! */
      pthread_cond_wait(&sema->vacio, &sema->mutex);
 
    sema->tickets--;
    pthread_mutex_unlock(&sema->mutex);
  }
 
  void sema_post(Semaphore *sema) {
      pthread_mutex_lock(&sema->mutex);
      sema->tickets++;
      pthread_cond_signal(&sema->vacio); /* siempre libero a uno solo */
      pthread_mutex_unlock(&sema->mutex);
  }

Observe que se uso signal en vez de broadcast. Esto se puede hacer porque hay un solo tipo de threads esperando: threads que esperan obtener un ticket. Por lo tanto basta una sola condición. En el problema del buffer hay 2 tipos de threads en espera: los que esperan extraer un ítem del buffer y los que esperan depositar un ítem. Por ello se necesitan 2 condiciones, o alternativamente si se usa una sola condición usar broadcast en vez de signal.

El problema de la implementación de más arriba es que un thread T que se despierta después de un sem_wait no necesariamente obtiene el ticket recién depositado con sem_post. Como T todavía tiene que esperar por la propiedad del mutex, puede ocurrir que llega un nuevo thread U que invoca sem_wait y le roba el mutex a T. El resultado es que U obtiene el ticket a pesar de que T lleva mucho más tiempo esperándolo.

Modifique su implementación de modo que se garantice que el thread T reciba el ticket. Base su implementanción en la metáfora de la Isapre: para evitar las colas hay un distribuidor de números en la entrada y un visor que indica a quién se atiende actualmente. De la misma forma, haga que el que invoca sem_wait obtenga un número y espera hasta que el visor indique al menos ese número.

La siguiente es una implementación de un verdadero monitor, el que fusiona el mutex con la condición, a la manera de los monitores de Java.

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

  typedef struct node {
    sem_t wait;
    struct node *next;
  } Node;

  typedef struct {
    sem_t mutex;
    Node *head;
  } Monitor;

  void mon_init(Monitor *mon) {
    sem_init(&mon->mutex, 0, 1);
    mon->head= NULL;
  }

  void mon_lock(Monitor *mon) {
    sem_wait(&mon->mutex);
  }

  void mon_unlock(Monitor *mon) {
    sem_post(&mon->mutex);
  }

  void mon_wait(Monitor *mon) {
    Node node;
    sem_init(&node.wait, 0, 0);
    node.next= mon->head;
    mon->head= &node;
    sem_post(&mon->mutex);
    sem_wait(&node.wait);   /* se bloquea a la espera del broadcast */
    sem_destroy(&node.wait);
    sem_wait(&mon->mutex);  /* se bloquea a la espera del monitor */
  }

  void mon_broadcast(Monitor *mon) {
    Node *node;
    for (node= mon->head; node!=NULL; node= node->next)
      sem_post(&node->wait);
    mon->head= NULL;
  }

Este código no efectúa ninguna validación con respecto al buen uso de los monitores, es decir que las operaciones mon_unlock, mon_wait y mon_broadcast solo se invoquen cuando se obtuvo previamente el monitor con mon_lock.

Implemente a partir de semáforos un monitor con múltiples condiciones, a la manera de los mutex y condiciones de pthreads.

Esta solución evita la hambruna por medio de la metáfora de la isapre. Como en una isapre los threads deben pedir un número antes de realizar su operación. Un thread solo puede entrar a realizar su operación cuando el número que aparece en un visor coincide con el número que se le asignó. De esta forma las entradas se satisfacen en orden FIFO (first in first out), aunque las lecturas se hacen en paralelo y por lo tanto las salidas pueden ocurrir en un orden no FIFO.

En esta solución los lectores no pueden concertarse para causar hambruna a un escritor porque una vez que el escritor recibe su número, ningún otro lector podrá entrar al diccionario. Tarde o temprano los lectores que habían entrado previamente tendrán que salir y será el turno del escritor.

  pthread_mutex_t mutex;
  pthread_cond_t cond;
  int readers= 0;
  int display= 0;
  int serial= 0;
  
  void enterRead() {
    int myNum;
    pthread_mutex_lock(&mutex);
    myNum= = serial++;
    while (myNum!=display)
      pthread_cond_wait(&cond, &mutex);
    readers++;
    display++;
    pthread_cond_broadcast(&cond); /* Ver nota */
    pthread_mutex_unlock(&mutex);
  }
  
  void exitRead() {
    pthread_mutex_lock(&mutex);
    readers--;
    if (readers==0)
      pthread_cond_broadcast(&cond);
    pthread_mutex_unlock(&mutex);
  }
  
  void enterWrite() {
    int myNum;
    pthread_mutex_lock(&mutex);
    myNum= serial++;
    while (readers>0 || myNum!=display)
      pthread_cond_wait(&cond, &mutex);
    pthread_mutex_unlock(&mutex);
  }
  
  void exitWrite() {
    pthread_mutex_lock(&mutex);
    display++;
    pthread_cond_broadcast(&cond);
    pthread_mutex_unlock(&mutex);
  }

Nota: el broadcast se necesita acá porque el incremento de display podría gatillar que un lector que se encontraba esperando ahora puede entrar.

Discusión: Esta solución funciona pero puede ser muy ineficiente. El problema es el siguiente. Supongamos que hay un escritor trabajando y n lectores esperando. Cuando el escritor se va, en el peor caso se pueden requerir hasta O(n^2) cambios de thread a thread para que todos los lectores comiencen a trabajar. Esto es muy ineficiente. Esto se puede bajar a O(n) usando múltiples variables de condición.