One-to-many producer/consumer implementation on top of POSIX API

producer_consumer.c

struct task {
  socket_t accepted_socket;
};

struct queue {
  struct task tasks[NUM_WORKER_THREADS];
  int num_tasks;
  pthread_mutex_t mutex;
  pthread_cond_t condvar;
};

void produce(struct queue *q, struct task *t) {
  pthread_mutex_lock(&q->mu);

  /* If the queue is full, wait */
  while (q->num_tasks >= NUM_WORKER_THREADS) {
    pthread_cond_wait(&q->condvar, &q->mutex);
  }

  /* Add task to the queue */
  q->tasks[q->num_tasks] = *t;
  q->num_tasks++;

  /* Notify workers that queue is not empty */
  pthread_cond_signal(&q->condvar);
  pthread_mutex_unlock(&q->mutex);
}

void consume(struct queue *q, struct task *t) {
  pthread_mutex_lock(&q->mu);

  /* If the queue is empty, wait */
  while (q->num_tasks == 0) {
    pthread_cond_wait(&q->condvar, &q->mutex);
  }

  /* Grab task from the queue */
  *t = q->tasks[q->num_tasks] = *t;
  q->num_tasks--;

  /* Notify master */
  pthread_cond_signal(&q->condvar);
  pthread_mutex_unlock(&q->mutex);
}