A producer consumer model with multiple producers and consumers.

There are no compilation errors, but it crashes as soon as it runs.

I run similar code with pthread and nothing goes wrong. I think there's something go wrong with unique_lock, but I can't find it out.

My environment is WIN10 VS2019.

Thank for your help.

#include <queue>
#include <iostream>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <memory>
#include <Windows.h>
#include <vector>

using namespace std;

template <class T>
class BlockQueue
{
public:

    BlockQueue(int size) : sz(size) {}
private:
    bool IsEmpty()
    {
        return q.size() == 0;
    }
    bool IsFull()
    {
        return q.size() == sz;
    }
    void Lock()
    {
        unique_lock<mutex> lck(mtx);
        lck.lock();
    }
    void UnLock()
    {
        unique_lock<mutex> lck(mtx);
        lck.unlock();
    }
    void CustomerWait()
    {
        unique_lock<mutex> ulk(mtx);
        is_full.wait(ulk);
    }
    void ProducterWait()
    {
        unique_lock<mutex> ulk(mtx);
        is_empty.wait(ulk);// condition_variable::wait need a unique_lock 
    }
    void WakeUpCustomer()
    {
        is_full.notify_one();
    }
    void WakeUpProducter()
    {
        is_empty.notify_one();
    }
    size_t size() const
    {
        return q.size();
    }
public:
    void push(const T& t)
    {
        Lock();
        while (IsFull())
        {
            ProducterWait();
        }
        q.push(t);
        WakeUpCustomer();
        UnLock();
    }
    void pop(T& out)
    {
        Lock();
        while (IsEmpty())
        {
            CustomerWait();
        }
    
        out = q.front();
        q.pop();
        WakeUpProducter();
        UnLock();
    }
private:
    queue<T> q;
    mutex mtx;
    condition_variable is_empty;
    condition_variable is_full;
    const int sz;// BlockQueue size
};

int main()
{
    try
    {
        shared_ptr<BlockQueue<int>> sp(new BlockQueue<int>(10));
        int n = 5;
        vector<thread> t1(n);
        vector<thread> t2(n);
        for (int i = 0; i < n;   i)
        {
            t1[i] = thread([sp] {
                while (true)
                {
                    int val = rand() % 10;
                    sp->push(val);
                    Sleep(1000);
                }
                });
            t2[i] = thread([sp] {
                int val;
                while (true)
                {
                    sp->pop(val);
                    // cout << "val = " << val << endl;
                    Sleep(500);
                }
                });
        }
        for (int i = 0; i < n;   i)
        {
            t1[i].join();
            t2[i].join();
        }
    }
    catch (const exception& e)
    {
        cout << e.what() << endl;
    }
    return 0;
}

CodePudding user response：

Your lock an unlock functions use local unique_lock objects. As such, they don't lock much. They only lock for their scope. They unlock as soon as they reach end of scope.

Read the doc, then use member object.

CodePudding user response：

I change my code and it runs right. The unique_lock may went wrong when recursively calls the lock in the condition_variable.

#include <queue>
#include <iostream>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <memory>
#include <Windows.h>
#include <vector>

using namespace std;

template <class T>
class BlockQueue
{
public:
    BlockQueue(int size) : sz(size) {}
private:
    bool IsEmpty()
    {
        return q.size() == 0;
    }
    bool IsFull()
    {
        return q.size() == sz;
    }
    void Lock()
    {   
        mtx.lock();
    }
    void UnLock()
    {
        mtx.unlock();
    }
    void CustomerWait()
    {
        unique_lock<mutex> lck(mtx, std::try_to_lock);
        is_full.wait(lck);
    }
    void ProducterWait()
    {
        unique_lock<mutex> lck(mtx, std::try_to_lock);
        is_empty.wait(lck);
    }
    void WakeUpCustomer()
    {
        is_full.notify_one();
    }
    void WakeUpProducter()
    {
        is_empty.notify_one();
    }
    size_t size() const
    {
        return q.size();
    }
public:
    void push(const T& t)
    {
        Lock();
        while (IsFull())
        {
            ProducterWait();
        }
        q.push(t);
        WakeUpCustomer();
        UnLock();
    }
    void pop(T& out)
    {
        Lock();
        while (IsEmpty())
        {
            CustomerWait();
        }
        out = q.front();
        q.pop();
        WakeUpProducter();
        UnLock();
    }
private:
    queue<T> q;
    mutex mtx;
    condition_variable is_empty;
    condition_variable is_full;
    const int sz;
};

int main()
{
    try
    {
        shared_ptr<BlockQueue<int>> sp(new BlockQueue<int>(10));
        int n = 5;
        srand((unsigned int)time(nullptr));
        vector<thread> t1(n);
        vector<thread> t2(n);
        for (int i = 0; i < n;   i)
        {
            t1[i] = thread([sp, i] {
                while (true)
                {
                    int val = rand() % 10   i;
                    sp->push(val);
                    Sleep(1000);
                }
                });
            t2[i] = thread([sp] {
                int val;
                while (true)
                {
                    sp->pop(val);
                    cout << "val = " << val << endl;
                    Sleep(500);
                }
                });
        }
        for (int i = 0; i < n;   i)
        {
            t1[i].join();
            t2[i].join();
        }
    }
    catch (const exception& e)
    {
        cout << e.what() << endl;
    }
    return 0;
}