Producer/consumer with bounded buffer

Question

Could someone check my code and tell me if I am on the right track.. It seems like I am a bit lost.. if you see my errors, please let me know them..

What I am trying to do is to solve bounded buffer using my own semaphores as well as GCD.

Thanks in advance..

sema.c

void procure( Semaphore *semaphore ) {

        pthread_mutex_lock(semaphore->mutex1);

        while(semaphore->value <= 0)
                pthread_cond_wait(&semaphore->condition, semaphore->mutex1);


        semaphore->value--;

        pthread_mutex_unlock(semaphore->mutex1);


}

void vacate( Semaphore *semaphore ) {

        pthread_mutex_lock(semaphore->mutex1);
        semaphore->value++;
        pthread_cond_signal(&semaphore->condition);
        pthread_mutex_unlock(semaphore->mutex1);        
}


void init ( Semaphore *semaphore ){

        semaphore->value = 1;

        pthread_mutex_t myMutex;
        semaphore->mutex1 = &myMutex;
        pthread_mutex_init( semaphore->mutex1, NULL);



}

void destroy ( Semaphore *semaphore ) {

        pthread_mutex_destroy(semaphore->mutex1);


}

and main.c

struct variables {

        Semaphore *sem;

};

struct variables vars;

void constructer (int *buffer, int *in, int *out) {

        init(vars.sem);

}


void deconstructer () {

        destroy(vars.sem);

}



int rand_num_gen() {
        uint_fast16_t buffer;
        int file;
        int *rand;

        file = open("/dev/random", O_RDONLY);


        while( 1 ) {

                read(file, &buffer, sizeof(buffer));

                printf("16 bit number: %hu\n", buffer );

                *rand = (int) buffer;

                close(file);


                break;
        }

        return *rand;
}


void put_buffer( int* buffer, int* in, int* out ) {

        buffer[*in] = rand_num_gen();    // produce

        procure(vars.sem);                // wait here
                *in = (*in + 1) % BUF_SIZE;
        vacate(vars.sem);

}

void get_buffer( int* buffer, int* in, int* out ) {

        int value;  

        procure(vars.sem);
                value = buffer[*out];  
        vacate(vars.sem);

        *out = (*out + 1) % BUF_SIZE;

}

int main (void) {


        int *in, *out, *buffer;     

        constructer(buffer, in, out);        



        dispatch_queue_t producer, consumer;    

        producer = dispatch_queue_create("put_buffer", NULL);
        consumer = dispatch_queue_create("get_buffer", NULL);



        dispatch_async(producer,
                       ^{
                               int i;
                               do
                               {
                                        put_buffer( buffer, in, out );


                                       dispatch_async(consumer,
                                                      ^{
                                                              get_buffer( buffer, in, out );

                                                              if (i == RUN_LENGTH) exit(EXIT_SUCCESS);
                                                      });
                               }
                               while (i < RUN_LENGTH);             
                       });

        dispatch_main();                

        deconstructer();

        exit (0);
}

Answer 1

Your code has a bug. In the init function you assign the address of a local variable to semaphore->mutex1, and when the function returns this address will be invalid. Later you still use this address, so this leads to undefined behavior.

You must either allocate the memory for the mutex directly in the semaphore (without a pointer) or allocate the memory via malloc.

Update:

Your program has so many bugs that you should definitely pick an easier topic to learn the basic concepts about memory management, how to allocate, use and reference a buffer, do proper error handling, etc. Here is a slightly edited version of your code. It still won't work, but probably has some ideas that you should follow.

#include <limits.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

void procure(Semaphore *semaphore) {
  pthread_mutex_lock(semaphore->mutex1);

  while (semaphore->value <= 0)
    pthread_cond_wait(&semaphore->condition, semaphore->mutex1);

  semaphore->value--;
  pthread_mutex_unlock(semaphore->mutex1);
}

void vacate(Semaphore *semaphore) {
  pthread_mutex_lock(semaphore->mutex1);
  semaphore->value++;
  pthread_cond_signal(&semaphore->condition);
  pthread_mutex_unlock(semaphore->mutex1);  
}

struct variables {
  mutex_t sem_mutex;
  Semaphore sem;
};

struct variables vars;

void constructor(int *buffer, int *in, int *out) {
  vars.sem.value = 1;
  vars.sem.mutex1 = &vars.sem_mutex;
  pthread_mutex_init(vars.sem.mutex1, NULL);
}

void deconstructor() {
  pthread_mutex_destroy(&semaphore->mutex1);
}

int rand_num_gen() {
  const char *randomfile = "/dev/random";
  unsigned char buffer[2]; // Changed: always treat files as byte sequences.
  FILE *f = fopen(randomfile, "rb");
  // Changed: using stdio instead of raw POSIX file access,
  // since the API is much simpler; you don't have to care
  // about interrupting signals or partial reads.

  if (f == NULL) { // Added: error handling
    fprintf(stderr, "E: cannot open %s\n", randomfile);
    exit(EXIT_FAILURE);
  }
  if (fread(buffer, 1, 2, f) != 2) { // Added: error handling
    fprintf(stderr, "E: cannot read from %s\n", randomfile);
    exit(EXIT_FAILURE);
  }
  fclose(f);
  int number = (buffer[0] << CHAR_BIT) | buffer[1];
  // Changed: be independent of the endianness of the system.
  // This doesn't matter for random number generators but is
  // still an important coding style.
  printf("DEBUG: random number: %x\n", (unsigned int) number);
  return number;
}

void put_buffer( int* buffer, int* in, int* out ) {
  buffer[*in] = rand_num_gen();    // produce
  procure(&vars.sem);    // wait here
  *in = (*in + 1) % BUF_SIZE;
  vacate(&vars.sem);
}

void get_buffer( int* buffer, int* in, int* out ) {
  int value;  
  procure(&vars.sem);
  value = buffer[*out];  
  vacate(&vars.sem);
  *out = (*out + 1) % BUF_SIZE;
}

int main (void) {
  int inindex = 0, outindex = 0;
  int buffer[BUF_SIZE];

  constructor(buffer, &inindex, &outindex);
  // Changed: provided an actual buffer and actual variables
  // for the indices into the buffer.
  dispatch_queue_t producer, consumer;    
  producer = dispatch_queue_create("put_buffer", NULL);
  consumer = dispatch_queue_create("get_buffer", NULL);

  dispatch_async(producer, ^{
    int i;
    do {
      put_buffer(buffer, &inindex, &outindex);
      dispatch_async(consumer, ^{
        get_buffer(buffer, &inindex, &outindex);
        if (i == RUN_LENGTH) exit(EXIT_SUCCESS);
      });
    } while (i < RUN_LENGTH);
  });

  dispatch_main();    
  deconstructor();
  exit (0);
}

As I said, I didn't catch all the bugs.