Simple producer/consumer in C

producer_consumer.c

#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>

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

#if !defined(BUFFER_SIZE)
#define BUFFER_SIZE 10
#endif

typedef uint8_t ITEM_T;

struct buffer_t {
  ITEM_T buffer[BUFFER_SIZE]; //< The actual buffer
  sem_t occupied;             //< Number of full buffers
  sem_t empty;                //< Number of empty buffers
  unsigned in_pos;            //< Input position
  unsigned out_pos;           //< Output position
  pthread_mutex_t produce_mutex;
  pthread_mutex_t consume_mutex;
};

static void producer(struct buffer_t *buf, ITEM_T item) {
  sem_wait(&buf->empty);
  pthread_mutex_lock(&buf->produce_mutex);

  buf->buffer[buf->in_pos] = item;
  buf->in_pos++;
  buf->in_pos %= BUFFER_SIZE;

  pthread_mutex_unlock(&buf->produce_mutex);
  sem_post(&buf->occupied);
}

static ITEM_T consumer(struct buffer_t *buf) {
  sem_wait(&buf->occupied);
  pthread_mutex_lock(&buf->consume_mutex);

  ITEM_T item = buf->buffer[buf->out_pos];
  buf->out_pos++;
  buf->out_pos %= BUFFER_SIZE;

  pthread_mutex_unlock(&buf->consume_mutex);
  sem_post(&buf->empty);

  return item;
}

static void *start_producer(void *arg) {
  struct buffer_t *buf = (struct buffer_t *)arg;

  while (1) {
    ITEM_T item = rand();
    printf("[producer] produced %d\n", item);
    producer(buf, item);
  }

  pthread_exit(NULL);
}

static void *start_consumer(void *arg) {
  struct buffer_t *buf = (struct buffer_t *)arg;

  while (1) {
    ITEM_T item = consumer(buf);
    printf("[consumer] consumed %d\n", item);
  }

  pthread_exit(NULL);
}

static void init_buffer(struct buffer_t *buf) {
  sem_init(&buf->occupied, 0, 0);
  sem_init(&buf->empty, 0, BUFFER_SIZE);
  pthread_mutex_init(&buf->produce_mutex, NULL);
  pthread_mutex_init(&buf->consume_mutex, NULL);
  buf->in_pos = 0;
  buf->out_pos = 0;
}

int main(int argc, char *argv[]) {
  pthread_t producer_thread, consumer_thread;
  pthread_attr_t attr;

  struct buffer_t buffer;
  init_buffer(&buffer);

  time_t t;
  srand((unsigned)time(&t));

  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  if (pthread_create(&producer_thread, &attr, start_producer, &buffer)) {
    fprintf(stderr, "[error] failed to create producer thread\n");
    return 1;
  }

  if (pthread_create(&consumer_thread, &attr, start_consumer, &buffer)) {
    fprintf(stderr, "[error] failed to create consumer thread\n");
    return 1;
  }

  pthread_attr_destroy(&attr);

  pthread_join(producer_thread, NULL);
  pthread_join(consumer_thread, NULL);

  pthread_exit(NULL);
}