C++ multithreading producer-consumer problem

Question

I wrote a code of multiply producers and consumers with condition variables. Even when I have only one producer and one consumer it doesn't work. Both producers and consumers should run in while(true). When I run the code it gets stuck around 50% of runs. I guess it gets in a deadlock caused by over wait. I don't succeed to debug where it is stucked and how to unlock the conds. By request, I must create the code with wait,signal and broadcast.

If the queue is full, producer is waiting. if the queue is empty, consumer is waiting.

void WaitableQueue::enqueue(size_t a_item)
{
    (m_cond.getMutex()).lock();

    while(m_itemsCounter==m_capacity && !m_isBeingDestroyed)
    {
        ++m_numberOfWaiting;
        m_cond.wait();
        --m_numberOfWaiting;
    }

    std::cout<<"enqueue "<<a_item<<"\n";

    m_queue.push(a_item);
    ++m_itemsCounter;
    ++m_numbOfProduced;
    if(m_isBeingDestroyed)
    {
        m_cond.broadcast(); 
    }

    (m_cond.getMutex()).unlock();
    m_cond.broadcast();
}

void WaitableQueue::dequeue()
{
    (m_cond.getMutex()).lock();

    while(m_itemsCounter==0 && !m_isBeingDestroyed)
    {
        ++m_numberOfWaiting;
        std::cout<<"Waiting\n";
        m_cond.wait();
        std::cout<<"Done waiting\n";
        --m_numberOfWaiting;
    }

    if (m_isBeingDestroyed)
    {
        (m_cond.getMutex()).unlock();
        m_cond.broadcast();
        return;
    }
    std::cout<<"dequeue "<<m_queue.front()<<"\n";
    m_queue.pop();
    --m_itemsCounter;
    ++m_numbOfConsumed;
    (m_cond.getMutex()).unlock();
    m_cond.broadcast();
}

void WaitableQueue::destroy()
{
    (m_cond.getMutex()).lock();
    m_isBeingDestroyed=true;
    (m_cond.getMutex()).unlock();
}



void Producer::run()
{
    for(size_t i=0;i<m_numOfItemsToProduce;++i)
    {
        usleep(m_delay);
        size_t item=produce();
        m_wq.enqueue(item);
    }
}


Producer::produce() const
{
    return rand()%m_numOfItemsToProduce;
}

void Consumer::run()
{
    m_numOfProducersMutex.lock();
    while(m_numOfProducers>0)
    {
        m_numOfProducersMutex.unlock();
        usleep(m_delay);
        m_wq.dequeue();
        m_numOfProducersMutex.lock();
    }
    m_numOfProducersMutex.unlock();
}


int main()
{
    size_t numProducers=1;
    size_t numConsumers=3;
    Mutex mutex;
    ConditionalVariable cond(mutex);

    WaitableQueue<size_t> wq(NUM_OF_ITEMS,cond);
    std::vector<Producer<size_t>*> producerArray;
    std::vector<Consumer<size_t>*> consumerArray;
    Mutex numOfProducersMutex;

    for(size_t i=0;i<numProducers;++i)
    {
        Producer<size_t>* tempP=new Producer<size_t>(wq,NUM_OF_ITEMS,DELAY);
        producerArray.push_back(tempP);
    }

    for(size_t i=0;i<numConsumers;++i)
    {
        Consumer<size_t>* tempC=new Consumer<size_t>(wq,numProducers,numOfProducersMutex,DELAY);
        consumerArray.push_back(tempC);
    }

    for(size_t i=0;i<numProducers;++i)
    {
        producerArray[i]->start();
    }

    for(size_t i=0;i<numConsumers;++i)
    {
        consumerArray[i]->start();
    }

    for(size_t i=0;i<numProducers;++i)
    {
        producerArray[i]->join();
        numOfProducersMutex.lock();
        --numProducers;
        numOfProducersMutex.unlock();
    }
    usleep(100);

    //tell the consumers stop waiting
    wq.destroy();
   for(size_t i=0;i<numConsumers;++i)
    {
        consumerArray[i]->join();
    }

   for(size_t i=0;i<numProducers;++i)
   {
        delete producerArray[i];
   }

    for(size_t i=0;i<numConsumers;++i)
   {
        delete consumerArray[i];
   }
}

It works around 50% of runnings. In other 50% it gets stucked.

Recommendation (that probably won't fix your current problem (but might fix a future one)): Don't lock and unlock the mutex yourself. Ensure that the mutex is unlocked with a RAII helper like std::lock_guard — user4581301
Looking over the code a bit more, I think you should take a look at the example here that shows a better way to use condition variables. — user4581301
You are not unlocking the mutex after locking it when you're waiting for empty/full queue. You need to release the mutex before you go waiting. If you debug the code, you will find that one thread is waiting on CV while locking the mutex, all while the other thread is waiting to get the mutex. Unlock mutex before you call CV.wait. This may have some race conditions so be careful. — Everyone
@Everyone Actually the mutex should always be locked before calling CV wait. In this case it looks like there is a custom class wrapper around CV that associates a mutex with the CV. (This is fine as long as you always protect the CV loop-waiting condition with that lock, which is the case here) — Humphrey Winnebago
Please provide the rest of the code. Note that if you pause the debugger while it's deadlocked, it will show you where the deadlock is. — Humphrey Winnebago

Ashish Khurange Ashish Khurange · Accepted Answer · 2019-10-07T06:40:13

To solve the producer-consumer problem using condition variable, first you need to understand bounded buffer problem.

Check here the implementation of thread safe buffer queue using condition variable in C++: https://codeistry.wordpress.com/2018/03/08/buffer-queue-handling-in-multithreaded-environment/

You can use this buffer queue as the building block to solve the multiple produce consumer problem. Please check here how the thread safe buffer queue is used to solve producer-consumer problem in C++: https://codeistry.wordpress.com/2018/03/09/unordered-producer-consumer/

C++ multithreading producer-consumer problem

2 Answers