C++11 concurrency: condition variables

In the previous post in this series we have seen the C++11 support for locks and in this post we continue on this topic with condition variables. A condition variable is a synchronization primitive that enables blocking of one or more threads until either a notification is received from another thread or a timeout or a spurious wake-up occurs.

There are two implementations of a condition variable that are provided by C++11:

  • condition_variable: requires any thread that wants to wait on it to acquire a std::unique_lock first.
  • condition_variable_any: is a more general implementation that works with any type that satisfies the condition of a basic lock (basically has a lock() and unlock() method). This might be more expensive to use (in terms of performance and operating system resources), therefore it should be preferred only if the additional flexibility it provides is necessary.

So how does a condition variable work?

  • There must be at least one thread that is waiting for a condition to become true. The waiting thread must first acquire a unique_lock. This lock is passed to the wait() method, that releases the mutex and suspends the thread until the condition variable is signaled. When that happens the thread is awaken and the lock is re-acquired.
  • There must be at least one thread that is signaling that a condition becomes true. The signaling can be done with notify_one() which unblocks one thread (any) that is waiting for the condition to be signaled or with notify_all which unblocks all the threads that are waiting for the condition.
  • Because of some complications in making the condition wake-up completely predictable on multiprocessor systems, spurious wake-ups can occur. That means a thread is awaken even if nobody signaled the condition variable. Therefore it is necessary to check if the condition is still true after the thread has awaken. And since spurious wake-ups can occur multiple times, that check must be done in a loop.

The code below shows an example of using a condition variable to synchronize threads: several “worker” threads may produce an error during their work and they put the error code in a queue. A “logger” thread processes these error codes, by getting them from the queue and printing them. The workers signal the logger when an error occurred. The logger is waiting on the condition variable to be signaled. To avoid spurious wakeups the wait happens in a loop where a boolean condition is checked.

Running this code produce an output that looks like this (notice this output is different with each run because each worker thread works, i.e. sleeps, for a random interval):

The wait() method seen above has two overloads:

  • one that only takes a unique_lock; this one releases the lock, blocks the thread and adds it to the queue of threads that are waiting on this condition variable; the thread wakes up when the the condition variable is signaled or when a spurious wakeup occurs. When any of those happen, the lock is reacquired and the function returns.
  • one that in addition to the unique_lock also takes a predicate that is used to loop until it returns false; this overload may be used to avoid spurious wakeups. It is basically equivalent to:

As a result the use of the boolean flag g_notified in the example above can be avoided by using the wait overload that takes a predicate that verifies the state of the queue (empty or not):

In addition to this wait() overloaded method there are two more waiting methods, both with similar overloads that take a predicate to avoid spurious wake-ups:

  • wait_for: blocks the thread until the condition variable is signaled or the specified timeout occurred.
  • wait_until: blocks the thread until the condition variable is signaled or the specified moment in time was reached.

The overload without a predicate of these two functions returns a cv_status that indicates whether a timeout occurred or the wake-up happened because the condition variable was signaled or because of a spurious wake-up.

The standard also provides a function called notified_all_at_thread_exit that implements a mechanism to notify other threads that a given thread has finished, including destroying all thread_local objects. This was introduced because waiting on threads with other mechanisms than join() could lead to incorrect and fatal behavior when thread_locals were used, since their destructors could have been called even after the waiting thread resumed and possible also finished (see N3070 and N2880 for more). Typically, a call to this function must happen just before the thread exists.

Below is an example of how notify_all_at_thread_exit can be used together with a condition_variable to synchronize two threads:

That would output either (if the worker finishes work before main)

or (if the main finishes work before the worker):

2 thoughts on “C++11 concurrency: condition variables

  1. Erik

    Thanks for the great article. Tried it on Fedora using g++ and it worked great. I look forward to experimenting with it. Compile line:

    g++ -std=c++11 -D_GLIBCXX_USE_NANOSLEEP -pthread concurrency.c++ -o concurrency

    Reply
  2. Erik

    I know this is just example code, but there is a race condition in this:

    std::thread loggerthread(loggerfunc);

    // start the working threads
    std::vector threads;
    for(int i = 0; i < 5; ++i)
    {
    threads.push_back(std::thread(workerfunc, i+1, std::ref(generator)));
    }

    It is possible that the logger thread will take too long to start up and that the worker threads will have already signalled some errors and that those errors will then be lost. This can be solved (as some of the other synchronization is handled) with a global variable that indicates whether the logger is ready to serve requests or not. Add "g_loggerReady = true;" in loggerfunc() wherever you consider that the logger is ready to handle requests and then change main() to wait for that state change.

    // start the logger
    g_loggerReady = false;
    std::thread loggerthread(loggerfunc);
    while ( !g_loggerReady )
    {
    cout << "Sleeping to wait for logger…" << endl;
    std::this_thread::sleep_for(std::chrono::seconds(1));
    cout << "…done sleeping" << endl;
    }

    // start the worker threads
    std::vector threads;
    for(int i = 0; i < 5; ++i)
    {
    threads.push_back(std::thread(workerfunc, i+1, std::ref(generator)));
    }

    If it is absolutely critical that the worker threads start as soon as possible then you could move the blocking check to the point where the worker threads signal an error. This way the workers can get started immediately on their work and really only find themselves waiting if they need to signal an error before the logger thread is running. Slightly more complex, but it gets the system started faster (especially if the logger is very slow to start or might be blocked on acquiring a resource).

    Reply

Leave a Reply