drd/tests/circular_buffer.c

/* Test program that performs producer-consumer style communication through
 * a circular buffer. This test program is a slightly modified version of the
 * test program made available by Miguel Ojeda
 * -- see also http://article.gmane.org/gmane.comp.debugging.valgrind/8782.
 */


#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <time.h>
#include <pthread.h>
#include <semaphore.h>
#include <fcntl.h>
#include "../../config.h"


/** gcc versions 4.1.0 and later have support for atomic builtins. */

#ifndef HAVE_BUILTIN_ATOMIC
#error Sorry, but this test program can only be compiled by a compiler that\
has built-in functions for atomic memory access.
#endif


#define BUFFER_MAX (2)
#define DATA_SEMAPHORE_NAME "cb-data-semaphore"
#define FREE_SEMAPHORE_NAME "cb-free-semaphore"


typedef int data_t;

typedef struct {
  /* Counting semaphore representing the number of data items in the buffer. */
  sem_t* data;
  /* Counting semaphore representing the number of free elements. */
  sem_t* free;
  /* Position where a new elements should be written. */
  int in;
  /* Position from where an element can be removed. */
  int out;
  /* Mutex that protects 'in'. */
  pthread_mutex_t mutex_in;
  /* Mutex that protects 'out'. */
  pthread_mutex_t mutex_out;
  /* Data buffer. */
  data_t buffer[BUFFER_MAX];
} buffer_t;

static int quiet = 0;
static int use_locking = 1;

static __inline__
int fetch_and_add(int* p, int i)
{
  return __sync_fetch_and_add(p, i);
}

static sem_t* create_semaphore(const char* const name, const int value)
{
#ifdef VGO_darwin
  char name_and_pid[32];
  snprintf(name_and_pid, sizeof(name_and_pid), "%s-%d", name, getpid());
  sem_t* p = sem_open(name_and_pid, O_CREAT | O_EXCL, 0600, value);
  if (p == SEM_FAILED) {
    perror("sem_open");
    return NULL;
  }
  return p;
#else
  sem_t* p = malloc(sizeof(*p));
  if (p)
    sem_init(p, 0, value);
  return p;
#endif
}

static void destroy_semaphore(const char* const name, sem_t* p)
{
#ifdef VGO_darwin
  sem_close(p);
  sem_unlink(name);
#else
  sem_destroy(p);
  free(p);
#endif
}

static void buffer_init(buffer_t * b)
{
  b->data = create_semaphore(DATA_SEMAPHORE_NAME, 0);
  b->free = create_semaphore(FREE_SEMAPHORE_NAME, BUFFER_MAX);

  pthread_mutex_init(&b->mutex_in, NULL);
  pthread_mutex_init(&b->mutex_out, NULL);

  b->in = 0;
  b->out = 0;
}

static void buffer_recv(buffer_t* b, data_t* d)
{
  int out;
  sem_wait(b->data);
  if (use_locking)
    pthread_mutex_lock(&b->mutex_out);
  out = fetch_and_add(&b->out, 1);
  if (out >= BUFFER_MAX)
  {
    fetch_and_add(&b->out, -BUFFER_MAX);
    out -= BUFFER_MAX;
  }
  *d = b->buffer[out];
  if (use_locking)
    pthread_mutex_unlock(&b->mutex_out);
  if (! quiet)
  {
    printf("received %d from buffer[%d]\n", *d, out);
    fflush(stdout);
  }
  sem_post(b->free);
}

static void buffer_send(buffer_t* b, data_t* d)
{
  int in;
  sem_wait(b->free);
  if (use_locking)
    pthread_mutex_lock(&b->mutex_in);
  in = fetch_and_add(&b->in, 1);
  if (in >= BUFFER_MAX)
  {
    fetch_and_add(&b->in, -BUFFER_MAX);
    in -= BUFFER_MAX;
  }
  b->buffer[in] = *d;
  if (use_locking)
    pthread_mutex_unlock(&b->mutex_in);
  if (! quiet)
  {
    printf("sent %d to buffer[%d]\n", *d, in);
    fflush(stdout);
  }
  sem_post(b->data);
}

static void buffer_destroy(buffer_t* b)
{
  destroy_semaphore(DATA_SEMAPHORE_NAME, b->data);
  destroy_semaphore(FREE_SEMAPHORE_NAME, b->free);

  pthread_mutex_destroy(&b->mutex_in);
  pthread_mutex_destroy(&b->mutex_out);
}

static buffer_t b;

static void producer(int* id)
{
  buffer_send(&b, id);
  pthread_exit(NULL);
}

#define MAXSLEEP (100 * 1000)

static void consumer(int* id)
{
  int d;
  usleep(rand() % MAXSLEEP);
  buffer_recv(&b, &d);
  if (! quiet)
  {
    printf("%i: %i\n", *id, d);
    fflush(stdout);
  }
  pthread_exit(NULL);
}

#define THREADS (10)

int main(int argc, char** argv)
{
  pthread_t producers[THREADS];
  pthread_t consumers[THREADS];
  int thread_arg[THREADS];
  int i;
  int optchar;

  while ((optchar = getopt(argc, argv, "nq")) != EOF)
  {
    switch (optchar)
    {
    case 'n':
      use_locking = 0;
      break;
    case 'q':
      quiet = 1;
      break;
    }
  }

  srand(time(NULL));

  buffer_init(&b);

  for (i = 0; i < THREADS; ++i)
  {
    thread_arg[i] = i;
    pthread_create(producers + i, NULL,
                   (void * (*)(void *)) producer, &thread_arg[i]);
  }

  for (i = 0; i < THREADS; ++i)
    pthread_create(consumers + i, NULL,
                   (void * (*)(void *)) consumer, &thread_arg[i]);

  for (i = 0; i < THREADS; ++i)
  {
    pthread_join(producers[i], NULL);
    pthread_join(consumers[i], NULL);
  }

  buffer_destroy(&b);

  return 0;
}