Lesson 18: Concurrency

Lesson 0018 — A Tour of C++, Chapter 18 (§18.1–§18.7)

Concurrency — the execution of several tasks simultaneously — is widely used to improve throughput (by using several processors for a single computation) or to improve responsiveness (by allowing one part of a program to progress while another is waiting for a response). The support provided by the C++ standard library is a portable and type-safe variant of what has been used in C++ for more than 20 years, aimed primarily at supporting systems-level concurrency; higher-level models can be supplied as libraries built using the standard-library facilities. The thread running through this chapter: think in terms of concurrent tasks, not threads and locks — and keep it simple.

18.1 Introduction

To allow concurrent execution of multiple threads in a single address space, C++ provides a suitable memory model and a set of atomic operations. The atomic operations allow lock-free programming. The memory model ensures that as long as a programmer avoids data races (uncontrolled concurrent access to mutable data), everything works as one would naively expect. Most users will see concurrency only in terms of the standard library and libraries built on top of that: threads, mutexes, lock() operations, packaged_tasks, and futures. But:

• Do not consider concurrency a panacea. If a task can be done sequentially, it is often simpler and faster to do so. Passing information from one thread to another can be surprisingly expensive.
• As an alternative to explicit concurrency features, we can often use a parallel algorithm to exploit multiple execution engines for better performance (§13.6, §17.3.1).
• C++ supports coroutines — functions that keep their state between calls (§18.6).

18.2 Tasks and Threads

We call a computation that can potentially be executed concurrently with other computations a task. A thread is the system-level representation of a task in a program. A task to be executed concurrently with other tasks is launched by constructing a thread (found in <thread>) with the task as its argument. A task is a function or a function object:

void f();                 // function
struct F {                // function object
    void operator()();    // F's call operator (§7.3.2)
};
void user() {
    thread t1 {f};        // f() executes in separate thread
    thread t2 {F{}};      // F{}() executes in separate thread
    t1.join();            // wait for t1
    t2.join();            // wait for t2
}

The join()s ensure that we don't exit user() until the threads have completed. To "join" a thread means to "wait for the thread to terminate." It is easy to forget to join(), and the results are usually bad — so the standard library provides jthread, a "joining thread" that follows RAII by having its destructor join():

void user() {
    jthread t1 {f};       // f() executes in separate thread
    jthread t2 {F{}};     // F{}() executes in separate thread
}                         // join automatically (reverse order: t2 before t1)

Threads of a program share a single address space — in this, threads differ from processes, which generally do not directly share data. Since threads share an address space, they can communicate through shared objects (§18.3). Such communication is typically controlled by locks or other mechanisms to prevent data races. Programming concurrent tasks can be very tricky; consider this bad error:

void f() { cout << "Hello "; }
struct F { void operator()() { cout << "Parallel World!\n"; } };

Here, f and F{} each use the object cout without any form of synchronization. The resulting output is unpredictable and can vary between executions — the order of the individual operations in the two tasks is not defined. The program may produce "odd" output, such as PaHerallllel o World!. Only a specific guarantee in the standard saves us from a data race within the definition of ostream that could lead to a crash. To avoid such problems with output streams, either have just one thread use a stream or use an osyncstream (§11.7.5).

When defining tasks of a concurrent program, our aim is to keep tasks completely separate except where they communicate in simple and obvious ways. The simplest way of thinking of a concurrent task is as a function that happens to run concurrently with its caller — we just have to pass arguments, get a result back, and make sure that there is no use of shared data in between (no data races).

18.2.1 Passing Arguments

We can easily pass data (or pointers or references to the data) as arguments:

void f(vector<double>& v);              // function: do something with v
struct F {                              // function object: do something with v
    vector<double>& v;
    F(vector<double>& vv) :v{vv} { }
    void operator()();
};
int main() {
    vector<double> some_vec {1, 2, 3, 4, 5, 6, 7, 8, 9};
    vector<double> vec2 {10, 11, 12, 13, 14};
    jthread t1 {f,ref(some_vec)};       // f(some_vec) executes in a separate thread
    jthread t2 {F{vec2}};               // F(vec2)() executes in a separate thread
}

F{vec2} saves a reference to the argument vector in F; F can now use that vector, and hopefully no other task accesses vec2 while F is executing (passing vec2 by value would eliminate that risk). The initialization {f,ref(some_vec)} uses the thread's variadic template constructor, which can accept an arbitrary sequence of arguments (§8.4). The ref() is a type function from <functional> that is needed to tell the variadic template to treat some_vec as a reference rather than as an object — without ref(), some_vec would be passed by value. The compiler checks that the first argument can be invoked given the following arguments and builds the necessary function object to pass to the thread.

18.2.2 Returning Results

Passing the arguments by non-const reference is a somewhat sneaky, but not uncommon, way of returning a result. A less obscure technique is to pass the input data by const reference and pass the location of a place to deposit the result as a separate argument:

void f(const vector<double>& v, double* res);   // take input from v; place result in *res
class F {
public:
    F(const vector<double>& vv, double* p) :v{vv}, res{p} { }
    void operator()();       // place result in *res
private:
    const vector<double>& v; // source of input
    double* res;             // target for output
};
double g(const vector<double>&);   // use return value

void user(vector<double>& vec1, vector<double> vec2, vector<double> vec3) {
    double res1, res2, res3;
    thread t1 {f,cref(vec1),&res1};        // f(vec1,&res1) in a separate thread
    thread t2 {F{vec2,&res2}};             // F{vec2,&res2}() in a separate thread
    thread t3 { [&](){ res3 = g(vec3); } };  // capture local variables by reference
    t1.join();  t2.join();  t3.join();      // join before using results
    cout << res1 << ' ' << res2 << ' ' << res3 << '\n';
}

Here, cref(vec1) passes a const reference to vec1. This works and is very common, but returning results through references is not particularly elegant — we return to this topic with future and promise in §18.5.1.

18.3 Sharing Data

Sometimes tasks need to share data. In that case, the access has to be synchronized so that at most one task at a time has access. (Experienced programmers will recognize this as a simplification — there is no problem with many tasks simultaneously reading immutable data.)

18.3.1 Mutexes and Locks

A mutex, a "mutual exclusion object," is a key element of general sharing of data between threads. A thread acquires a mutex using a lock() operation:

mutex m;                  // controlling mutex
int sh;                   // shared data
void f() {
    scoped_lock lck {m};  // acquire mutex
    sh += 7;              // manipulate shared data
}                         // release mutex implicitly

The type of lck is deduced to be scoped_lock<mutex> (§7.2.3). The scoped_lock's constructor acquires the mutex (through a call m.lock()); if another thread has already acquired the mutex, the thread waits ("blocks") until the other thread completes its access. Once a thread has completed its access, the scoped_lock releases the mutex (with a call m.unlock()), and threads waiting for it resume executing ("are woken up"). The mutual exclusion and locking facilities are found in <mutex>. Note the use of RAII (§6.3): resource handles such as scoped_lock and unique_lock (§18.4) are simpler and far safer than explicitly locking and unlocking mutexes.

The correspondence between shared data and a mutex relies on convention: the programmer has to know which mutex corresponds to which data. Obviously this is error-prone, and obviously we try to make the correspondence clear — e.g., class Record { mutex rm; /* ... */ };: for a Record called rec, you are supposed to acquire rec.rm before accessing the rest of rec.

It is not uncommon to need to simultaneously access several resources to perform some action. This can lead to deadlock: if thread1 acquires mutex1 and then tries to acquire mutex2 while thread2 acquires mutex2 and then tries to acquire mutex1, neither task will ever proceed further. The scoped_lock helps by enabling us to acquire several locks simultaneously:

void f() {
    scoped_lock lck {mutex1,mutex2,mutex3};   // acquire all three locks
    // ... manipulate shared data ...
}                                             // implicitly release all mutexes

This scoped_lock proceeds only after acquiring all its mutexes and will never block while holding a mutex. The destructor ensures the mutexes are released when the thread leaves the scope.

Communicating through shared data is pretty low level — the programmer has to devise ways of knowing what work has and has not been done by various tasks; in that regard, use of shared data is inferior to the notion of call and return. Some people are convinced that sharing must be more efficient than copying arguments and returns; that can indeed be so when large amounts of data are involved, but locking and unlocking are relatively expensive operations, and modern machines are very good at copying compact data such as vector elements. So don't choose shared data for communication because of "efficiency" without thought — and preferably not without measurement.

One of the most common ways of sharing data is among many readers and a single writer. This "reader-writer lock" idiom is supported by shared_mutex:

shared_mutex mx;                  // a mutex that can be shared
void reader() {
    shared_lock lck {mx};         // willing to share access with other readers
    // ... read ...
}
void writer() {
    unique_lock lck {mx};         // needs exclusive (unique) access
    // ... write ...
}

18.3.2 Atomics

A mutex is a fairly heavyweight mechanism involving the operating system; it allows arbitrary amounts of work to be done free of data races. But there is a far simpler and cheaper mechanism for doing just a tiny amount of work: an atomic variable. Here is a simple variant of the classic double-checked locking:

mutex mut;
atomic<bool> init_x;    // initially false
X x;                    // variable that requires nontrivial initialization
if (!init_x) {
    lock_guard lck {mut};
    if (!init_x) {
        // ... do nontrivial initialization of x ...
        init_x = true;
    }
}
// ... use x ...

The atomic saves us from most uses of the much more expensive mutex. Had init_x not been atomic, that initialization would have failed ever so infrequently, causing mysterious and hard-to-find errors because there would have been a data race on init_x. Here lock_guard was used rather than scoped_lock because only one mutex was needed, so the simplest lock sufficed.

18.4 Waiting for Events

Sometimes a thread needs to wait for some kind of external event, such as another thread completing a task or a certain amount of time having passed. The simplest "event" is simply time passing:

using namespace chrono;   // see §16.2.1
auto t0 = high_resolution_clock::now();
this_thread::sleep_for(milliseconds{20});
auto t1 = high_resolution_clock::now();
cout << duration_cast<nanoseconds>(t1-t0).count() << " nanoseconds passed\n";

We didn't even have to launch a thread — by default, this_thread can refer to the one and only thread. The basic support for communicating using external events is provided by condition_variables found in <condition_variable>. A condition_variable is a mechanism allowing one thread to wait for another — in particular, it allows a thread to wait for some condition (often called an event) to occur as the result of work done by other threads. Using condition_variables supports many forms of elegant and efficient sharing but can be rather tricky. Consider the classic example of two threads communicating by passing messages through a queue:

class Message { /* ... */ };        // object to be communicated
queue<Message> mqueue;               // the queue of messages
condition_variable mcond;            // the variable communicating events
mutex mmutex;                        // for synchronizing access to mcond

void consumer() {
    while(true) {
        unique_lock lck {mmutex};    // acquire mmutex
        mcond.wait(lck,[] { return !mqueue.empty(); });   // release mmutex and wait;
                                                          // re-acquire mmutex upon wakeup;
                                                          // don't wake up unless mqueue is non-empty
        auto m = mqueue.front();     // get the message
        mqueue.pop();
        lck.unlock();                // release mmutex
        // ... process m ...
    }
}

Waiting on a condition_variable releases its lock argument until the wait is over and then reacquires it. The explicit check of the condition, here !mqueue.empty(), protects against waking up just to find that some other task has "gotten there first" so the condition no longer holds. We used a unique_lock rather than a scoped_lock for two reasons:

• We need to pass the lock to the condition_variable's wait(). A scoped_lock cannot be moved, but a unique_lock can.
• We want to unlock the mutex protecting the condition variable before processing the message. A unique_lock offers operations such as lock() and unlock() for low-level control of synchronization.

On the other hand, unique_lock can handle only a single mutex. The corresponding producer:

void producer() {
    while(true) {
        Message m;
        // ... fill the message ...
        scoped_lock lck {mmutex};    // protect operations
        mqueue.push(m);
        mcond.notify_one();          // notify
    }                                // release mmutex (at end of scope)
}

18.5 Communicating Tasks

The standard library provides a few facilities to allow programmers to operate at the conceptual level of tasks (work to potentially be done concurrently) rather than directly at the lower level of threads and locks — all found in <future>:

• future and promise for returning a value from a task spawned on a separate thread;
• packaged_task to help launch tasks and connect up the mechanisms for returning a result;
• async() for launching a task in a manner very similar to calling a function.

18.5.1 future and promise

The important point about future and promise is that they enable a transfer of a value between two tasks without explicit use of a lock; "the system" implements the transfer efficiently. The basic idea is simple: when a task wants to pass a value to another, it puts the value into a promise. Somehow, the implementation makes that value appear in the corresponding future, from which it can be read (typically by the launcher of the task). If we have a future<X> called fx, we can get() a value of type X from it:

X v = fx.get();    // if necessary, wait for the value to get computed

If the value isn't there yet, our thread is blocked until it arrives. If the value couldn't be computed, get() might throw an exception (from the system or transmitted from the promise). The main purpose of a promise is to provide simple "put" operations — set_value() and set_exception() — to match future's get(). (The names "future" and "promise" are historical.)

void f(promise<X>& px) {        // a task: place the result in px
    try {
        X res;
        // ... compute a value for res ...
        px.set_value(res);
    }
    catch (...) {                // oops: couldn't compute res
        px.set_exception(current_exception());
    }
}

void g(future<X>& fx) {          // a task: get the result from fx
    // ...
    X v = fx.get();              // if necessary, wait for the value to get computed
    // ... use v ...
}

To deal with an exception transmitted through a future, the caller of get() must be prepared to catch it somewhere. If the error doesn't need to be handled by g() itself, the code reduces to the minimal form above — an exception thrown from f()'s function is then implicitly propagated to g()'s caller, exactly as it would have been had g() called f() directly.

18.5.2 packaged_task

How do we get a future into the task that needs a result and the corresponding promise into the thread that should produce that result? The packaged_task type simplifies setting up tasks connected with futures and promises to be run on threads: it provides wrapper code to put the return value or exception from the task into a promise (like the code shown in §18.5.1). If you ask it by calling get_future(), a packaged_task will give you the future corresponding to its promise. For example, we can set up two tasks to each add half of the elements of a vector<double> using the standard-library accumulate() (§17.3):

double accum(double* beg, double* end, double init) {   // compute the sum of [beg:end) starting with init
    return accumulate(beg,end,init);
}
double comp2(vector<double>& v) {
    packaged_task pt0 {accum};              // package the task (i.e., accum)
    packaged_task pt1 {accum};
    future<double> f0 {pt0.get_future()};   // get hold of pt0's future
    future<double> f1 {pt1.get_future()};   // get hold of pt1's future
    double* first = &v[0];
    thread t1 {move(pt0),first,first+v.size()/2,0};        // start a thread for pt0
    thread t2 {move(pt1),first+v.size()/2,first+v.size(),0}; // start a thread for pt1
    // ...
    return f0.get()+f1.get();               // get the results
}

The packaged_task template takes the type of the task as its template argument and the task as its constructor argument. The move() operations are needed because a packaged_task cannot be copied: it is a resource handle — it owns its promise and is (indirectly) responsible for whatever resources its task may own. Please note the absence of explicit mention of locks in this code: we are able to concentrate on tasks to be done, rather than on the mechanisms used to manage their communication. The two tasks will be run on separate threads and thus potentially in parallel.

18.5.3 async()

The line of thinking pursued in this chapter is the simplest yet still among the most powerful: treat a task as a function that may happen to run concurrently with other tasks. To launch tasks to potentially run asynchronously, we can use async():

double comp4(vector<double>& v) {           // spawn many tasks if v is large enough
    if (v.size()<10'000)                     // is it worth using concurrency?
        return accum(v.begin(),v.end(),0.0);
    auto v0 = &v[0];
    auto sz = v.size();
    auto f0 = async(accum,v0,v0+sz/4,0.0);   // first quarter
    auto f1 = async(accum,v0+sz/4,v0+sz/2,0.0);   // second quarter
    auto f2 = async(accum,v0+sz/2,v0+sz*3/4,0.0); // third quarter
    auto f3 = async(accum,v0+sz*3/4,v0+sz,0.0);   // fourth quarter
    return f0.get()+f1.get()+f2.get()+f3.get();   // collect and combine the results
}

Basically, async() separates the "call part" of a function call from the "get the result part," and separates both from the actual execution of the task. Using async(), you don't have to think about threads and locks — you think in terms of tasks that potentially compute their results asynchronously. There is an obvious limitation: don't even think of using async() for tasks that share resources needing locking. With async() you don't even know how many threads will be used — that's up to async() to decide based on the system resources available at the time of a call (it may check whether any idle cores are available). A guess about the cost of computation relative to the cost of launching a thread, such as v.size()<10'000, is very primitive and prone to gross mistakes about performance — don't take such an estimate as more than a simple and probably poor guess. It is rarely necessary to manually parallelize a standard-library algorithm such as accumulate(), because the parallel algorithms (e.g., reduce(par_unseq, ...)) usually do a better job (§17.3.1) — however, the technique is general. Note that async() is not just a mechanism for parallel computation: it can also be used to spawn a task for getting information from a user, leaving the "main program" active with something else.

18.5.4 Stopping a thread

Sometimes we want to stop a thread because we no longer are interested in its result. Just "killing" it is usually not acceptable because a thread can own resources that must be released (e.g., locks, sub-threads, and database connections). Instead, the standard library provides a mechanism for politely requesting a thread to clean up and go away: a stop_token. A thread can be programmed to terminate if it has a stop_token and is requested to stop. Consider a parallel find_any() that spawns many threads looking for a result; when a thread returns with an answer, we would like to stop the remaining threads. The find() task has a main loop in which we can insert a test of whether to continue or stop:

atomic<int> result = -1;         // put a resulting index here
template<class T> struct Range { T* first; T* last; };   // a way of passing a range of Ts

void find(stop_token tok, const string* base, const Range<string> r, const string target) {
    for (string* p = r.first; p!=r.last && !tok.stop_requested(); ++p)
        if (match(*p,target)) {   // match() applies some matching criteria to the two strings
            result = p - base;    // the index of the found element
            return;
        }
}

Here, !tok.stop_requested() tests whether some other thread has requested this thread to terminate. A stop_token is the mechanism for safely (no data races) communicating such a request. The stop_sources produce the stop_tokens through which requests to stop are communicated to threads:

void find_all(vector<string>& vs, const string& key) {
    int mid = vs.size()/2;
    string* pvs = &vs[0];
    stop_source ss1{};
    jthread t1(find, ss1.get_token(), pvs, Range{pvs,pvs+mid}, key);       // first half of vs
    stop_source ss2{};
    jthread t2(find, ss2.get_token(), pvs, Range{pvs+mid,pvs+vs.size()}, key); // second half of vs
    while (result == -1) this_thread::sleep_for(10ms);
    ss1.request_stop();    // we have a result: stop all threads
    ss2.request_stop();
    // ... use result ...
}

The synchronization and returning of a result here is the simplest possible: put the result in an atomic variable (§18.3.2) and do a spin loop on that. Of course, we could elaborate this simple example — many searcher threads, more general result return, different element types — but that would obscure the basic role of stop_source and stop_token.

18.6 Coroutines

A coroutine is a function that maintains its state between calls. In that, it's a bit like a function object, but the saving and restoring of its state between calls are implicit and complete. Consider a classic example:

generator<long long> fib() {      // generate Fibonacci numbers
    long long a = 0;
    long long b = 1;
    while (a<b) {
        auto next = a+b;
        co_yield next;            // save state, return value, and wait
        a = b;
        b = next;
    }
    co_return 0;                  // a fib too far
}
void user(int max) {
    for (int i=0; i++<max;) cout << fib() << ' ';
}   // 1 2 3 5 8 13 ...

The generator return value is where the coroutine stores its state between calls. We could, of course, have made a function object Fib that worked the same way — but then we would have had to maintain its state ourselves. For larger states and more complex computations, saving and restoring state get tedious, hard to optimize, and error-prone. In effect, a coroutine is a function that saves its stack frame between calls. The co_yield returns a value and waits for the next call; the co_return returns a value and terminates the coroutine. Coroutines can be synchronous (the caller waits for the result — the Fibonacci example is obviously synchronous) or asynchronous (the caller does some other work until it looks for the result). The coroutines are implemented as an extremely flexible framework designed by and for experts; the library facilities to make simple uses simple are still missing in C++20 — for example, generator is not (yet) part of the standard library (there are proposals, and a Web search will find good implementations, e.g., the Cppcoro library).

18.6.1 Cooperative Multitasking

Donald Knuth praises the usefulness of coroutines, but also bemoans that it is hard to give brief examples because coroutines are most useful in simplifying complex systems. Coroutines are ideal for event-driven simulations — a system represented as a network of simple tasks (coroutines) that collaborate to complete complex tasks. The keys to such designs:

• Many different coroutines that maintain their state between calls.
• A form of polymorphism that allows us to keep lists of events containing different kinds of coroutines and invoke them independently of their types.
• A scheduler that selects the next coroutine(s) to run from the list(s).

It is essential for such systems not to use too much space — that's why we don't use processes or threads: a thread requires a megabyte or two (mostly for its stack), a coroutine often only a couple of dozen bytes. If you need many thousands of tasks, that can make a big difference. Context switching between coroutines is also far faster than between threads or processes.

struct Event_base {
    virtual void operator()() = 0;
    virtual ~Event_base() {}
};
template<class Act> struct Event : Event_base {
    Event(const string n, Act a) : name{n}, act{move(a)} {}
    string name;
    Act act;
    void operator()() override { act(); }
};

void test() {
    vector<Event_base*> events = {       // a couple of Events holding coroutines
        new Event{"integers", sequencer(10)},
        new Event{"chars", char_seq('a')}
    };
    vector order {0, 1, 1, 0, 1, 0, 1, 0, 0};   // choose some order
    for (int x : order)                 // invoke coroutines in order
        (*events[x])();
    for (auto p : events) delete p;     // clean up
}

So far, there is nothing specifically coroutine about this — it's just a conventional object-oriented framework for executing operations on a set of objects of potentially differing types. However, sequencer and char_seq happen to be coroutines: the fact that they maintain their state between calls is essential for real-world uses of such frameworks.

task sequencer(int start, int step =1) {
    auto value = start;
    while (true) {
        cout << "value: " << value << '\n';   // communicate a result
        co_yield 0;                            // sleep until someone resumes this coroutine
        value += step;                         // update state
    }
}

We can see that sequencer is a coroutine because it uses co_yield to suspend itself between calls; this implies that task must be a coroutine handle. The "magic" is in the return type task: it holds the state of the coroutine (in effect the function's stack frame) between calls and determines the meaning of co_yield. From a user's point of view task is trivial — it simply provides an operator to invoke the coroutine. If task had been in a library, preferably the standard library, that would be all we needed to know; the implementation is a promise_type mapping to the language features (initial_suspend, final_suspend, yield_value, get_return_object, coroutine_handle<promise_type>). Stroustrup strongly encourages not writing such code yourself unless you are a library implementer trying to save others from the bother.

18.7 Advice

Here is a summary of the guidance from this chapter. All 36 items, with the section where each is introduced. The C++ Core Guidelines link each item to its recommended practice.

Retrieval Quiz

Notes

Primary source: Stroustrup, B. (2022). A Tour of C++, 3rd ed., Chapter 18: "Concurrency." Addison-Wesley.
Reference: Chapter 1–12 Quick Reference & Glossary — keep it beside you while you study.
Recommended supplement: cppreference on thread support library, future/promise, and coroutines.

Questions? Ask your agent — your teacher — about anything unclear: when a data race actually crashes a program, why scoped_lock can't be passed to wait(), or how async() decides about threads. Follow-ups are expected, not optional.

Chapters

1. Lesson 1: The Basics (Ch. 1)
2. Lesson 2: User-Defined Types (Ch. 2)
3. Lesson 3: Separate Compilation (Ch. 3)
4. Lesson 4: Error Handling (Ch. 4)
5. Lesson 5: Classes (Ch. 5)
6. Lesson 6: Essential Operations (Ch. 6)
7. Lesson 7: Templates (Ch. 7)
8. Lesson 8: Concepts and Generic Programming (Ch. 8)
9. Lesson 9: Library Overview (Ch. 9)
10. Lesson 10: Strings and Regular Expressions (Ch. 10)
11. Lesson 11: Input and Output (Ch. 11)
12. Lesson 12: Containers (Ch. 12)
13. Lesson 13: Algorithms (Ch. 13)
14. Lesson 14: Ranges (Ch. 14)
15. Lesson 15: Pointers and Containers (Ch. 15)
16. Lesson 16: Utilities (Ch. 16)
17. Lesson 17: Numerics (Ch. 17)
18. Lesson 18: Concurrency (Ch. 18 — this lesson)
19. Lesson 19: History and Compatibility (Ch. 19)