Producer Consumer program using semaphores and pthreads

Question

I have written a code for producer-consumer problem.But I am not getting the output.There is no compilation error,but warning in my program.I am confused.Trying very hard.But can't get it.Please tell me what is wrong in my program.What will be the correct program.I am getting frustrated.Please help guys. Here is the code-

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include </usr/include/semaphore.h>

#define BUFF_SIZE   5           /* total number of slots */
#define NP          3           /* total number of producers */
#define NC          3           /* total number of consumers */
#define NITERS      4           /* number of items produced/consumed */

typedef struct {
    int buf[BUFF_SIZE];   /* shared var */
    int in;               /* buf[in%BUFF_SIZE] is the first empty slot */
    int out;              /* buf[out%BUFF_SIZE] is the first full slot */
    sem_t full;           /* keep track of the number of full spots */
    sem_t empty;          /* keep track of the number of empty spots */
    sem_t mutex;          /* enforce mutual exclusion to shared data */
} sbuf_t;

sbuf_t shared;


void *Producer(void *arg) {
    int i, item, index;

    index = (int) arg;


    for (i = 0; i < NITERS; i++) {
        /* Produce item */
        item = i;

        /* Prepare to write item to buf */

        /* If there are no empty slots, wait */
        sem_wait(&shared.empty);
        /* If another thread uses the buffer, wait */
        sem_wait(&shared.mutex);
        shared.buf[shared.in] = item;
        shared.in = (shared.in+1)%BUFF_SIZE;
        printf("[P%d] Producing %d ...\n", index, item); fflush(stdout);
        /* Release the buffer */
        sem_post(&shared.mutex);
        /* Increment the number of full slots */
        sem_post(&shared.full);

        /* Interleave  producer and consumer execution */
        if (i % 2 == 1) sleep(1);
    }
    return NULL;
}

void *Consumer(void *arg) {
    int i, item, index;

    index = (int) arg;
    for (i = NITERS; i > 0; i--) {
      sem_wait(&shared.full);
      sem_wait(&shared.mutex);
      item = i;
      item = shared.buf[shared.out];
      shared.out = (shared.out + 1) % BUFF_SIZE;
      printf("[C%d] Consuming  %d ...\n", index, item); fflush(stdout);
      /* Release the buffer */
      sem_post(&shared.mutex);
      /* Increment the number of full slots */
      sem_post(&shared.empty);

      /* Interleave  producer and consumer execution */
      if (i % 2 == 1) sleep(1);
    }
    return NULL;
}

int main() {
    pthread_t idP, idC;
    int index;

    sem_init(&shared.full, 0, 0);
    sem_init(&shared.empty, 0, BUFF_SIZE);
    pthread_mutex_init(&shared.mutex, NULL);
    for (index = 0; index < NP; index++) {
       /* Create a new producer */
       pthread_create(&idP, NULL, Producer, (void*)index);
    }
    /*create a new Consumer*/
    for (index = 0;index < NC;index++) {
        pthread_create(&idC, NULL, Consumer, (void*)index);
    }
    pthread_exit(NULL);
}

Solution 1

Maybe you should take the Compiler warnings more serious. Incorrect types and undefined functions are usually shown as warning...

I haven't checked the Logic of your program, but the principle should work:

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include </usr/include/semaphore.h>

// for sleep
#include <unistd.h>

#define BUFF_SIZE   5           /* total number of slots */
#define NP          3           /* total number of producers */
#define NC          3           /* total number of consumers */
#define NITERS      4           /* number of items produced/consumed */

typedef struct
{
    int buf[BUFF_SIZE];   /* shared var */
    int in;               /* buf[in%BUFF_SIZE] is the first empty slot */
    int out;              /* buf[out%BUFF_SIZE] is the first full slot */
    sem_t full;           /* keep track of the number of full spots */
    sem_t empty;          /* keep track of the number of empty spots */

    // use correct type here
    pthread_mutex_t mutex;          /* enforce mutual exclusion to shared data */
} sbuf_t;

sbuf_t shared;


void *Producer(void *arg)
{
    int i, item, index;

    index = (int)arg;


    for (i=0; i < NITERS; i++)
    {

        /* Produce item */
        item = i;

        /* Prepare to write item to buf */

        /* If there are no empty slots, wait */
        sem_wait(&shared.empty);
        /* If another thread uses the buffer, wait */
        pthread_mutex_lock(&shared.mutex);
        shared.buf[shared.in] = item;
        shared.in = (shared.in+1)%BUFF_SIZE;
        printf("[P%d] Producing %d ...\n", index, item);
        fflush(stdout);
        /* Release the buffer */
        pthread_mutex_unlock(&shared.mutex);
        /* Increment the number of full slots */
        sem_post(&shared.full);

        /* Interleave  producer and consumer execution */
        if (i % 2 == 1) sleep(1);
    }
    return NULL;
}

void *Consumer(void *arg)
{
    int i, item, index;

    index = (int)arg;
    for (i=NITERS; i > 0; i--) {
        sem_wait(&shared.full);
        pthread_mutex_lock(&shared.mutex);
        item=i;
        item=shared.buf[shared.out];
        shared.out = (shared.out+1)%BUFF_SIZE;
        printf("[C%d] Consuming  %d ...\n", index, item);
        fflush(stdout);
        /* Release the buffer */
        pthread_mutex_unlock(&shared.mutex);
        /* Increment the number of full slots */
        sem_post(&shared.empty);

        /* Interleave  producer and consumer execution */
        if (i % 2 == 1) sleep(1);
    }
    return NULL;
}

int main()
{
    pthread_t idP, idC;
    int index;

    sem_init(&shared.full, 0, 0);
    sem_init(&shared.empty, 0, BUFF_SIZE);
    pthread_mutex_init(&shared.mutex, NULL);
    for (index = 0; index < NP; index++)
    {
        /* Create a new producer */
        pthread_create(&idP, NULL, Producer, (void*)index);
    }
    /*create a new Consumer*/
    for(index=0; index<NC; index++)
    {
        pthread_create(&idC, NULL, Consumer, (void*)index);
    }



    pthread_exit(NULL);
}

I hope this helps.

Solution 2

There are still compiler warnings. The correct way to get the integer value from a void pointer is:

index = *(int*)arg;

And, also to pass an integer pointer, is below:

pthread_create(&idC,NULL,Consumer,(void*)&index);

I still have some doubts as to how the Consumer or the Producer thread will receive the passed integer value since that is an address of the index variable in the for loop in the main thread. As soon as index gets incremented, the Consumer or Producer thread is affected by this increment.

Comments

[Twity 56876]

Its working.But why it does not work if we place sem_wait(&shared.mutex) and sem_post(&shared.mutex); instead of pthread_mutex_lock(&shared.mutex); and pthread_mutex_unlock(&shared.mutex);???Please explain.

[user2859193]

You have to decide wether you use a semaphore or mutex, then you have to use sem_init instead of pthread_mutex_init and sould set the initial value to 1, as wait_sem decreases the value of the semaphore, if the Value is > 0 else it blocks until the Value gets > 0. What never happens.