C++: condition-variable wait

Question

#include <thread>
#include <mutex>
#include <condition_variable>
#include <iostream>

std::mutex globalMutex;
std::condition_variable globalCondition;
int global = 0;
int activity = 0;
int CountOfThread = 1; // or more than 1

// just for console display, not effect the problem
std::mutex consoleMutex;

void producer() {
    while (true) {
        {
            std::unique_lock<std::mutex> lock(globalMutex);
            while (activity == 0) {
                lock.unlock();
                std::this_thread::yield();
                lock.lock();
            }
            global++;
            globalCondition.notify_one();
        }
        std::this_thread::yield();
    }
}


void customer() {
    while (true) {
        int x;
        {
            std::unique_lock<std::mutex> lock(globalMutex);
            activity++;
            globalCondition.wait(lock); // <- problem
            activity--;
            x = global;
        }
        {
            std::lock_guard<std::mutex> lock(consoleMutex);
            std::cout << x << std::endl;
        }
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}


int _tmain(int argc, _TCHAR* argv[])
{
    for (int i = 0; i < CountOfThread; ++i) {
        std::thread(customer).detach();
    }
    std::thread(producer).detach();
    getchar();
    return 0;
}

what i want is to make sure everytime there is a customer thread to get an increased global, expect display like: 1, 2, 3, ..., but what i see is the global value will be increased between wait and activity--, thus, actual display is: 1, 23, 56, 78, ....

I've found out the problem is in wait(), acutully there are 3 steps in wait(), 'unlock, wait, lock', between signaled(wait return) and mutex.lock, it's not a atomic operation, the producer thread may lock mutex before wait() to lock mutex, and the activity is still not zero, so the global will increased, unexpectedly

is there a way to make sure what i expect?

Answer 1

I find that it helps me to thing in the context of thread. For example, if you were a customer, what are you waiting on? By you, I mean in the context of thread. When you think this way, it makes it very simple to code with monitors.

Now, as Martin said, I believe that making repeated called to thread.yield is a little scary. That can lead to scary interleavings of the code.

To show an example as to why your code does not work, let's walk through a quick interleaving:

1. Several customers are created, and the one that grabs the lock increase activity. Then that thread goes to sleep due to the call to wait.
2. Another thread wakes up after the call to wait by the initial customer thread. This is because wait unlocks the mutex passed into it.
3. That thread acquires globalMutex and increases activity. Then it waits.
4. Repeat for CountOfThread number of threads, as this is entirely possible with multithreaded code.
5. Finally, the producer thread runs, sees activity == 0, and unlocks. However, it didn't notify_one (i.e. signal). Then it yields. This can lead to undefined and confusing code.

My suggestions:

1. Never call yield while waiting on a condition. This can lead to complicated and hard to read invalid monitor code.
2. Think about what condition each thread is waiting on. If different sections of code are waiting on different conditions, create a different lock. Only use one lock for one piece of shared code.
3. Use a different condition variable for different conditions in certain circumstances. If the condition is dependent on different data, definitely use a different condition variable.

In your circumstance, the solution is not as complicated as you might think. Use my original suggestion: think in the context of a thread. For example, when the producer thread is running, it wants to wait when the customer hasn't noticed that global has been changed. When a customer thread is running, it wants to wait whenever producer hasn't changed global.

Here is an example of the behavior you desire:

mutex m;

condition_variable cv;
int global = 0, prev_global = 0;

void producer()
{
    while (true)
    {
        unique_lock<mutex> lock(m);

        while (prev_global != global)
        {
            cv.wait(lock);
        }
        prev_global = global++;
        cv.notify_one();
        lock.unlock();
    }
}

void customer()
{
    while (true)
    {
        unique_lock<mutex> lock(m);

        while (prev_global == global)
        {
            cv.wait(lock);
        }

        prev_global = global;
        cv.notify_one();
        lock.unlock();
    }
}

int main()
{
    vector<thread> pool;
    for (int i = 0; i < 5; ++i)
    {   

        pool.push_back(thread (customer));
    }

    pool.push_back(thread (producer));

    for (auto it = pool.begin(); it != pool.end(); ++it)
    {
        it->join();
    }
    return 0;
}