Race Conditions

Race conditions can occur when multiple threads want to access any kind of shared resources. Race conditions in the context of shared memory are called data races. A data race can occur when multiple threads access shared memory, and at least one thread writes to the shared memory. For example, suppose you have a shared variable and one thread increments this value while another thread decrements it. Incrementing and decrementing the value means that the current value needs to be retrieved from memory, incremented or decremented, and stored back in memory. On old architectures, such as PDP-11 and VAX, this used to be implemented with an INC processor instruction, which was atomic. On modern x86 processors, the INC instruction is not atomic anymore, meaning that other instructions can be executed in the middle of this operation, which might cause the code to retrieve a wrong value.

The following table shows the result when the increment is finished before the decrement starts, and assumes that the initial value is 1.

The final value stored in memory is 1. When the decrement thread is finished before the increment thread starts, the final value is also 1, as shown in the following table.

However, when the instructions are interleaved, the result is different, as shown in the following table.

The final result in this case is 0. In other words, the effect of the increment operation is lost. This is a data race.

Tearing

Tearing is a specific case or consequence of a data race. There are two kinds of tearing: torn read and torn write. If a thread has written part of your data to memory, while another part hasn’t been written yet by the same thread, any other thread reading that data at that exact moment sees inconsistent data, a torn read. If two threads are writing to the data at the same time, one thread might have written part of the data, while another thread might have written another part of the data. The final result will be inconsistent, a torn write.

Deadlocks

If you opt to solve a race condition by using a synchronization method, such as mutual exclusion, you might run into another common problem with multithreaded programming: deadlocks. Two threads are deadlocked if they are both waiting for the other thread to do something. This can be extended to more than two threads. For example, if two threads want to acquire access to a shared resource, they need to ask for permission to access this resource. If one of the threads currently holds the permission to access the resource, but is blocked indefinitely for some other reason, then the other thread will block indefinitely as well when trying to acquire permission for the same resource. One mechanism to acquire permission for a shared resource is called a mutual exclusion object, discussed in detail later in this chapter. For example, suppose you have two threads and two resources protected with two mutual exclusion objects, A and B. Both threads acquire permission for both resources, but they acquire the permission in different order. The following table shows this situation in pseudo-code.

Now, imagine that the code in the two threads is executed in the following order.

• Thread 1: Acquire A
• Thread 2: Acquire B
• Thread 1: Acquire B (waits/blocks, because B is held by thread 2)
• Thread 2: Acquire A (waits/blocks, because A is held by thread 1)

Both threads are now waiting indefinitely in a deadlock situation. Figure 23-2 shows a graphical representation of the deadlock. Thread 1 has acquired permission for resource A and is waiting to acquire permission for resource B. Thread 2 has acquired permission for resource B and is waiting to acquire permission for resource A. In this graphical representation, you see a cycle that depicts the deadlock. Both threads will wait indefinitely.

It’s best to always acquire permissions in the same order to avoid these kinds of deadlocks. You can also include mechanisms in your program to break these deadlocks. One possible solution is to try for a certain time to acquire permission for a resource. If the permission could not be obtained within a certain time interval, the thread could stop waiting and possibly releases other permissions it is currently holding. The thread might then sleep for a little bit and try again later to acquire all the resources it needs. This method might give other threads the opportunity to acquire necessary permissions and continue their execution. Whether this method works or not depends heavily on your specific deadlock case.

Instead of using a workaround as described in the previous paragraph, you should try to avoid any possible deadlock situation altogether. If you need to acquire permission to multiple resources protected with several mutual exclusion objects, instead of acquiring permission for each resource individually, the recommended way is to use the standard std::lock() or std::try_lock() functions described later in the section “Mutual Exclusion.” These functions obtain or try to obtain permission for several resources with one call.

False-Sharing

Most caches work with so-called cache lines. For modern CPUs, cache lines are usually 64 bytes. If something needs to be written to a cache line, the entire line needs to be locked. This can bring a serious performance penalty for multithreaded code if your data structure is not properly designed. For example, if two threads are using two different pieces of data, but that data shares a cache line, then when one thread writes something, the other thread is blocked because the entire cache line is locked. You can optimize your data structures by using explicit memory alignments to make sure data that is worked on by multiple threads does not share any cache lines. To do this in a portable manner, C++17 introduces a constant called hardware_destructive_interference_size, defined in <new>, which returns you the minimum recommended offset between two concurrently accessed objects to avoid cache line sharing. You can use that value in combination with the alignas keyword to properly align your data.

Thread with Function Pointer

Functions such as CreateThread(), _beginthread(), and so on, on Windows, and pthread_create() with the pthreads library, require that the thread function has only one parameter. On the other hand, a function that you want to use with the standard C++ std::thread class can have as many parameters as you want.

Suppose you have a counter() function accepting two integers: the first representing an ID and the second representing the number of iterations that the function should loop. The body of the function is a single loop that loops the given number of iterations. On each iteration, a message is printed to standard output:

void counter(int id, int numIterations)
{
    for (int i = 0; i < numIterations; ++i) {
        cout << "Counter " << id << " has value " << i << endl;
    }
}

You can launch multiple threads executing this function using std::thread. You can create a thread t1, executing counter() with arguments 1 and 6 as follows:

thread t1(counter, 1, 6);

The constructor of the thread class is a variadic template, which means that it accepts any number of arguments. Variadic templates are discussed in detail in Chapter 22. The first argument is the name of the function to execute in the new thread. The subsequent variable number of arguments are passed to this function when execution of the thread starts.

A thread object is said to be joinable if it represents or represented an active thread in the system. Even when the thread has finished executing, a thread object remains in the joinable state. A default constructed thread object is unjoinable. Before a joinable thread object is destroyed, you need to make sure to call either join() or detach() on it. A call to join() is a blocking call, it waits until the thread has finished its work. A call to detach() detaches a thread object from its underlying OS thread, in which case the OS thread keeps running independently. Both methods cause the thread to become unjoinable. If a thread object that is still joinable is destroyed, the destructor will call std::terminate(), which abruptly terminates all threads and the application itself.

The following code launches two threads executing the counter() function. After launching the threads, main() calls join() on both threads.

thread t1(counter, 1, 6);
thread t2(counter, 2, 4);
t1.join();
t2.join();

A possible output of this example looks as follows:

Counter 2 has value 0
Counter 1 has value 0
Counter 1 has value 1
Counter 1 has value 2
Counter 1 has value 3
Counter 1 has value 4
Counter 1 has value 5
Counter 2 has value 1
Counter 2 has value 2
Counter 2 has value 3

The output on your system will be different and it will most likely be different every time you run it. This is because two threads are executing the counter() function at the same time, so the output depends on the number of processing cores in your machine and on the thread scheduling of the operating system.

By default, accessing cout from different threads is thread-safe and doesn’t cause any data races, unless you have called cout.sync_with_stdio(false) before the first output or input operation. However, even though there are no data races, output from different threads can still be interleaved! This means that the output of the previous example can be mixed together as follows:

Counter Counter 2 has value 0
1 has value 0
Counter 1 has value 1
Counter 1 has value 2
…

This can be fixed using synchronization methods, which are discussed later in this chapter.

Thread with Function Object

Instead of using function pointers, you can also use a function object to execute in a thread. With the function pointer technique of the previous section, the only way to pass information to the thread is by passing arguments to the function. With function objects, you can add member variables to your function object class, which you can initialize and use however you want. The following example first defines a class called Counter, which has two member variables: an ID and the number of iterations for the loop. Both variables are initialized with the constructor. To make the Counter class a function object, you need to implement operator(), as discussed in Chapter 18. The implementation of operator() is the same as the counter() function from the previous section. Here is the code:

class Counter
{
    public:
        Counter(int id, int numIterations)
            : mId(id), mNumIterations(numIterations)
        {
        }

        void operator()() const
        {
            for (int i = 0; i < mNumIterations; ++i) {
                cout << "Counter " << mId << " has value " << i << endl;
            }
        }
    private:
        int mId;
        int mNumIterations;
};

Three methods for initializing threads with a function object are demonstrated in the following code snippet. The first method uses the uniform initialization syntax. You create an instance of Counter with its constructor arguments and give it to the thread constructor between curly braces.

The second method defines a named instance of Counter and gives this named instance to the constructor of the thread class.

The third method looks similar to the first one; it creates an instance of Counter and gives it to the constructor of the thread class, but uses parentheses instead of curly braces. The ramifications of this are discussed after the code.

// Using uniform initialization syntax
thread t1{ Counter{ 1, 20 }};

// Using named variable
Counter c(2, 12);
thread t2(c);

// Using temporary
thread t3(Counter(3, 10));

// Wait for threads to finish
t1.join();
t2.join();
t3.join();

If you compare the creation of t1 with the creation of t3, the only difference seems to be that the first method uses curly braces while the third method uses parentheses. However, when your function object constructor doesn’t require any parameters, the third method as written earlier does not work. Here is an example:

class Counter
{
    public:
        Counter() {}
        void operator()() const { /* Omitted for brevity */ }
};

int main()
{
    thread t1(Counter());    // Error!
    t1.join();
}

This results in a compilation error because C++ interprets the first line in main() as a declaration of a function called t1, which returns a thread object and accepts a pointer to a function without parameters returning a Counter object. For this reason, it’s recommended to use the uniform initialization syntax:

thread t1{ Counter{} };  // OK

Thread with Lambda

Lambda expressions fit nicely with the standard C++ threading library. Here is an example that launches a thread to execute a given lambda expression:

int main()
{
    int id = 1;
    int numIterations = 5;
    thread t1([id, numIterations] {
        for (int i = 0; i < numIterations; ++i) {
            cout << "Counter " << id << " has value " << i << endl;
        }
    });
    t1.join();
}

Thread with Member Function

You can specify a member function of a class to be executed in a thread. The following example defines a basic Request class with a process() method. The main() function creates an instance of the Request class, and launches a new thread, which executes the process() method of the Request instance req.

class Request
{
    public:
        Request(int id) : mId(id) { }

        void process()
        {
            cout << "Processing request " << mId << endl;
        }
    private:
        int mId;
};

int main()
{
    Request req(100);
    thread t{ &Request::process, &req };
    t.join();
}

With this technique, you are executing a method on a specific object in a separate thread. If other threads are accessing the same object, you need to make sure this happens in a thread-safe way to avoid data races. Mutual exclusion, discussed later in this chapter, can be used as a synchronization mechanism to make it thread-safe.

Thread Local Storage

The C++ standard supports the concept of thread local storage. With a keyword called thread_local, you can mark any variable as thread local, which means that each thread will have its own unique copy of the variable and it will last for the entire duration of the thread. For each thread, the variable is initialized exactly once. For example, the following code defines two global variables. Every thread shares one—and only one—copy of k, while each thread has its own unique copy of n:

int k;
thread_local int n;

Note that if the thread_local variable is declared in the scope of a function, its behavior is as if it were declared static, except that every thread has its own unique copy and is initialized exactly once per thread, no matter how many times that function is called in that thread.

Cancelling Threads

The C++ standard does not include any mechanism for cancelling a running thread from another thread. The best way to achieve this is to provide some communication mechanism that the two threads agree upon. The simplest mechanism is to have a shared variable, which the target thread checks periodically to determine if it should terminate. Other threads can set this shared variable to indirectly instruct the thread to shut down. You have to be careful here, because this shared variable is being accessed by multiple threads, of which at least one is writing to the shared variable. It’s recommended to use atomic variables or condition variables, both discussed later in this chapter.

Retrieving Results from Threads

As you saw in the previous examples, launching a new thread is pretty easy. However, in most cases you are probably interested in results produced by the thread. For example, if your thread performs some mathematical calculations, you really would like to get the results out of the thread once the thread is finished. One way is to pass a pointer or reference to a result variable to the thread in which the thread stores the results. Another method is to store the results inside class member variables of a function object, which you can retrieve later once the thread has finished executing. This only works if you use std::ref() to pass your function object by reference to the thread constructor.

However, there is another easier method to obtain a result from threads: futures. Futures also make it easier to handle errors that occur inside your threads. They are discussed later in this chapter.

Copying and Rethrowing Exceptions

The whole exception mechanism in C++ works perfectly fine, as long as it stays within one single thread. Every thread can throw its own exceptions, but they need to be caught within their own thread. If a thread throws an exception and it is not caught inside the thread, the C++ runtime calls std::terminate(), which terminates the whole application. Exceptions thrown in one thread cannot be caught in another thread. This introduces quite a few problems when you would like to use exception handling in combination with multithreaded programming.

Without the standard threading library, it’s very difficult if not impossible to gracefully handle exceptions across threads. The standard threading library solves this issue with the following exception-related functions. These functions work not only with std::exceptions, but also with other kinds of exceptions, ints, strings, custom exceptions, and so on:

• exception_ptr current_exception() noexcept;

  This is intended to be called from inside a catch block. It returns an exception_ptr object that refers to the exception currently being handled, or a copy of the currently handled exception. A null exception_ptr object is returned if no exception is being handled. This referenced exception object remains valid for as long as there is an object of type exception_ptr that is referencing it. exception_ptr is of type NullablePointer, which means it can easily be tested with a simple if statement, as the example later in this section will demonstrate.

• [[noreturn]] void rethrow_exception(exception_ptr p);

  This function rethrows the exception referenced by the exception_ptr parameter. Rethrowing the referenced exception does not have to be done in the same thread that generated the referenced exception in the first place, which makes this feature perfectly suited for handling exceptions across different threads. The [[noreturn]] attribute makes it clear that this function never returns normally. Attributes are introduced in Chapter 11.

• template<class E> exception_ptr make_exception_ptr(E e) noexcept;

  This function creates an exception_ptr object that refers to a copy of the given exception object. This is basically a shorthand notation for the following code:

try {
    throw e;
} catch(…) {
    return current_exception();
}

Let’s see how handling exceptions across different threads can be implemented using these functions. The following code defines a function that does some work and throws an exception. This function will ultimately be running in a separate thread:

void doSomeWork()
{
    for (int i = 0; i < 5; ++i) {
        cout << i << endl;
    }
    cout << "Thread throwing a runtime_error exception…" << endl;
    throw runtime_error("Exception from thread");
}

The following threadFunc() function wraps the call to the preceding function in a try/catch block, catching all exceptions that doSomeWork() might throw. A single argument is supplied to threadFunc(), which is of type exception_ptr&. Once an exception is caught, the function current_exception() is used to get a reference to the exception being handled, which is then assigned to the exception_ptr parameter. After that, the thread exits normally:

void threadFunc(exception_ptr& err)
{
    try {
        doSomeWork();
    } catch (…) {
        cout << "Thread caught exception, returning exception…" << endl;
        err = current_exception();
    }
}

The following doWorkInThread() function is called from within the main thread. Its responsibility is to create a new thread and start executing threadFunc() in it. A reference to an object of type exception_ptr is given as argument to threadFunc(). Once the thread is created, the doWorkInThread() function waits for the thread to finish by using the join() method, after which the error object is examined. Because exception_ptr is of type NullablePointer, you can easily check it using an if statement. If it’s a non-null value, the exception is rethrown in the current thread, which is the main thread in this example. Because you are rethrowing the exception in the main thread, the exception has been transferred from one thread to another thread.

void doWorkInThread()
{
    exception_ptr error;
    // Launch thread
    thread t{ threadFunc, ref(error) };
    // Wait for thread to finish
    t.join();
    // See if thread has thrown any exception
    if (error) {
        cout << "Main thread received exception, rethrowing it…" << endl;
        rethrow_exception(error);
    } else {
        cout << "Main thread did not receive any exception." << endl;
    }
}

The main() function is pretty straightforward. It calls doWorkInThread() and wraps the call in a try/catch block to catch exceptions thrown by the thread spawned by doWorkInThread():

int main()
{
    try {
        doWorkInThread();
    } catch (const exception& e) {
        cout << "Main function caught: '" << e.what() << "'" << endl;
    }
}

The output is as follows:

0
1
2
3
4
Thread throwing a runtime_error exception…
Thread caught exception, returning exception…
Main thread received exception, rethrowing it…
Main function caught: 'Exception from thread'

To keep the examples in this chapter compact and to the point, their main() functions usually use join() to block the main thread, and to wait until threads have finished. Of course, in real-world applications you do not want to block your main thread. For example, in a GUI application, blocking your main thread means that the UI becomes unresponsive. In that case, you can use a messaging paradigm to communicate between threads. For example, the earlier threadFunc() function could send a message to the UI thread with as argument a copy of the result of current_exception(). But even then, you need to make sure to call either join() or detach() on any spawned threads, as discussed earlier in this chapter.

Atomic Type Example

This section explains in more detail why you should use atomic types. Suppose you have a function called increment() that increments an integer reference parameter in a loop. This code uses std::this_thread::sleep_for() to introduce a small delay in each loop. The argument to sleep_for() is an std::chrono::duration, which is explained in Chapter 20.

void increment(int& counter)
{
    for (int i = 0; i < 100; ++i) {
        ++counter;
        this_thread::sleep_for(1ms);
    }
}

Now, you would like to run several threads in parallel, all executing this increment() function on a shared counter variable. By implementing this naively without atomic types or without any kind of thread synchronization, you introduce data races. The following code launches ten threads, after which it waits for all threads to finish by calling join() on each thread:

int main()
{
    int counter = 0;
    vector<thread> threads;
    for (int i = 0; i < 10; ++i) {
        threads.push_back(thread{ increment, ref(counter) });
    }

    for (auto& t : threads) {
        t.join();
    }
    cout << "Result = " << counter <<endl;
}

Because increment() increments its given integer 100 times, and ten threads are launched, each of which executes increment() on the same shared counter, the expected result is 1,000. If you execute this program several times, you might get the following output but with different values:

Result = 982
Result = 977
Result = 984

This code is clearly showing a data race. In this example, you can use an atomic type to fix the code. The following code highlights the required changes:

#include <atomic>

void increment(atomic<int>& counter)
{
    for (int i = 0; i < 100; ++i) {
        ++counter;
        this_thread::sleep_for(1ms);
    }
}

int main()
{
    atomic<int> counter(0);
    vector<thread> threads;
    for (int i = 0; i < 10; ++i) {
        threads.push_back(thread{ increment, ref(counter) });
    }

    for (auto& t : threads) {
        t.join();
    }
    cout << "Result = " << counter << endl;
}

The changes add the <atomic> header file, and change the type of the shared counter to std::atomic<int> instead of int. When you run this modified version, you always get 1,000 as the result:

Result = 1000
Result = 1000
Result = 1000

Without explicitly adding any synchronization mechanism to the code, it is now thread-safe and data-race-free because the ++counter operation on an atomic type loads, increments, and stores the value in one atomic transaction, which cannot be interrupted.

However, there is a new problem with this modified code: a performance problem. You should try to minimize the amount of synchronization, either atomic or explicit synchronization, because it lowers performance. For this simple example, the best and recommended solution is to let increment() calculate its result in a local variable, and only after the loop, add it to the counter reference. Note that it is still required to use an atomic type, because you are still writing to counter from multiple threads.

void increment(atomic<int>& counter)
{
    int result = 0;
    for (int i = 0; i < 100; ++i) {
        ++result;
        this_thread::sleep_for(1ms);
    }
    counter += result;
}

Atomic Operations

The C++ standard defines a number of atomic operations. This section describes a few of those operations. For a full list, consult a Standard Library Reference, see Appendix B.

A first example of an atomic operation is the following:

bool atomic<T>::compare_exchange_strong(T& expected, T desired);

The logic implemented atomically by this operation is as follows, in pseudo-code:

if (*this == expected) {
    *this = desired;
    return true;
} else {
    expected = *this;
    return false;
}

Although this logic might seem fairly strange on first sight, this operation is a key building block for writing lock-free concurrent data structures. Lock-free concurrent data structures allow operations on their data without requiring any synchronization mechanisms. However, implementing such data structures is an advanced topic, outside the scope of this book.

A second example is atomic<T>::fetch_add(), which works for integral atomic types. It fetches the current value of the atomic type, adds the given increment to the atomic value, and returns the original non-incremented value. Here is an example:

atomic<int> value(10);
cout << "Value = " << value << endl;
int fetched = value.fetch_add(4);
cout << "Fetched = " << fetched << endl;
cout << "Value = " << value << endl;

If no other threads are touching the contents of the fetched and value variables, the output is as follows:

Value = 10
Fetched = 10
Value = 14

Atomic integral types support the following atomic operations: fetch_add(), fetch_sub(), fetch_and(), fetch_or(), fetch_xor(), ++, --, +=, -=, &=, ^=, and |=. Atomic pointer types support fetch_add(), fetch_sub(), ++, --, +=, and -=.

Most of the atomic operations can accept an extra parameter specifying the memory ordering that you would like. Here is an example:

T atomic<T>::fetch_add(T value, memory_order = memory_order_seq_cst);

You can change the default memory_order. The C++ standard provides memory_order_relaxed, memory_order_consume, memory_order_acquire, memory_order_release, memory_order_acq_rel, and memory_order_seq_cst, all of which are defined in the std namespace. However, you will rarely want to use them instead of the default. While another memory order may perform better than the default, according to some metrics, if you use them in a slightly incorrect way, you will again introduce data races or other difficult-to-track threading-related problems. If you do want to know more about memory ordering, consult one of the multithreading references in Appendix B.

Mutex Classes

Mutex stands for mutual exclusion. The basic mechanism of using a mutex is as follows:

• A thread that wants to use (read/write) memory shared with other threads tries to lock a mutex object. If another thread is currently holding this lock, the new thread that wants to gain access blocks until the lock is released, or until a timeout interval expires.
• Once the thread has obtained the lock, it is free to use the shared memory. Of course, this assumes that all threads that want to use the shared data correctly acquire a lock on the mutex.
• After the thread is finished with reading/writing to the shared memory, it releases the lock to give some other thread an opportunity to obtain the lock to the shared memory. If two or more threads are waiting on the lock, there are no guarantees as to which thread will be granted the lock and thus allowed to proceed.

The C++ standard provides non-timed mutex and timed mutex classes.

Locks

A lock class is an RAII class that makes it easier to correctly obtain and release a lock on a mutex; the destructor of the lock class automatically releases the associated mutex. The C++ standard defines four types of locks: std::lock_guard, unique_lock, shared_lock, and scoped_lock. The latter has been introduced with C++17.

template <class L1, class L2, class… L3> void lock(L1&, L2&, L3&…);

mutex mut1;
mutex mut2;

void process()
{
    unique_lock lock1(mut1, defer_lock);  // C++17
    unique_lock lock2(mut2, defer_lock);  // C++17
    //unique_lock<mutex> lock1(mut1, defer_lock);
    //unique_lock<mutex> lock2(mut2, defer_lock);
    lock(lock1, lock2);
    // Locks acquired
} // Locks automatically released

mutex mut1;
mutex mut2;

void process()
{
    scoped_lock locks(mut1, mut2);
    // Locks acquired
} // Locks automatically released

scoped_lock<mutex, mutex> locks(mut1, mut2);

std::call_once

You can use std::call_once() in combination with std::once_flag to make sure a certain function or method is called exactly one time, no matter how many threads try to call call_once() with the same once_flag. Only one call_once() invocation actually calls the given function or method. If the given function does not throw any exceptions, then this invocation is called the effective call_once() invocation. If the given function does throw an exception, the exception is propagated back to the caller, and another caller is selected to execute the function. The effective invocation on a specific once_flag instance finishes before all other call_once() invocations on the same once_flag. Other threads calling call_once() on the same once_flag block until the effective call is finished. Figure 23-3 illustrates this with three threads. Thread 1 performs the effective call_once() invocation, thread 2 blocks until the effective invocation is finished, and thread 3 doesn’t block because the effective invocation from thread 1 has already finished.

The following example demonstrates the use of call_once(). The example launches three threads running processingFunction() that use some shared resources. These shared resources should be initialized only once by calling initializeSharedResources() once. To accomplish this, each thread calls call_once() with a global once_flag. The result is that only one thread effectively executes initializeSharedResources(), and exactly one time. While this call_once() call is in progress, other threads block until initializeSharedResources() returns.

once_flag gOnceFlag;

void initializeSharedResources()
{
    // … Initialize shared resources to be used by multiple threads.
    cout << "Shared resources initialized." << endl;
}

void processingFunction()
{
    // Make sure the shared resources are initialized.
    call_once(gOnceFlag, initializeSharedResources);

    // … Do some work, including using the shared resources
    cout << "Processing" << endl;
}

int main()
{
    // Launch 3 threads.
    vector<thread> threads(3);
    for (auto& t : threads) {
        t = thread{ processingFunction };
    }
    // Join on all threads
    for (auto& t : threads) {
        t.join();
    }
} 

The output of this code is as follows:

Shared resources initialized.
Processing
Processing
Processing

Of course, in this example, you could call initializeSharedResources() once in the beginning of the main() function before the threads are launched; however, that wouldn’t demonstrate the use of call_once().

Examples Using Mutual Exclusion Objects

The following sections give a couple of examples on how to use mutual exclusion objects to synchronize multiple threads.

class Counter
{
    public:
        Counter(int id, int numIterations)
            : mId(id), mNumIterations(numIterations)
        {
        }

        void operator()() const
        {
            for (int i = 0; i < mNumIterations; ++i) {
                lock_guard lock(sMutex);
                cout << "Counter " << mId << " has value " << i << endl;
            }
        }
    private:
        int mId;
        int mNumIterations;
        static mutex sMutex;
};

mutex Counter::sMutex;

class Counter
{
    public:
        Counter(int id, int numIterations)
            : mId(id), mNumIterations(numIterations)
        {
        }

        void operator()() const
        {
            for (int i = 0; i < mNumIterations; ++i) {
                unique_lock lock(sTimedMutex, 200ms);
                if (lock) {
                    cout << "Counter " << mId << " has value " << i << endl;
                } else {
                    // Lock not acquired in 200ms, skip output.
                }
            }
        }
    private:
        int mId;
        int mNumIterations;
        static timed_mutex sTimedMutex;
};

timed_mutex Counter::sTimedMutex;

void initializeSharedResources()
{
    // … Initialize shared resources to be used by multiple threads.
    cout << "Shared resources initialized." << endl;
}

atomic<bool> gInitialized(false);
mutex gMutex;

void processingFunction()
{
    if (!gInitialized) {
        unique_lock lock(gMutex);
        if (!gInitialized) {
            initializeSharedResources();
            gInitialized = true;
        }
    }
    cout << "OK" << endl;
}

int main()
{
    vector<thread> threads;
    for (int i = 0; i < 5; ++i) {
        threads.push_back(thread{ processingFunction });
    }
    for (auto& t : threads) {
        t.join();
    }
}

Shared resources initialized.
OK
OK
OK
OK
OK

Spurious Wake-Ups

Threads waiting on a condition variable can wake up when another thread calls notify_one() or notify_all(), with a relative timeout, or when the system time reaches a certain time. However, they can also wake up spuriously. This means that a thread can wake up, even if no other thread has called any notify method, and no timeouts have been reached yet. Thus, when a thread waits on a condition variable and wakes up, it needs to check why it woke up. One way to check for this is by using one of the versions of wait() accepting a predicate, as shown in the following example.

Using Condition Variables

As an example, condition variables can be used for background threads processing items from a queue. You can define a queue in which you insert items to be processed. A background thread waits until there are items in the queue. When an item is inserted into the queue, the thread wakes up, processes the item, and goes back to sleep, waiting for the next item. Suppose you have the following queue:

queue<string> mQueue;

You need to make sure that only one thread is modifying this queue at any given time. You can do this with a mutex:

mutex mMutex;

To be able to notify a background thread when an item is added, you need a condition variable:

condition_variable mCondVar;

A thread that wants to add an item to the queue first acquires a lock on the mutex, then adds the item to the queue, and notifies the background thread. You can call notify_one() or notify_all() whether or not you currently have the lock; both work.

// Lock mutex and add entry to the queue.
unique_lock lock(mMutex);
mQueue.push(entry);
// Notify condition variable to wake up thread.
mCondVar.notify_all();

The background thread waits for notifications in an infinite loop, as follows. Note the use of wait() accepting a predicate to correctly handle spurious wake-ups. The predicate checks if there really is something in the queue. When the call to wait() returns, you are sure there is something in the queue.

unique_lock lock(mMutex);
while (true) {
    // Wait for a notification.
    mCondVar.wait(lock, [this]{ return !mQueue.empty(); });
    // Condition variable is notified, so something is in the queue.
    // Process queue item…
}

The section “Example: Multithreaded Logger Class,” toward the end of this chapter, provides a complete example on how to use condition variables to send notifications to other threads.

The C++ standard also defines a helper function called std::notify_all_at_thread_exit(cond, lk) where cond is a condition variable and lk is a unique_lock<mutex> instance. A thread calling this function should already have acquired the lock lk. When the thread exits, it automatically executes the following:

lk.unlock();
cond.notify_all();

std::promise and std::future

C++ provides the std::promise class as one way to implement the concept of a promise. You can call set_value() on a promise to store a result, or you can call set_exception() on it to store an exception in the promise. Note that you can only call set_value() or set_exception() once on a specific promise. If you call it multiple times, an std::future_error exception will be thrown.

A thread A that launches another thread B to calculate something can create an std::promise and pass this to the launched thread. Note that a promise cannot be copied, but it can be moved into a thread! Thread B uses that promise to store the result. Before moving the promise into thread B, thread A calls get_future() on the created promise to be able to get access to the result once B has finished. Here is a simple example:

void DoWork(promise<int> thePromise)
{
    // … Do some work …
    // And ultimately store the result in the promise.
    thePromise.set_value(42);
}

int main()
{
    // Create a promise to pass to the thread.
    promise<int> myPromise;
    // Get the future of the promise.
    auto theFuture = myPromise.get_future();
    // Create a thread and move the promise into it.
    thread theThread{ DoWork, std::move(myPromise) };

    // Do some more work…

    // Get the result.
    int result = theFuture.get();
    cout << "Result: " << result << endl;

    // Make sure to join the thread.
    theThread.join();
}

std::packaged_task

An std::packaged_task makes it easier to work with promises than explicitly using std::promise, as in the previous section. The following code demonstrates this. It creates a packaged_task to execute CalculateSum(). The future is retrieved from the packaged_task by calling get_future(). A thread is launched, and the packaged_task is moved into it. A packaged_task cannot be copied! After the thread is launched, get() is called on the retrieved future to get the result. This blocks until the result is available.

Note that CalculateSum() does not need to store anything explicitly in any kind of promise. A packaged_task automatically creates a promise, automatically stores the result of the called function, CalculateSum() in this case, in the promise, and automatically stores any exceptions thrown from the function in the promise.

int CalculateSum(int a, int b) { return a + b; }

int main()
{
    // Create a packaged task to run CalculateSum.
    packaged_task<int(int, int)> task(CalculateSum);
    // Get the future for the result of the packaged task.
    auto theFuture = task.get_future();
    // Create a thread, move the packaged task into it, and
    // execute the packaged task with the given arguments.
    thread theThread{ std::move(task), 39, 3 };

    // Do some more work…

    // Get the result.
    int result = theFuture.get();
    cout << result << endl;

    // Make sure to join the thread.
    theThread.join();
}

std::async

If you want to give the C++ runtime more control over whether or not a thread is created to calculate something, you can use std::async(). It accepts a function to be executed, and returns a future that you can use to retrieve the result. There are two ways in which async() can run your function:

• Run your function on a separate thread asynchronously
• Run your function on the calling thread synchronously at the time you call get() on the returned future

If you call async() without additional arguments, the runtime automatically chooses one of the two methods depending on factors such as the number of CPU cores in your system and the amount of concurrency already taking place. You can influence the runtime’s behavior by specifying a policy argument.

• launch::async: forces the runtime to execute the function asynchronously on a different thread.
• launch::deferred: forces the runtime to execute the function synchronously on the calling thread when get() is called.
• launch::async | launch::deferred: lets the runtime choose (= default behavior).

The following example demonstrates the use of async():

int calculate()
{
    return 123;
}

int main()
{
    auto myFuture = async(calculate);
    //auto myFuture = async(launch::async, calculate);
    //auto myFuture = async(launch::deferred, calculate);

    // Do some more work…

    // Get the result.
    int result = myFuture.get();
    cout << result << endl;
}

As you can see in this example, std::async() is one of the easiest methods to perform some calculations either asynchronously (on a different thread), or synchronously (on the same thread), and retrieve the result afterward.

Exception Handling

A big advantage of using futures is that they can transport exceptions between threads. Calling get() on a future either returns the calculated result, or rethrows any exception that has been stored in the promise linked to the future. When you use packaged_task or async(), any exception thrown from the launched function is automatically stored in the promise. If you use std::promise as your promise, you can call set_exception() to store an exception in it. Here is an example using async():

int calculate()
{
    throw runtime_error("Exception thrown from calculate().");
}

int main()
{
    // Use the launch::async policy to force asynchronous execution.
    auto myFuture = async(launch::async, calculate);

    // Do some more work…

    // Get the result.
    try {
        int result = myFuture.get();
        cout << result << endl;
    } catch (const exception& ex) {
        cout << "Caught exception: " << ex.what() << endl;
    }
}

std::shared_future

std::future<T> only requires T to be move-constructible. When you call get() on a future<T>, the result is moved out of the future and returned to you. This means you can call get() only once on a future<T>.

If you want to be able to call get() multiple times, even from multiple threads, then you need to use an std::shared_future<T>, in which case T needs to be copy-constructible. A shared_future can be created by using std::future::share(), or by passing a future to the shared_future constructor. Note that a future is not copyable, so you have to move it into the shared_future constructor.

shared_future can be used to wake up multiple threads at once. For example, the following piece of code defines two lambda expressions to be executed asynchronously on different threads. The first thing each lambda expression does is set a value to their respective promise to signal that they have started. Then they both call get() on signalFuture which blocks until a parameter is made available through the future, after which they continue their execution. Each lambda expression captures their respective promise by reference, and captures signalFuture by value, so both lambda expressions have a copy of signalFuture. The main thread uses async() to execute both lambda expressions asynchronously on different threads, waits until both threads have started, and then sets the parameter in the signalPromise which wakes up both threads.

promise<void> thread1Started, thread2Started;

promise<int> signalPromise;
auto signalFuture = signalPromise.get_future().share();
//shared_future<int> signalFuture(signalPromise.get_future());

auto function1 = [&thread1Started, signalFuture] {
    thread1Started.set_value();
    // Wait until parameter is set.
    int parameter = signalFuture.get();
    // …
};

auto function2 = [&thread2Started, signalFuture] {
    thread2Started.set_value();
    // Wait until parameter is set.
    int parameter = signalFuture.get();
    // …
};

// Run both lambda expressions asynchronously.
// Remember to capture the future returned by async()!
auto result1 = async(launch::async, function1);
auto result2 = async(launch::async, function2);

// Wait until both threads have started.
thread1Started.get_future().wait();
thread2Started.get_future().wait();

// Both threads are now waiting for the parameter.
// Set the parameter to wake up both of them.
signalPromise.set_value(42);