11.2: Producers and consumers with semaphores

Using these semaphore wrapper functions, we can write a solution to the Producer-Consumer problem from Section 10.2. The code in this section is in queue_sem.c.

Here’s the new definition of Queue, replacing the mutex and condition variables with semaphores:

typedef struct {
    int *array;
    int length;
    int next_in;
    int next_out;
    Semaphore *mutex;       //-- new
    Semaphore *items;       //-- new
    Semaphore *spaces;      //-- new
} Queue;

And here’s the new version of make_queue:

Queue *make_queue(int length)
{
    Queue *queue = (Queue *) malloc(sizeof(Queue));
    queue->length = length;
    queue->array = (int *) malloc(length * sizeof(int));
    queue->next_in = 0;
    queue->next_out = 0;
    queue->mutex = make_semaphore(1);
    queue->items = make_semaphore(0);
    queue->spaces = make_semaphore(length-1);
    return queue;
}

mutex is used to guarantee exclusive access to the queue; the initial value is 1, so the mutex is initially unlocked.

items is the number of items in the queue, which is also the number of consumer threads that can execute queue_pop without blocking. Initially there are no items in the queue.

spaces is the number of empty spaces in the queue, which is the number of producer threads that can execute queue_push without blocking. Initially the number of spaces is the capacity of the queue, which is length-1, as explained in Section 10.1.

Here is the new version of queue_push, which is run by producer threads:

void queue_push(Queue *queue, int item) {
  semaphore_wait(queue->spaces);
  semaphore_wait(queue->mutex);

  queue->array[queue->next_in] = item;
  queue->next_in = queue_incr(queue, queue->next_in);

  semaphore_signal(queue->mutex);
  semaphore_signal(queue->items);
}

Notice that queue_push doesn’t have to call queue_full any more; instead, the semaphore keeps track of how many spaces are available and blocks producers if the queue is full.

Here is the new version of queue_pop:

int queue_pop(Queue *queue) {
  semaphore_wait(queue->items);
  semaphore_wait(queue->mutex);
  
  int item = queue->array[queue->next_out];
  queue->next_out = queue_incr(queue, queue->next_out);

  semaphore_signal(queue->mutex);
  semaphore_signal(queue->spaces);

  return item;
}

This solution is explained, using pseudo-code, in Chapter 4 of The Little Book of Semaphores.

Using the code in the repository for this book, you should be able to compile and run this solution like this:

$ make queue_sem
$ ./queue_sem