Using a C++11 condition variable in VS2012

Question

I can't get code working reliably in a simple VS2012 console application consisting of a producer and consumer that uses a C++11 condition variable. I am aiming at producing a small reliable program (to use as the basis for a more complex program) that uses the 3 argument wait_for method or perhaps the wait_until method from code I have gathered at these websites:

condition_variable: wait_for, wait_until

I'd like to use the 3 argument wait_for with a predicate like below except it will need to use a class member variable to be most useful to me later. I am receiving "Access violation writing location 0x__" or "An invalid parameter was passed to a service or function" as errors after only about a minute of running.

Would steady_clock and the 2 argument wait_until be sufficient to replace the 3 argument wait_for? I've also tried this without success.

Can someone show how to get the code below to run indefinitely with no bugs or weird behavior with either changes in wall-clock time from daylight savings time or Internet time synchronizations?

A link to reliable sample code could be just as helpful.

// ConditionVariable.cpp : Defines the entry point for the console application.
//

#include "stdafx.h"

#include <condition_variable>
#include <mutex>
#include <thread>
#include <iostream>
#include <queue>
#include <chrono>
#include <atomic>

#define TEST1

std::atomic<int> 
//int 
    qcount = 0; //= ATOMIC_VAR_INIT(0);

int _tmain(int argc, _TCHAR* argv[])
{
    std::queue<int> produced_nums;
    std::mutex m;
    std::condition_variable cond_var;
    bool notified = false;
    unsigned int count = 0;

    std::thread producer([&]() {
        int i = 0;
        while (1) {
            std::this_thread::sleep_for(std::chrono::microseconds(1500));
            std::unique_lock<std::mutex> lock(m);
            produced_nums.push(i);
            notified = true;
            qcount = produced_nums.size();
            cond_var.notify_one();
            i++;
        }   
        cond_var.notify_one();
    }); 

    std::thread consumer([&]() {
        std::unique_lock<std::mutex> lock(m);
        while (1) {
#ifdef TEST1
            // Version 1
            if (cond_var.wait_for(
                lock,
                std::chrono::microseconds(1000),
                [&]()->bool { return qcount != 0; }))
            {
                if ((count++ % 1000) == 0)
                    std::cout << "consuming " << produced_nums.front    () << '\n';
                produced_nums.pop();
                qcount = produced_nums.size();
                notified = false;
            }
#else
            // Version 2
            std::chrono::steady_clock::time_point timeout1 =
                std::chrono::steady_clock::now() +
                //std::chrono::system_clock::now() +
                std::chrono::milliseconds(1);

            while (qcount == 0)//(!notified)
            {
                if (cond_var.wait_until(lock, timeout1) == std::cv_status::timeout)
                    break;
            }

            if (qcount > 0)
            {
                if ((count++ % 1000) == 0)
                std::cout << "consuming " << produced_nums.front() << '\n';
                produced_nums.pop();
                qcount = produced_nums.size();
                notified = false;
            }
#endif
        }
    });

    while (1);
    return 0;
}

Visual Studio Desktop Express had 1 important update which it installed and Windows Update has no other important updates. I'm using Windows 7 32-bit.

Answer 1

Sadly, this is actually a bug in VS2012's implementation of condition_variable, and the fix will not be patched in. You'll have to upgrade to VS2013 when it's released.

See:

http://connect.microsoft.com/VisualStudio/feedback/details/762560

Answer 2

First of all, while using condition_variables I personally prefer some wrapper classes like AutoResetEvent from C#:

struct AutoResetEvent
{
    typedef std::unique_lock<std::mutex> Lock;

    AutoResetEvent(bool state = false) :
        state(state)
    { }

    void Set()
    {
        auto lock = AcquireLock();
        state = true;
        variable.notify_one();
    }

    void Reset()
    {
        auto lock = AcquireLock();
        state = false;
    }

    void Wait(Lock& lock)
    {
        variable.wait(lock, [this] () { return this->state; });
        state = false;
    }

    void Wait()
    {
        auto lock = AcquireLock();
        Wait(lock);
    }

    Lock AcquireLock()
    {
        return Lock(mutex);
    }
private:

    bool state;
    std::condition_variable variable;
    std::mutex mutex;
};

This may not be the same behavior as C# type or may not be as efficient as it should be but it gets things done for me.

Second, when I need to implement a producing/consuming idiom I try to use a concurrent queue implementation (eg. tbb queue) or write a one for myself. But you should also consider making things right by using Active Object Pattern. But for simple solution we can use this:

template<typename T>
struct ProductionQueue
{
    ProductionQueue()
    { }

    void Enqueue(const T& value)
    {
        {
            auto lock = event.AcquireLock();
            q.push(value);
        }
        event.Set();
    }

    std::size_t GetCount()
    {
        auto lock = event.AcquireLock();

        return q.size();
    }

    T Dequeue()
    {
        auto lock = event.AcquireLock();
        event.Wait(lock);

        T value = q.front();
        q.pop();

        return value;
    }

private:
    AutoResetEvent event;
    std::queue<T> q;
};

This class has some exception safety issues and misses const-ness on the methods but like I said, for a simple solution this should fit.

So as a result your modified code looks like this:

int main(int argc, char* argv[])
{
    ProductionQueue<int> produced_nums;
    unsigned int count = 0;

    std::thread producer([&]() {
        int i = 0;
        while (1) {
            std::this_thread::sleep_for(std::chrono::microseconds(1500));
            produced_nums.Enqueue(i);
            qcount = produced_nums.GetCount();
            i++;
        }
    }); 

    std::thread consumer([&]() {
        while (1) {
            int item = produced_nums.Dequeue();
            {
                if ((count++ % 1000) == 0)
                    std::cout << "consuming " << item << '\n';
                qcount = produced_nums.GetCount();
            }
        }
    });

    producer.join();
    consumer.join();

    return 0;
}