drd/tests/circular_buffer.c

   1 /* Test program that performs producer-consumer style communication through
   2  * a circular buffer. This test program is a slightly modified version of the
   3  * test program made available by Miguel Ojeda
   4  * -- see also http://article.gmane.org/gmane.comp.debugging.valgrind/8782.
   5  */
   6
   7
   8 #include <stdio.h>
   9 #include <string.h>
  10 #include <stdlib.h>
  11 #include <unistd.h>
  12 #include <time.h>
  13 #include <pthread.h>
  14 #include <semaphore.h>
  15 #include <fcntl.h>
  16 #include "../../config.h"
  17
  18
  19 /** gcc versions 4.1.0 and later have support for atomic builtins. */
  20
  21 #ifndef HAVE_BUILTIN_ATOMIC
  22 #error Sorry, but this test program can only be compiled by a compiler that\
  23 has built-in functions for atomic memory access.
  24 #endif
  25
  26
  27 #define BUFFER_MAX (2)
  28 #define DATA_SEMAPHORE_NAME "cb-data-semaphore"
  29 #define FREE_SEMAPHORE_NAME "cb-free-semaphore"
  30
  31
  32 typedef int data_t;
  33
  34 typedef struct {
  35   /* Counting semaphore representing the number of data items in the buffer. */
  36   sem_t* data;
  37   /* Counting semaphore representing the number of free elements. */
  38   sem_t* free;
  39   /* Position where a new elements should be written. */
  40   int in;
  41   /* Position from where an element can be removed. */
  42   int out;
  43   /* Mutex that protects 'in'. */
  44   pthread_mutex_t mutex_in;
  45   /* Mutex that protects 'out'. */
  46   pthread_mutex_t mutex_out;
  47   /* Data buffer. */
  48   data_t buffer[BUFFER_MAX];
  49 } buffer_t;
  50
  51 static int quiet = 0;
  52 static int use_locking = 1;
  53
  54 static __inline__
  55 int fetch_and_add(int* p, int i)
  56 {
  57   return __sync_fetch_and_add(p, i);
  58 }
  59
  60 static sem_t* create_semaphore(const char* const name, const int value)
  61 {
  62 #ifdef VGO_darwin
  63   char name_and_pid[32];
  64   snprintf(name_and_pid, sizeof(name_and_pid), "%s-%d", name, getpid());
  65   sem_t* p = sem_open(name_and_pid, O_CREAT | O_EXCL, 0600, value);
  66   if (p == SEM_FAILED) {
  67     perror("sem_open");
  68     return NULL;
  69   }
  70   return p;
  71 #else
  72   sem_t* p = malloc(sizeof(*p));
  73   if (p)
  74     sem_init(p, 0, value);
  75   return p;
  76 #endif
  77 }
  78
  79 static void destroy_semaphore(const char* const name, sem_t* p)
  80 {
  81 #ifdef VGO_darwin
  82   sem_close(p);
  83   sem_unlink(name);
  84 #else
  85   sem_destroy(p);
  86   free(p);
  87 #endif
  88 }
  89
  90 static void buffer_init(buffer_t * b)
  91 {
  92   b->data = create_semaphore(DATA_SEMAPHORE_NAME, 0);
  93   b->free = create_semaphore(FREE_SEMAPHORE_NAME, BUFFER_MAX);
  94
  95   pthread_mutex_init(&b->mutex_in, NULL);
  96   pthread_mutex_init(&b->mutex_out, NULL);
  97
  98   b->in = 0;
  99   b->out = 0;
 100 }
 101
 102 static void buffer_recv(buffer_t* b, data_t* d)
 103 {
 104   int out;
 105   sem_wait(b->data);
 106   if (use_locking)
 107     pthread_mutex_lock(&b->mutex_out);
 108   out = fetch_and_add(&b->out, 1);
 109   if (out >= BUFFER_MAX)
 110   {
 111     fetch_and_add(&b->out, -BUFFER_MAX);
 112     out -= BUFFER_MAX;
 113   }
 114   *d = b->buffer[out];
 115   if (use_locking)
 116     pthread_mutex_unlock(&b->mutex_out);
 117   if (! quiet)
 118   {
 119     printf("received %d from buffer[%d]\n", *d, out);
 120     fflush(stdout);
 121   }
 122   sem_post(b->free);
 123 }
 124
 125 static void buffer_send(buffer_t* b, data_t* d)
 126 {
 127   int in;
 128   sem_wait(b->free);
 129   if (use_locking)
 130     pthread_mutex_lock(&b->mutex_in);
 131   in = fetch_and_add(&b->in, 1);
 132   if (in >= BUFFER_MAX)
 133   {
 134     fetch_and_add(&b->in, -BUFFER_MAX);
 135     in -= BUFFER_MAX;
 136   }
 137   b->buffer[in] = *d;
 138   if (use_locking)
 139     pthread_mutex_unlock(&b->mutex_in);
 140   if (! quiet)
 141   {
 142     printf("sent %d to buffer[%d]\n", *d, in);
 143     fflush(stdout);
 144   }
 145   sem_post(b->data);
 146 }
 147
 148 static void buffer_destroy(buffer_t* b)
 149 {
 150   destroy_semaphore(DATA_SEMAPHORE_NAME, b->data);
 151   destroy_semaphore(FREE_SEMAPHORE_NAME, b->free);
 152
 153   pthread_mutex_destroy(&b->mutex_in);
 154   pthread_mutex_destroy(&b->mutex_out);
 155 }
 156
 157 static buffer_t b;
 158
 159 static void producer(int* id)
 160 {
 161   buffer_send(&b, id);
 162   pthread_exit(NULL);
 163 }
 164
 165 #define MAXSLEEP (100 * 1000)
 166
 167 static void consumer(int* id)
 168 {
 169   int d;
 170   usleep(rand() % MAXSLEEP);
 171   buffer_recv(&b, &d);
 172   if (! quiet)
 173   {
 174     printf("%i: %i\n", *id, d);
 175     fflush(stdout);
 176   }
 177   pthread_exit(NULL);
 178 }
 179
 180 #define THREADS (10)
 181
 182 int main(int argc, char** argv)
 183 {
 184   pthread_t producers[THREADS];
 185   pthread_t consumers[THREADS];
 186   int thread_arg[THREADS];
 187   int i;
 188   int optchar;
 189
 190   while ((optchar = getopt(argc, argv, "nq")) != EOF)
 191   {
 192     switch (optchar)
 193     {
 194     case 'n':
 195       use_locking = 0;
 196       break;
 197     case 'q':
 198       quiet = 1;
 199       break;
 200     }
 201   }
 202
 203   srand(time(NULL));
 204
 205   buffer_init(&b);
 206
 207   for (i = 0; i < THREADS; ++i)
 208   {
 209     thread_arg[i] = i;
 210     pthread_create(producers + i, NULL,
 211                    (void * (*)(void *)) producer, &thread_arg[i]);
 212   }
 213
 214   for (i = 0; i < THREADS; ++i)
 215     pthread_create(consumers + i, NULL,
 216                    (void * (*)(void *)) consumer, &thread_arg[i]);
 217
 218   for (i = 0; i < THREADS; ++i)
 219   {
 220     pthread_join(producers[i], NULL);
 221     pthread_join(consumers[i], NULL);
 222   }
 223
 224   buffer_destroy(&b);
 225
 226   return 0;
 227 }