One producer, two consumers acting on one 'queue' produced by producer

Question

Preface: I'm new to multithreaded programming, and a little rusty with C++. My requirements are to use one mutex, and two conditions mNotEmpty and mEmpty. I must also create and populate the vectors in the way mentioned below.

I have one producer thread creating a vector of random numbers of size n*2, and two consumers inserting those values into two separate vectors of size n.

I am doing the following in the producer:

1. Lock the mutex: pthread_mutex_lock(&mMutex1)
2. Wait for consumer to say vector is empty: pthread_cond_wait(&mEmpty,&mMutex1)
3. Push back a value into the vector
4. Signal the consumer that the vector isn't empty anymore: pthread_cond_signal(&mNotEmpty)
5. Unlock the mutex: pthread_mutex_unlock(&mMutex1)
6. Return to step 1

In the consumer:

1. Lock the mutex: pthread_mutex_lock(&mMutex1)
2. Check to see if the vector is empty, and if so signal the producer: pthread_cond_signal(&mEmpty)
3. Else insert value into one of two new vectors (depending on which thread) and remove from original vector
4. Unlock the mutex: pthread_mutex_unlock(&mMutex1)
5. Return to step 1

What's wrong with my process? I keep getting segmentation faults or infinite loops.

Edit: Here's the code:

void Producer()
{
    srand(time(NULL));
    for(unsigned int i = 0; i < mTotalNumberOfValues; i++){
        pthread_mutex_lock(&mMutex1);

        pthread_cond_wait(&mEmpty,&mMutex1);
        mGeneratedNumber.push_back((rand() % 100) + 1);

        pthread_cond_signal(&mNotEmpty);
        pthread_mutex_unlock(&mMutex1);
    }
}

void Consumer(const unsigned int index)
{
    for(unsigned int i = 0; i < mNumberOfValuesPerVector; i++){
        pthread_mutex_lock(&mMutex1);
        if(mGeneratedNumber.empty()){
            pthread_cond_signal(&mEmpty);
        }else{
            mThreadVector.at(index).push_back[mGeneratedNumber.at(0)];
            mGeneratedNumber.pop_back();
        }

        pthread_mutex_unlock(&mMutex1);
    }
}

Answer 1

I'm not sure I understand the rationale behind the way you're doing things. In the usual consumer-provider idiom, the provider pushes as many items as possible into the channel, waiting only if there is insufficient space in the channel; it doesn't wait for empty. So the usual idiom would be:

provider (to push one item):

pthread_mutex_lock( &mutex );
while ( ! spaceAvailable() ) {
    pthread_cond_wait( &spaceAvailableCondition, &mutex );
}
pushTheItem();
pthread_cond_signal( &itemAvailableCondition );
pthread_mutex_unlock( &mutex );

and on the consumer side, to get an item:

pthread_mutex_lock( &mutex );
while ( ! itemAvailable() ) {
    pthread_cond_wait( &itemAvailableCondition, &mutex );
}
getTheItem();
pthread_cond_signal( &spaceAvailableCondition );
pthread_mutex_unlock( &mutex );

Note that for each condition, one side signals, and the other waits. (I don't see any wait in your consumer.) And if there is more than one process on either side, I'd recommend using pthread_cond_broadcast, rather than pthread_cond_signal.

There are a number of other issues in your code. Some of them look more like typos: you should copy/paste actual code to avoid this. Do you really mean to read and pop mGeneratedValues, when you push into mGeneratedNumber, and check whether that is empty? (If you actually do have two different queues, then you're popping from a queue where no one has pushed.) And you don't have any loops waiting for the conditions; you keep iterating through the number of elements you expect (incrementing the counter each time, so you're likely to gerninate long before you should)—I can't see an infinite loop, but I can readily see a endless wait in pthread_cond_wait in the producer. I don't see a core dump off hand, but what happens when one of the processes terminates (probably the consumer, because it never waits for anything); if it ends up destroying the mutex or the condition variables, you could get a core dump when another process attempts to use them.