prod_cons.c

Go to the documentation of this file.

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include "prod_cons.h"


prod_cons_t *create_prod_cons(int sz) {
        prod_cons_t *prod_cons = malloc(sizeof(prod_cons_t));


        prod_cons->buffer = calloc(sz, sizeof(int));
        prod_cons->buf_size = sz;
        pthread_mutex_init( &(prod_cons->prod_mutex), NULL);
        pthread_mutex_init( &(prod_cons->cons_mutex), NULL);
        // The buffer starts with all positions free
        sem_init( &(prod_cons->prod_sem), 0, sz);
        // The buffer starts without any filled positions
        sem_init( &(prod_cons->cons_sem), 0, 0);
        prod_cons->prod_pos = 0;
        prod_cons->cons_pos = 0;
        

        return prod_cons;
}


void put(prod_cons_t *prod_cons, void *val) {
        int status = -1;
        
        // Wait for a free position
        do {
                // In NPTL, sem_wait can be interrupted by a signal.
                // If it occours, it will return with error.
                status = sem_wait( &(prod_cons->prod_sem) );
                if (status < 0) {
                        int saved_errno = errno;
                        if (saved_errno != EINTR) {
                                printf("WARNING: %s (%d): ", __FILE__, __LINE__);
                                errno = saved_errno; // printf can overwirte errno
                                perror("sem_wait(): ");
#ifdef DEBUG
                        } else {
                                printf("%s (%d): sem_wait(): interrupted by signal\n", __FILE__, __LINE__);
#endif                          
                        }
                }
        } while (status < 0);
                
        pthread_mutex_lock( &(prod_cons->prod_mutex) ); 
        prod_cons->buffer[prod_cons->prod_pos] = val;
        prod_cons->prod_pos = (prod_cons->prod_pos+1) % prod_cons->buf_size;
        pthread_mutex_unlock( &(prod_cons->prod_mutex) );
        
        // Let the consumer get another position
        sem_post( &(prod_cons->cons_sem) );
        
}


void *get(prod_cons_t *prod_cons) {
        void *val = NULL;
        int status = -1;

        // Wait for someting to be consumed
        do {
                // In NPTL, sem_wait can be interrupted by a signal.
                // If it occours, it will return with error.
                status = sem_wait( &(prod_cons->cons_sem) );
                if (status < 0) {
                        int saved_errno = errno;
                        if (saved_errno != EINTR) {
                                printf("WARNING: %s (%d): ", __FILE__, __LINE__);
                                errno = saved_errno; // printf can overwirte errno
                                perror("sem_wait(): ");
#ifdef DEBUG
                        } else {
                                printf("%s (%d): sem_wait(): interrupted by signal\n", __FILE__, __LINE__);
#endif                          
                        }
                }
        } while (status < 0);
        
        pthread_mutex_lock( &(prod_cons->cons_mutex) );
        val = prod_cons->buffer[prod_cons->cons_pos];
        
        // Coutinho: clean the consumed position
        prod_cons->buffer[prod_cons->cons_pos] = NULL;

        prod_cons->cons_pos = (prod_cons->cons_pos+1) % prod_cons->buf_size;
        pthread_mutex_unlock( &(prod_cons->cons_mutex) );
        
        // Free one buffer's position
        sem_post( &(prod_cons->prod_sem) );
        return val;
}


void destroy_prod_cons(prod_cons_t *prod_cons) {

        // Coutinho: XXX: For now we are destroying the buffer even if there's someting inside
        sem_destroy( &(prod_cons->prod_sem) );
        sem_destroy( &(prod_cons->cons_sem) );
        free(prod_cons->buffer);
        free(prod_cons);
        
}

---