Multiple producer & consumer using 3 sempahores

Question

With the given shared buffer of size 24,

Here is the below solution, for single producer & consumer using counting semaphore,

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

#define BUFFERSIZE 24

void *producer(void *);
void *consumer(void *);

sem_t mutex;          //Controls buffer access
sem_t fullBuffer;     //Prevents underflow
sem_t emptyBuffer;    // Prevents overflow

char buf[BUFFERSIZE] = {0}; //Shared resource

int main(int argc, char *argv[])
{
    pthread_t thread1, thread2;

    sem_init(&mutex, 0, 1);
    sem_init(&fullBuffer, 0, 0);
    sem_init(&emptyBuffer, 0, BUFFERSIZE);

    int ret1, ret2;

    ret1 = pthread_create(&thread1, NULL, producer, (void *)0);
    ret2 = pthread_create(&thread2, NULL, consumer, (void *)0);

    printf("Main function after pthread create\n");

    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    sem_destroy(&mutex);
    sem_destroy(&fullBuffer);
    sem_destroy(&emptyBuffer);

    return 0;
}

void *producer(void *index)
{
    int * startIndex = index;
    int currentProduceIndex = *startIndex;
    while(1) {
        sem_wait(&emptyBuffer); // Prevents overflow, when value is 0
        sem_wait(&mutex);       // Controls buffer access
        buf[currentProduceIndex] = currentProduceIndex;
        sem_post(&mutex);
        sem_post(&fullBuffer);
        currentProduceIndex = (currentProduceIndex+1) % BUFFERSIZE;
    }
}

void *consumer(void *index)
{
    int *startIndex = index;
    int currentConsumeIndex = *startIndex;
    while(1) {
        sem_wait(&fullBuffer); // Prevents underflow, when value is 0
        sem_wait(&mutex);
        printf("Consumed %d: ", buf[currentConsumeIndex]);
        sem_post(&mutex);
        sem_post(&emptyBuffer);
        currentConsumeIndex = (currentConsumeIndex+1) % BUFFERSIZE;
    }
}

Above code has single producer and consumer on buf, using 3 semaphores mutex, fullBuffer & emptyBuffer.

To add further, Does 3 semaphore, mutex, fullBuffer & emptyBuffer suffice to introduce multiple producers and consumers to access buf, in the above code?

Semaphores are hard. Why not use a pthread_cond_t and pthread_mutex_t ? Also, why do you want to pass startIndex as the thread parameter? As you have it now, you are passing nullptr, and it will crash when either thread deferences index. I could show a simple example if that's what you are after. — selbie
MayurK's answer below is close to what I would show. I left him a comment suggesting he just use one mutex and a few other tweaks. — selbie
@selbie Based on your expectation, Is your comment passed to Mayurk? — overexchange

MayurK MayurK · Accepted Answer · 2016-10-22T07:38:02

As seibie pointed out, it is hard to implement using semaphores. I have modified code and used two mutex and two cond variables. You can create multiple producer and consumer threads and it should work fine.

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

#define BUFFERSIZE 24

void *producer(void *);
void *consumer(void *);

pthread_mutex_t consMutex;
pthread_mutex_t prodMutex;
pthread_cond_t  consCond;
pthread_cond_t  prodCond;

// Shared resource: Using it to store flags.
// If value at index is 0, it means the index is free
// and producer can write data. If it is 1, it means
// that index is filled, consumer can consume it.
// You need use separate array for actual data.
char buf[BUFFERSIZE] = {0};

int main(void)
{
   pthread_t thread1, thread2;

    pthread_mutex_init(&consMutex, NULL);
    pthread_mutex_init(&prodMutex, NULL);

    pthread_cond_init(&consCond, NULL);
    pthread_cond_init(&prodCond, NULL);

    int ret1, ret2;

    int prodIndex = 0;
    int consIndex = 0;

    ret1 = pthread_create(&thread1, NULL, producer, (void *)&prodIndex);
    ret2 = pthread_create(&thread2, NULL, consumer, (void *)&consIndex);

    printf("Main function after pthread create\n");

    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    pthread_mutex_destroy(&prodMutex);
    pthread_mutex_destroy(&consMutex);

    pthread_cond_destroy(&prodCond);
    pthread_cond_destroy(&consCond);

    return 0;
}

void *producer(void *index)
{
    int *currentProduceIndex = index;
    while(1) {
        pthread_mutex_lock(&prodMutex);

        /* Check if current index is free to fill. */
        while(1 == buf[*currentProduceIndex])
        {
            /* If index is not free, wait for consumer to consume. */
            pthread_cond_wait(&prodCond, &prodMutex);
        }

        /* Now the current index is free. Fill it. */
        buf[*currentProduceIndex] = 1;

        /* Update the producer index. */
        *currentProduceIndex = (*currentProduceIndex+1)%BUFFERSIZE;
        pthread_mutex_unlock(&prodMutex);

        /* Notify consumer that an item has been produced. */
        pthread_mutex_lock(&consMutex);
        pthread_cond_signal(&consCond);
        pthread_mutex_unlock(&consMutex);
    }

    return NULL;
}

void *consumer(void *index)
{
    int *currentConsumeIndex = index;
    while(1) {

        pthread_mutex_lock(&consMutex);

        /* Check if current index is empty. */
        while(0 == buf[*currentConsumeIndex])
        {
            /* If index is empty, wait for producer to produce. */
            pthread_cond_wait(&consCond, &consMutex);
        }

        /* Index is filled, consume it. */
        buf[*currentConsumeIndex] = 0;

        /* Update the consumer index. */
        *currentConsumeIndex = (*currentConsumeIndex+1)%BUFFERSIZE;
        pthread_mutex_unlock(&consMutex);

        /* Notify producer that an item has been consumed. */
        pthread_mutex_lock(&prodMutex);
        pthread_cond_signal(&prodCond);
        pthread_mutex_unlock(&prodMutex);
    }

    return NULL;
}

Multiple producer & consumer using 3 sempahores

1 Answers