9

Concurrency and Parallelism

Concurrency and parallelism refer to the ability to run code in separate threads of execution .

More specifically, concurrency is the ability to run threads in the background, and parallelism is the ability to run threads simultaneously in separate cores of a processor. The run-time library, along with the host operating system, will choose between concurrent and parallel execution models for a given thread on a given hardware environment.

In a modern multi-tasking operating system, the main() function already represents a thread of execution. When a new thread is started, it's said to be spawned by an existing thread. A group of threads may be called a swarm .

In the C++ standard library, the std::thread class provides the basic unit of threaded execution. Other classes build upon thread to provide locks , mutexes , and other concurrency patterns. Depending on system architecture, execution threads may run concurrently on one processor or in parallel on separate cores.

In this chapter, we will cover these tools and more, in the following recipes:

• Sleep a process for a specific amount of time
• Use std::thread for concurrency
• Use std::async for concurrency
• Run STL algorithms in parallel with execution policies
• Use mutex and lock to safely share data
• Share flags and values with std::atomic
• Initialize threads with std::call_once
• Resolve the producer-consumer problem with std::condition_variable
• Implement multiple producers and consumers
• Suspend and resume execution with coroutines
• Use coroutines for a producer-consumer solution

Sleep a process for a specific amount of time

The <thread> header provides three functions for putting a thread to sleep: sleep() , sleep_for() , and sleep_until() . These functions are in the std::this_thread namespace.

This recipe explores the use of these functions, as we will be using them later in this chapter.

How to do it

Let's look at how to use the sleep_for() and sleep_until() functions.

• The sleep-related functions are in the std::this_thread namespace. Because it has just a few symbols, we'll go ahead and issue using directives for std::this_thread and std::chrono_literals :

  using namespace std::this_thread; using namespace std::chrono_literals;

  The chrono_literals namespace has symbols for representing durations, like 1s for one second, or 100ms for 100 milliseconds.

• In main() , we'll mark a point in time with steady_clock::now() so we can time our test:

  int main() {     auto t1 = steady_clock::now();     println("sleep for 1.3 seconds");     sleep_for(1s + 300ms);      println("sleep for 2 seconds");     sleep_until(steady_clock::now() + 2s);      duration<double> dur1 = steady_clock::now() - t1;     println("total duration: {:.5}s", dur1.count()); }

  The sleep_for() function takes a duration object to specify the amount of time to sleep. The argument (1s + 300ms) uses chrono_literal operators to return a duration object representing 1.3 seconds.

  The sleep_until() function takes a time_point object to specify a specific time to resume from sleep. In this case, the chrono_literal operators are used to modify the time_point object returned from steady_clock::now() .

  This is our output:

  sleep for 1.3 seconds sleep for 2 seconds total duration: 3.3004s

  Your results will vary based on many factors, including system hardware, OS, and scheduling of other resources.

How it works

The sleep_for(duration) and sleep_until(time_point) functions suspend execution of the current thread for the specified duration or until the time_point is reached, respectively.

The sleep_for() function will use the steady_clock implementation, if supported. Otherwise, the duration may be subject to time adjustments. Both functions may block for longer due to scheduling or resource delays.

There's more…

Some systems support a POSIX function sleep() , which suspends execution for the number of seconds specified:

unsigned int sleep(unsigned int seconds);

The sleep() function is part of the POSIX C standard and is not part of the C++ standard. If available, POSIX sleep() can be found in the unistd.h header.

Use std::thread for concurrency

A thread is a unit of concurrent execution. The main() function may be thought of as the main thread of execution . Within the context of the operating system, the main thread runs concurrently with other threads owned by other processes.

The std::thread class is the primary interface for creating threads in the C++ Standard Library. In this recipe we will examine the basics of std::thread , and how join() and detach() determine its execution context.

How to do it

In this recipe we will create some threads using std::thread , and experiment with their execution options.

• We'll start with a convenience function for sleeping a thread, in milliseconds:

  void sleepms(const unsigned ms) {     using std::chrono::milliseconds;     std::this_thread::sleep_for(milliseconds(ms)); }

  The sleep_for() function takes a duration object and blocks execution of the current thread for the specified duration. This sleepms() function serves as a convenience wrapper that takes an unsigned value for the number of milliseconds to sleep.

• Now, we need a function for our thread. This function sleeps for a variable number of milliseconds, based on an integer parameter:

  void fthread(const int n) {     println("This is t{}", n);      for (size_t i {}; i < 5; ++i) {         sleepms(100 * n);         println("t{}: {}", n, i + 1);     }     println("Finishing t{}", n); }

  fthread() calls sleepms() five times, sleeping each time for 100 * n milliseconds.

• We can run this in a separate thread with std::thread from main() :

  int main() {     thread t1(fthread, 1);     println("end of main()"); }

  This code compiles but we get this error when we run it:

  terminate called without an active exception Aborted

  (Your error message will vary. This is the error message on Debian with GCC.)

  The problem is that the operating system doesn't know what to do with the thread object when it goes out of scope. We must specify if the caller waits for the thread, or if it's detached and runs independently.

• We use the join() method to indicate that the caller will wait for the thread to finish:

  int main() {     thread t1(fthread, 1);     t1.join();     println("end of main()"); }

  Output:

  This is t1 t1: 1 t1: 2 t1: 3 t1: 4 t1: 5 Finishing t1 end of main()

  Now main() waits for the thread to finish.

• If we call detach() instead of join() , then main() doesn't wait and the program ends before the thread can run:

  thread t1(fthread, 1); t1.detach();

  Output:

  end of main()

  When the thread is detached, we need to give it time to run:

  thread t1(fthread, 1); t1.detach(); println("main() sleep 2 sec"); sleepms(2000);

  Output:

  main() sleep 2 sec This is t1 t1: 1 t1: 2 t1: 3 t1: 4 t1: 5 Finishing t1 end of main()

• Let's start and detach a second thread and see what happens:

  int main() {     thread t1(fthread, 1);     thread t2(fthread, 2);     t1.detach();     t2.detach();     println("main() sleep 2 sec");     sleepms(2000);     println("end of main()"); }

  Output:

  main() sleep 2 sec This is t1 This is t2 t1: 1 t2: 1 t1: 2 t1: 3 t2: 2 t1: 4 t1: 5 Finishing t1 t2: 3 t2: 4 t2: 5 Finishing t2 end of main()

  Because our fthread() function uses its parameter as a multiplier for sleepms() , the second thread runs a bit slower than the first. We can see the timers interlaced in the output.

• If we do this with join() instead of detatch() we get a similar result:

  int main() {     thread t1(fthread, 1);     thread t2(fthread, 2);     t1.join();     t2.join();     println("end of main()"); }

  Output:

  This is t1 This is t2 t1: 1 t2: 1 t1: 2 t1: 3 t2: 2 t1: 4 t1: 5 Finishing t1 t2: 3 t2: 4 t2: 5 Finishing t2 end of main()

  Because join() waits for the thread to finish, we no longer need the 2-second sleepms() in main() to wait for the threads to finish.

How it works

A std::thread object represents a thread of execution. There is a 1:1 relationship between the object and thread. One thread object represents one thread, and one thread is represented by one thread object. A thread object cannot be copied or assigned but it can be moved.

The thread constructor looks like this:

explicit thread(Function&&f,Args&&...args);

A thread is constructed with a function pointer and zero or more arguments. The function is called immediately with the arguments provided:

thread t1(fthread, 1);

This creates the object t1 and immediately calls the function fthread(int) with the literal value 1 as the argument.

After creating the thread , we must use either join() or detach() on the thread:

t1.join();

The join() method blocks execution of the calling thread until the t1 thread has completed.

t1.detach();

The detach() method allows the calling thread to continue independently of the t1 thread.

There's more…

C++20 provides std::jthread , which automatically joins the caller at the end of its scope.

int main() {     std::jthread t1(fthread, 1);     println("end of main()"); }

Output:

end of main() This is t1 t1: 1 t1: 2 t1: 3 t1: 4 t1: 5 Finishing t1

jthread is a thread type that automatically joins on destruction. This allows both the t1 and main() threads to execute independently without an explicit join() and without accidentally terminating. This allows a safe, structured shutdown and simpler thread management.

Of course, this works just as safely with multiple threads:

int main() {     std::jthread t1(fthread, 1);     std::jthread t2(fthread, 2);     println("end of main()"); }

Output:

This is t1 This is t2 end of main() t1: 1 t2: 1 t1: 2 t1: 3 t2: 2 t1: 4 t1: 5 Finishing t1 t2: 3 t2: 4 t2: 5 Finishing t2

Notice that the threads in this example run after the "end of main()" message. This is because the join occurs at the end of the scope of the jthread() call.

While the combined jthread() can be more convenient than the separate thread and join() , we lose the ability to specify when the join occurs. If you wish to control when the thread is joined, you'll want to use thread() and join() explicitly instead of jthread() .

Use std::async for concurrency

std::async() runs a target function asynchronously and returns a std::future object to carry the target's return value. In this way, async() operates much like std::thread but allows return values.

Let's consider the use of std::async() with a few examples.

How to do it

In its simplest forms, the std::async() function performs much the same task as std::thread , without the need to call join() or detach() , and also allows return values via a std::future object.

In this recipe we'll use a function that counts the number of primes in a range. We'll use chrono::steady_clock to time the execution of each thread.

• We start with a couple of convenient aliases:

  using launch = std::launch; using secs = std::chrono::duration<double>;

  std::launch has launch policy constants for use with the async() call. The secs alias is a duration class for timing our primes calculations.

• Our target function counts prime numbers in a range. This is essentially a way to understand the execution policies by eating some clock cycles.

  struct prime_time {     secs dur{};     uint64_t count{}; };  prime_time count_primes(const uint64_t max) {     prime_time ret {};     constexpr auto isprime = [](uint64_t n) {         if (n == 2) return true;         if (n < 2 || n % 2 == 0) return false;         for (uint64_t i{3}; i <= n / i; i += 2) {             if (n % i == 0) return false;         }         return true;     };     ret.count = max >= 2 ? 1 : 0;     uint64_t start {3};     uint64_t end {max};     auto t1 = steady_clock::now();     for (uint64_t i {start}; i <= end ; ++i) {         if (isprime(i)) ++ret.count;     }     ret.dur = steady_clock::now() - t1;     return ret; }

  The prime_time structure is for the return value, with elements for duration and count . This allows us to time the loop itself. The isprime lambda returns true if a value is prime. We use steady_clock to calculate the duration of the loop that counts primes.

• In main() , we call our function and report its timing:

  int main() {     constexpr uint64_t max_prime {0x4FFFFF};     auto pt = count_primes(max_prime);     println("primes: {} {}", pt.count, pt.dur); }

  Output (on an Intel NUC, 1.61 GHz 6 core i7):

  primes: 364195 5.30279

• Now, we can run count_primes() asynchronously with std::async() :

  int main() {     constexpr uint64_t max_prime {0x4FFFFF};     auto primes1 = async(count_primes, max_prime);     auto pt = primes1.get();     println("primes: {} {}", pt.count, pt.dur); }

  Here we call async() with our count_primes function and the max_prime parameter. This runs count_primes() in the background.

  async() returns a std::future object, which carries the return value of an asynchronous operation. The future object get() method blocks execution until the asynchronous function has completed, then returns the prime_time object from the function.

  This runs with roughly the same timing as we got without async() :

  primes: 364195 5.40484s

• The async() function optionally takes execution policy flags as its first parameter:

  auto primes1 = async(launch::async, count_primes, max_prime);

  The choices are async or deferred . These flags are in the std::launch namespace.

  The async flag enables asynchronous operation, and the deferred flag enables lazy evaluation. These flags are bitmapped and may be combined with the bitwise or | operator.

  The default is for both bits to be set, as if async | deferred was specified.

• We can run several instances of our function simultaneously with async() :

  int main() {     constexpr uint64_t max_prime {0x4FFFFF};     constexpr size_t num_threads {15};     list<std::future<prime_time>> swarm;      println("start parallel primes");     auto t1 {steady_clock::now()};     for (size_t i {}; i < num_threads; ++i) {         swarm.emplace_back(           async(launch::async, count_primes, max_prime)         );     }      for (auto& f : swarm) {         static size_t i{};         auto pt = f.get();         println("primes({:02}): {} {:.5}",           ++i, pt.count, pt.dur);     }      secs dur_total{ steady_clock::now() - t1 };     println("total duration: {:.5}s", dur_total.count()); }

  We know that async returns a future object. So, we can run 15 threads by storing the future objects in a container. Here's the output on a 6-core i7 running Windows:

  start parallel primes primes(01): 364195 4.56607s primes(02): 364195 4.56273s primes(03): 364195 4.55887s primes(04): 364195 4.55954s primes(05): 364195 5.37375s primes(06): 364195 5.43635s primes(07): 364195 5.34973s primes(08): 364195 4.56051s primes(09): 364195 4.56134s primes(10): 364195 4.5587s primes(11): 364195 5.40484s primes(12): 364195 4.58699s primes(13): 364195 5.38693s primes(14): 364195 3.82778s primes(15): 364195 3.82953s total duration: 8.3893s

  Even though the 6-core i7 cannot run all the processes in separate cores, it still completes 15 instances in about 8.4 seconds.

  It looks like it finishes the first 13 threads in about 5.4 seconds and then takes another 3 seconds to finish the last 2 threads. This likely takes advantage of the processor's hyper-threading technology that allows 2 threads to run in one core under some circumstances.

  When we run the same code on a 32-core Apple M1 Ultra under macOS we get this result:

  start parallel primes primes(01): 364195 0.476 primes(02): 364195 0.477 primes(03): 364195 0.477 primes(04): 364195 0.476 primes(05): 364195 0.477 primes(06): 364195 0.485 primes(07): 364195 0.480 primes(08): 364195 0.476 primes(09): 364195 0.496 primes(10): 364195 0.483 primes(11): 364195 0.480 primes(12): 364195 0.480 primes(13): 364195 0.476 primes(14): 364195 0.476 primes(15): 364195 0.476 total duration: 0.49679s

  The 32-core M1 Ultra easily runs all 15 parallel processes simultaneously and gets through the set in under 500 milliseconds.

How it works

The key to understanding std::async is in its use of std::promise and std::future .

The promise class allows a thread to store an object that may later be retrieved asynchronously by a future object.

For example, if we have a function like this:

void f() {     println("this is f()"); }

We could run it with std::thread like this:

int main() {     std::thread t1(f);     t1.join();     println("end of main()"); }

That works fine for a simple function with no return value. When we want to return a value from f() , we can use promise and future .

We pass the promise object to our function:

void f(std::promise<int> value) {     println("this is f()");     value.set_value(47); }

And then use promise and future objects in the main() thread to pass and return values from f() :

int main() {     std::promise<int> value_promise;     std::future<int> value_future = value_promise.get_future();     std::thread t1(f, std::move(value_promise));     t1.detach();     println("value is {}", value_future.get());     println("end of main()"); }

Notice that a promise object cannot be copied, so we need to use std::move to pass it to the function.

The promise object serves as a bridge to a future object, which allows us to retrieve the value when it becomes available.

Simplify promise and future with std::async()

std::async() is a helper function to simplify the creation of the promise and future objects. With async() , we can accomplish the same task like this:

int f(int value) {     println("this is f()");     return value; }  int main() {     auto value_future = async(launch::async, f, 47);     println("value is {}", value_future.get());     println("end of main()"); }

Result:

this is f() value is 47 end of main()

That's the value of async() . For many purposes, it makes the use of promise and future trivially simple.

Run STL algorithms in parallel with execution policies

Beginning with C++17, many of the standard STL algorithms can run with parallel execution . This feature allows an algorithm to split its work into sub-tasks to run simultaneously on multiple CPU cores. These algorithms accept an execution policy object that specifies the kind of parallelism applied to the algorithm. This feature requires hardware support.

How to do it

Execution policies are defined in the <execution> header and in the std::execution namespace. In this recipe we will test the available policies using the std::transform() algorithm.

• For timing purposes, we'll use the duration object with the std::milli ratio, so we can measure in milliseconds:

  using dur_t = duration<double, std::milli>;

  For demonstration purposes, we'll start with a vector of int with 20 million random values:

  int main() {     std::vector<unsigned> v(20'000'000);     std::random_device rng;     println("generate randoms");     for (auto &i : v) i = rng() % 0xFFFF'FFFF;     ...

  Now we apply a simple transformation:

  auto mul2 = [](int n) {return n * 2;}; auto t0 = steady_clock::now(); std::transform(v.begin(), v.end(), v.begin(), mul2); dur_t dur0 = steady_clock::now() - t0; println("no policy: {:.3}ms", dur0.count());

  The mul2 lambda simply multiplies a value by 2 . The transform() algorithm applies mul2 to every member of the vector .

  This transformation does not specify an execution policy.

  Output:

  no policy: 8.74ms

• We can specify an execution policy in the first argument of the algorithm:

  std::transform(execution::seq,     v.begin(), v.end(), v.begin(), mul2);

  The seq policy means that the algorithm shall not be parallelized. This is the same as no execution policy.

  Output:

  execution::seq: 10.9ms

  Notice that the duration is somewhat slower than without a policy. This is because the seq policy may inhibit inlining and other compiler optimizations.

• The execution::par policy allows the algorithm to parallelize its workload:

  std::transform(execution::par,     v.begin(), v.end(), v.begin(), mul2); println("execution::par: {:.3}ms", dur2.count());

  Output:

  execution::par: 6.2ms

  The algorithm runs somewhat faster with the parallel execution policy, which allows the implementation to share its work among different processor cores.

  The execution::par_unseq policy allows unsequenced parallel execution of the workload:

  std::transform(execution::par_unseq,     v.begin(), v.end(), v.begin(), mul2); println("execution::par_unseq: {:.3}ms", dur3.count());

  Output:

  execution::par_unseq: 5.99ms

  Here, we notice another increase in performance with this policy.

  The execution::par_unseq policy allows the implementation greater flexibility in assigning work to the different cores. However, it also imposes stricter requirements of the algorithm. The algorithm must not perform operations that rely on ordering or synchronized behavior.

How it works

The execution policies interface doesn't specify how the algorithm workloads are parallelized. It's designed to work with a diverse set of hardware and processors under varying loads and circumstances. It may be implemented entirely in the library or may rely on compiler or hardware support.

Parallelization will show the most improvement on algorithms that do more than O(n) work. For example, sort() shows a dramatic improvement. Here's a sort() with no parallelization:

auto t0 = steady_clock::now(); std::sort(v.begin(), v.end()); dur_t dur0 = steady_clock::now() - t0; println("sort: {:.3}ms ", dur0.count());

Output:

sort: 835ms

With execution::par we see significant performance gains:

std::sort(execution::par, v.begin(), v.end());

Output:

sort: 158ms

The improvement with execution::par_unseq is incrementally better:

std::sort(execution::par_unseq, v.begin(), v.end());

Output:

sort: 152ms

It's a good idea to do a lot of testing when using the parallelized algorithms. If your algorithm or predicates do not lend themselves well to parallelization, you may end up with minimal performance gains or unintended side-effects.

Use mutex and lock to safely share data

The term mutex refers to mutually exclusive access to shared resources. A mutex is commonly used to avoid data corruption and race conditions, due to multiple threads of execution attempting to access the same data. A mutex will typically use locks to restrict access to one thread at a time.

The STL provides mutex and lock classes in the <mutex> header.

How to do it

In this recipe we will use a simple Animal class to experiment with locking and unlocking a mutex .

• We start with creating a mutex object

  std::mutex animal_mutex {};

  The mutex is declared in global scope so it's accessible to all the relevant objects.

• Our Animal class has a name and a list of friends:

  class Animal {     using friend_t = list<Animal>;     string_view s_name {"unk"};     friend_t l_friends {}; public:     Animal() = delete;     Animal(const string_view n) : s_name {n} {}     ... }

  Adding and deleting friends will be a useful test case for our mutex .

• The equality operator is the only operator we'll need:

  bool operator==(const Animal& o) const {     return s_name.data() == o.s_name.data(); }

  The s_name member is a string_view , so we can test the address of its data store for equality. This is a simple and useful shortcut.

• The is_friend() method tests if another Animal is in the l_friends list:

  bool is_friend(const Animal& o) const {     for (const auto& a : l_friends) {         if (a == o) return true;     }     return false; }

• The find_friend() method returns an optional, with an iterator to the Animal if found:

  optional<friend_t::iterator> find_friend(const Animal& o) noexcept {     for (auto it {l_friends.begin()};       it != l_friends.end(); ++it) {         if (*it == o) return it;     }     return {}; }

• The display() method prints s_name along with the names of each of the Animal objects in the l_friends list:

  void display() const noexcept {     auto n_animals {l_friends.size()};     print("Animal: {}, friends: ", s_name);     if (!n_animals) print("none");     else {         for (auto n : l_friends) {             print("{}", n.s_name);             if (--n_animals) print(", ");         }     }     print("\n"); }

• The add_friend() method adds an Animal object to the l_friends list:

  bool add_friend(Animal& o) noexcept {     println("add_friend {} -> {}", s_name, o.s_name);     if (*this == o) return false;     if (!is_friend(o)) {         l_friends.emplace_back(o);     }     if (!o.is_friend(*this)) {         o.l_friends.emplace_back(*this);     }      return true; }

• The delete_friend() method removes an Animal object from the l_friends list:

  bool delete_friend(Animal& o) noexcept {     println("delete_friend {} -> {}", s_name, o.s_name);     if (*this == o) return false;     if (auto it = find_friend(o)) {         l_friends.erase(it.value());     }     if (auto it = o.find_friend(*this)) {         o.l_friends.erase(it.value());     }      return true; }

• In the main() function we create some Animal objects:

  int main() {     auto cat1 = std::make_unique<Animal>("Felix");     auto tiger1 = std::make_unique<Animal>("Hobbes");     auto dog1 = std::make_unique<Animal>("Astro");     auto rabbit1 = std::make_unique<Animal>("Bugs");     ...

• We call add_friends() on our objects with async() , to run them in separate threads:

  auto a1 = std::async([&]{cat1->add_friend(*tiger1); }); auto a2 = std::async([&]{cat1->add_friend(*rabbit1); }); auto a3 = std::async([&]{rabbit1->add_friend(*dog1); }); auto a4 = std::async([&]{rabbit1->add_friend(*cat1); }); a1.wait(); a2.wait(); a3.wait(); a4.wait();

  We call wait() to allow our threads to complete before continuing.

• We create a lambda with async() to call display() with wait() to see our Animal objects and their relationships:

  auto p1 = std::async([&]{cat1->display(); }); auto p2 = std::async([&]{tiger1->display(); }); auto p3 = std::async([&]{dog1->display(); }); auto p4 = std::async([&]{rabbit1->display(); }); p1.wait(); p2.wait(); p3.wait(); p4.wait();

• And finally, we call delete_friend() to remove one of our relationships:

  auto a5 = std::async([&]{     cat1->delete_friend(*rabbit1); }); a5.wait(); auto p5 = std::async([&]{cat1->display(); }); auto p6 = std::async([&]{rabbit1->display(); });

• At this point, our output looks like this:

  add_friend Bugs -> Felix add_friend Felix -> Hobbes add_friend Felix -> Bugs add_friend Bugs -> Astro Animal: Felix, friends: Bugs, Hobbes Animal: Hobbes, friends: Animal: Bugs, friends: FelixAnimal: Astro, friends: Felix , Astro Bugs delete_friend Felix -> Bugs Animal: Felix, friends: Hobbes Animal: Bugs, friends: Astro

  This output is somewhat scrambled. It will be different each time you run it. It may be fine sometimes, but don't let that fool you. We need to add some mutex locks to control access to the data.

• One way to use mutex is with its lock() and unlock() methods. Let's add them to the add_friend() function:

  bool add_friend(Animal& o) noexcept {     println("add_friend {} -> {}", s_name, o.s_name);     if (*this == o) return false;     animal_mutex.lock();     if (!is_friend(o)) {         l_friends.emplace_back(o);     }     if (!o.is_friend(*this)) {         o.l_friends.emplace_back(*this);     }     animal_mutex.unlock();     return true; }

  The lock() method attempts to acquire a lock on the mutex . If the mutex is already locked, it will wait ( block execution ) until the mutex is unlocked.

• We also need to add a lock to delete_friend() :

  bool delete_friend(Animal& o) noexcept {     println("delete_friend {} -> {}", s_name, o.s_name);     if (*this == o) return false;     animal_mutex.lock();     if (auto it = find_friend(o)) {         l_friends.erase(it.value());     }     if (auto it = o.find_friend(*this)) {         o.l_friends.erase(it.value());     }     animal_mutex.unlock();     return true; }

• And we need to add a lock to display() so that data is not changed while printing the output:

  void display() const noexcept {     animal_mutex.lock();     auto n_animals {l_friends.size()};     print("Animal: {}, friends: ", s_name);     if (!n_animals) print("none");     else {         for (auto n : l_friends) {             print("{}", n.s_name);             if (--n_animals) print(", ");         }     }     print("\n");     animal_mutex.unlock(); }

  Now our output is sensible:

  add_friend Bugs -> Felix add_friend Bugs -> Astro add_friend Felix -> Hobbes add_friend Felix -> Bugs Animal: Felix, friends: Bugs, Hobbes Animal: Hobbes, friends: Felix Animal: Astro, friends: Bugs Animal: Bugs, friends: Felix, Astro delete_friend Felix -> Bugs Animal: Felix, friends: Hobbes Animal: Bugs, friends: Astro

  Your output may have the lines in a different order due to asynchronous operation but the details should be the same.

• The lock() and unlock() methods are rarely called directly. The std::lock_guard class manages locks with a proper RAII ( Resource Acquisition Is Initialization ) pattern that automatically releases the lock upon destruction. Here's the add_friend() method with lock_guard :

  bool add_friend(Animal& o) noexcept {     println("add_friend {} -> {}", s_name, o.s_name);     if (*this == o) return false;      std::lock_guard<std::mutex> l(animal_mutex);     if (!is_friend(o)) {         l_friends.emplace_back(o);     }     if (!o.is_friend(*this)) {         o.l_friends.emplace_back(*this);     }     return true; }

  The lock_guard object is created and holds a lock until it is destroyed. Like the lock() method, lock_guard also blocks until a lock is available.

• Let's apply lock_guard to the delete_friend() and display() methods

  Here is delete_friend() :

  bool delete_friend(Animal& o) noexcept {     println("delete_friend {} -> {}", s_name, o.s_name);     if (*this == o) return false;     std::lock_guard<std::mutex> l(animal_mutex);     if (auto it = find_friend(o)) {         l_friends.erase(it.value());     }     if (auto it = o.find_friend(*this)) {         o.l_friends.erase(it.value());     }      return true; }

  And here is display() :

  void display() const noexcept {     std::lock_guard<std::mutex> l(animal_mutex);     auto n_animals {l_friends.size()};     print("Animal: {}, friends: ", s_name);     if (!n_animals) print("none");     else {         for (auto n : l_friends) {             print("{}", n.s_name);             if (--n_animals) print(", ");         }     }     print("\n"); }

  Our output remains coherent:

  add_friend Felix -> Hobbes add_friend Bugs -> Astro add_friend Felix -> Bugs add_friend Bugs -> Felix Animal: Felix, friends: Bugs, Hobbes Animal: Astro, friends: Bugs Animal: Hobbes, friends: Felix Animal: Bugs, friends: Astro, Felix delete_friend Felix -> Bugs Animal: Felix, friends: Hobbes Animal: Bugs, friends: Astro

  As before, your output may have the lines in a different order due to asynchronous operation.

How it works

It's important to understand that a mutex does not lock data; it blocks execution. As shown in this recipe, when a mutex is applied in object methods it can be used to enforce mutually exclusive access to data.

When one thread locks a mutex , with either lock() or lock_guard , that thread is said to own the mutex . Any other thread that tries to lock the same mutex will be blocked until it is unlocked by the owner.

The mutex object must not be destroyed while it's owned by any thread. Likewise, a thread must not be destroyed while it owns a mutex. An RAII-compliant wrapper like lock_guard will help ensure this doesn't happen.

There's more…

While std::mutex provides an exclusive mutex suitable for many purposes, the STL does provide a few other choices.

• shared_mutex allows more than one thread to simultaneously own a mutex .
• recursive_mutex allows one thread to stack multiple locks on a single mutex .
• timed_mutex provides a timeout for mutex blocks. Both shared_mutex and recursive_mutex also have timed versions available.

Share flags and values with std::atomic

The std::atomic class encapsulates a single object and guarantees it to be atomic , such that concurrent access to that object does not cause a race condition. Memory order semantics may also be specified, controlling how operations on different threads are ordered and made visible. It is typically used for low-level synchronization and lock-free communication among threads.

std::atomic defines an atomic type from its template type. The type must be trivial . A type is trivial if it occupies contiguous memory, has no user-defined constructor, and has no virtual member functions. All primitive types are trivial.

While it is possible to construct a trivial type, std::atomic is most often used with simple primitives, such as bool , int , long , float , and double .

How to do it

This recipe uses a simple function that loops over a counter to demonstrate sharing atomic objects. We will spawn a swarm of these loops as threads that share atomic values.

• Atomic objects are often placed in the global namespace. They must be accessible to all the threads that need to share their value:

  std::atomic<bool> ready {}; std::atomic<uint64_t> g_count {}; std::atomic_flag winner {};

  The ready object is a bool that gets set true when all the threads are ready to start counting.

  The g_count object is a global counter. It is incremented by each of the threads.

  The winner object is a special atomic_flag type. It is used to indicate which thread finishes first.

• We use a couple of constants to control the number of threads and the number of loops for each thread:

  constexpr int max_count {1000}; constexpr int max_threads {100};

  I've set it to run 100 threads and count 1,000 iterations in each thread.

• The countem() function is spawned for each thread. It loops max_count times and increments g_count for each iteration of the loop. This is where we use our atomic values:

  void countem (int id) {     while (!ready) std::this_thread::yield();     for (int i {}; i < max_count; ++i) ++g_count;     if (!winner.test_and_set()) {         println("thread {:02} won!", id);     } };

  The ready atomic value is used to synchronize the threads. Each thread will call yield() until the ready value is set true. The yield() function yields execution to other threads.

  Each iteration of the for loop increments the g_count atomic value. The final value should be equal to max_count * max_threads .

  After the loop is complete, the test_and_set() method of the winner object is used to report the winning thread. test_and_set() is a method of the atomic_flag class. It sets the flag and returns the boolean value from before it is set.

• We've used the make_commas() function before. It displays a number with thousands separators:

  string make_commas(const uint64_t num) {     auto s = std::to_string(num);     for (auto i = s.size(); i > 3; i -= 3) {         s.insert(i - 3, ",");     }     return s; }

  The main() function spawns the threads and reports the results:

  int main() {     vector<std::thread> swarm;     println("spawn {} threads", max_threads);     for (int i {}; i < max_threads; ++i) {         swarm.emplace_back(countem, i);     }     ready = true;     for (auto&t:swarm)t.join();     println("global count: {}", make_commas(g_count)); }

  Here we create a vector<std::thread> object to hold the threads.

  In the for loop, we use emplace_back() to create each thread in the vector .

  Once the threads have been spawned, we set the ready flag so the threads may start their loops.

  Output:

  spawn 100 threads thread 67 won! global count: 100,000 Every time you run it a different thread will win.

How it works

The std::atomic class encapsulates an object to synchronize access among multiple threads.

The encapsulated object must be a trivial type , which means it occupies contiguous memory, has no user-defined constructor, and has no virtual member functions. All primitive types are trivial.

It is possible to use a simple struct with atomic:

struct Trivial {     int a;     int b; }; std::atomic<Trivial> triv1;

While this usage is possible, it's not practical. Anything beyond setting and retrieving compound values loses the benefits of atomicity and ends up requiring a mutex . The atomic class is best suited for scalar values.

Specializations

There are specializations of the atomic class for a few different purposes:

1. Pointers and smart pointers
2. Floating point types
3. Integral types

Pointers and smart pointers

The std::atomic<U*> specialization includes support for atomic pointer arithmetic operations, including fetch_add() for addition and fetch_sub() for subtraction.

Floating-point types

When used with the floating-point types float , double , and long double , std::atomic includes support for atomic floating-point arithmetic operations, including fetch_add() for addition and fetch_sub() for subtraction.

Integral types

When used with one of the integral types, std::atomic provides support for additional atomic operations, including fetch_add() , fetch_sub() , fetch_and() , fetch_or() , and fetch_xor() .

Standard aliases

The STL provides type aliases for all the standard scalar integral types. This means that instead of these declarations in our code:

std::atomic<bool> ready{}; std::atomic<uint64_t> g_count{}; We could use: std::atomic_bool ready{}; std::atomic_uint64_t g_count{};

There are 46 standard aliases, one for each of the standard integral types:

Lock-free variations

Most modern architectures provide atomic CPU instructions for performing atomic operations. std::atomic should use hardware support for atomic instructions, where supported by your hardware. Some atomic types may not be supported on some hardware. std::atomic may use a mutex to ensure thread-safe operations for those specializations, causing threads to block while waiting for other threads to complete operations. Specializations that use hardware support are said to be lock-free because they don't require a mutex.

The is_lock_free() method checks whether a specialization is lock-free:

println("is g_count lock-free? {}",g_count.is_lock_free());

Output:

is g_count lock-free? true

This result will be true for most modern architectures.

There are a few guaranteed lock-free variations of std::atomic available. These specializations guarantee the use of the most efficient hardware atomic operations for each purpose:

std::atomic_signed_lock_free is an alias for the most efficient lock-free specialization of a signed integral type.

std::atomic_unsigned_lock_free is an alias for the most efficient lock-free specialization of an unsigned integral type.

The std::atomic_flag class provides a lock-free atomic boolean type.

There's more…

When multiple threads read and write variables simultaneously, one thread may observe the changes in a different order than they were written. std::memory_order specifies how memory access is ordered around an atomic operation.

std::atomic provides methods for accessing and changing its managed value. Unlike the associated operators, these access methods provide for a memory order to be specified. For example:

g_count.fetch_add(1, std::memory_order_seq_cst);

In this case, memory_order_seq_cst specifies sequentially consistent ordering. So, this call to fectch_add() will add 1 to the value of g_count with sequentially consistent ordering.

Other possible memory order constants are:

• memory_order_relaxed

  Relaxed operation: no synchronization or ordering constraints imposed, only atomicity is guaranteed.

• memory_order_consume

  This is a consume operation . Access in the current thread dependent on the value cannot be reordered before this load.

• memory_order_acquire

  This is an acquire operation . Access cannot be reordered before this load.

• memory_order_release

  This is a store operation . Access in the current thread cannot be reordered after this store.

• memory_order_acq_rel

  This is both acquire and release . Access in the current thread cannot be reordered before or after this store.

• memory_order_seq_cst

  This is sequentially consistent ordering, either acquire or release , depending on the context. Load performs acquire, store performs release, and read/write/modify performs both. All threads observe all modifications in the same order.

If memory order is not specified, memory_order_seq_cst is the default.

Initialize threads with std::call_once

You may need to run the same code in many threads but must initialize that code only once.

One solution would be to call the initialization code before running the threads. This approach can work but has some drawbacks. By separating the initialization, it may be called when unnecessary, or it may be missed when necessary.

The std::call_once function provides a more robust solution. call_once is in the <mutex> header.

How to do it

In this recipe we use a print function for the initialization, so we can clearly see when it's called.

• We'll use a constant for the number of threads to spawn:

  constexpr size_t max_threads {25};

• We also need a std::once_flag to synchronize the std::call_once function:

  std::once_flag init_flag;

  The std::once_flag is in static memory and global scope.

• Our initialization function simply prints a string to let us know it's been called:

  void do_init(size_t id) {     print("do_init ({}): ", id); }

• Our worker function, do_print() , uses std::call_once to call the initialization function, then prints its own id:

  void do_print(size_t id) {     std::call_once(init_flag, do_init, id);     print("{} ", id); }

  In main() , we use a list container to manage a set of jthread objects:

  int main() {     list<jthread> spawn;     for (size_t id{}; id < max_threads; ++id) {         spawn.emplace_back(do_print, id);     } }

  Our output shows the initialization happens first, and once:

  do_init (8): 12 0 2 1 9 6 13 10 11 5 16 3 4 17 7 15 8 14 18 19 20 21 22 23 24

  Notice that it's not always the first spawned thread (0) that ends up calling the initialization function, but it is always called first. If you run this repeatedly, you'll see thread 0 gets the initialization often, but not every time. You'll see thread 0 in the initialization more often on a system with fewer cores.

How it works

std::call_once is a template function that takes a flag, a callable (function or functor), and a parameter pack of arguments:

template<class Callable, class... Args> void call_once(once_flag&flag,Callable&&f,Args&&...args);

The callable f is called exactly one time. Even if call_once is called concurrently from several threads, f is still called once and only once.

This requires a std::once_flag object for coordination. The once_flag constructor sets its state to indicate that the callable has not yet been called.

When call_once invokes the callable, any other calls on the same once_flag are blocked until the callable returns. After the callable returns, the once_flag is set and any subsequent calls to call_once return without invoking f .

Resolve the producer-consumer problem with std::condition_variable

The simplest version of the producer-consumer problem is where you have one process that produces data, and another that consumes data, using one buffer or container to hold the data. This requires coordination between the producer and consumer, to manage the buffer and prevent unwanted side-effects.

How to do it

In this recipe we consider a simple solution to the producer-consumer problem using std::condition_variable to coordinate the processes.

• We begin with some namespace and alias declarations for convenience:

  using namespace std::chrono_literals; namespace this_thread = std::this_thread;  using guard_t = std::lock_guard<std::mutex>; using lock_t = std::unique_lock<std::mutex>;

  The guard_t and lock_t aliases make it easier to use lock_guard and unique_lock without error.

• We use a couple of constants:

  constexpr size_t num_items {10}; constexpr auto delay_time {200ms};

  Keeping these in one place makes it safer and easier to experiment with different values.

• We're using these global variables to coordinate the data store:

  std::deque<size_t> q {}; std::mutex mtx {}; std::condition_variable cond {}; bool finished {};

  We're using deque to hold the data as a first-in-first-out (FIFO) queue.

  The mutex is used with the condition_variable to coordinate the movement of data from producer to consumer.

  The finished flag indicates that there is no more data.

• The producer thread will use this function:

  void producer() {     for (size_t i {}; i < num_items; ++i) {         this_thread::sleep_for(delay_time);         guard_t x {mtx};         q.push_back(i);         cond.notify_all();     }      guard_t x {mtx};     finished = true;     cond.notify_all(); }

  The producer() function loops num_items iterations and pushes a number onto the deque each time through the loop.

  We include a sleep_for() call to simulate a delay in producing each value.

  The conditional_variable requires a mutex lock to operate. We use lock_guard (via the guard_t alias) to obtain the lock, then push the value onto the deque , then call notify_all() on the conditional_variable . This tells the consumer thread that there is a new value available.

  When the loop completes, we set the finished flag and notify the consumer thread that the producer is completed.

• The consumer thread waits for each value from the producer, displays it on the console, and waits for the finished flag

  void consumer() {     while(!finished) {         lock_t lck {mtx};         cond.wait(lck, []{             return !q.empty() || finished;         });         while (!q.empty()) {             println("Got {} from the queue", q.front());             q.pop_front();         }     }     println("finished!"); }

  The wait() method waits to be notified by the producer. It uses a lambda as a predicate to continue waiting until the deque is not empty or the finished flag is set.

  When we get a value, we display it and then pop it from the deque .

• We run this in main with simple thread objects:

  int main() {     jthread t1 {producer};     jthread t2 {consumer}; }

  Output:

  Got 0 from the queue Got 1 from the queue Got 2 from the queue Got 3 from the queue Got 4 from the queue Got 5 from the queue Got 6 from the queue Got 7 from the queue Got 8 from the queue Got 9 from the queue finished!

  Notice that there's a 200ms delay between each line. This tells us that the producer-consumer coordination is working as expected.

How it works

The producer-consumer problem requires coordination between writing and reading a buffer or container. In this example, our container is a deque<size_t> :

std::deque<size_t> q {};

The condition_variable class can block a thread, or multiple threads, while a shared variable is modified. It may then notify other threads that the value is available.

condition_variable requires a mutex to perform the lock:

std::lock_guard x {mtx}; q.push_back(i); cond.notify_all();

std::lock_guard acquires a lock so we can push a value onto our deque.

The wait() method on condition_variable is used to block the current thread until it receives a notification:

void wait( std::unique_lock<std::mutex>&lock); void wait( std::unique_lock<std::mutex>& lock,   Pred stop_waiting );

The predicate form of wait() is equivalent to:

while (!stop_waiting()) {     wait(lock); }

The predicate form is used to prevent spurious waking while waiting for a specific condition. We use it with a lambda in our example:

cond.wait(lck, []{ return !q.empty() || finished; });

This prevents the consumer from waking until the deque has data or the finished flag is set.

The condition_variable class has two notification methods:

• notify_one() unblocks one waiting thread
• notify_all() unblocks all waiting threads

We used notify_all() in our example. Because there is only one consumer thread, either notification method would work the same.

Implement multiple producers and consumers

The producer-consumer problem is really a set of problems. Solutions will differ if the buffer is bounded or unbounded, or if there are multiple producers, multiple consumers, or both.

Let's consider a case with multiple producers, multiple consumers, and a bounded (limited capacity) buffer. This is a common condition.

How to do it

In this recipe we'll look at a case with multiple producers and consumers, and a bounded buffer , using a variety of techniques we've covered in this chapter.

• We'll start with some constants for convenience and reliability:

  constexpr auto delay_time {50ms}; constexpr auto consumer_wait {100ms}; constexpr size_t queue_limit {5}; constexpr size_t num_items {15}; constexpr size_t num_producers {3}; constexpr size_t num_consumers {5};

  • delay_time is a duration object, used with sleep_for().
  • consumer_wait is a duration object used with the consumer condition variable.
  • queue_limt is the buffer limit; the maximum number of items in the deque .
  • num_items is the maximum number of items produced per producer.
  • num_producers is the number of spawned producers.
  • num_producers is the number of spawned consumers.
• Now we need some objects to control the process:

  deque<string> qs {}; mutex q_mutex {}; condition_variable cv_producer {}; condition_variable cv_consumer {}; bool production_complete {};

  • qs is a deque of string that holds the produced objects.
  • q_mutex controls access to the deque.
  • cv_producer is a condition variable that coordinates producers.
  • cv_consumer is a condition variable that coordinates consumers.
  • production_complete is set true when all producer threads have finished.
• The producer() threads run this function:

  void producer(const size_t id) {     for (size_t i {}; i < num_items; ++i) {         this_thread::sleep_for(delay_time * id);         unique_lock<mutex> lock(q_mutex);         cv_producer.wait(lock, [&]{             return qs.size() < queue_limit;         });         qs.push_back(format("pid {}, qs  {}, item {:02}\n",             id, qs.size(), i + 1));         cv_consumer.notify_all();     } }

  The passed value id is a sequential number used to identify the producer.

  The for loop repeats num_item times. The sleep_for() function is used to simulate work required to produce an item.

  Then we obtain a unique_lock from q_mutex and invoke wait() on cv_producer , using a lambda that checks the size of the deque against the queue_limit constant. If the deque has reached maximum size, the producer waits for consumer threads to reduce the size of the deque . This represents the bounded buffer limit on the producer.

  Once the condition is satisfied, we push an item onto the deque. The item is a formatted string with the producer id, the size of qs , and an item number ( i + 1 ) from the loop control variable.

  Finally, we notify the consumers that new data is available, with notify_all() on the cv_consumer condition variable.

• The consumer() threads run this function:

  void consumer(const size_t id) {     while (!production_complete) {         unique_lock<mutex> lock(q_mutex);         cv_consumer.wait_for(lock, consumer_wait, [&]{             return !qs.empty();         });         if (!qs.empty()) {             print("cid {}: {}", id, qs.front());             qs.pop_front();         }         cv_producer.notify_all();     } }

  The passed value id is a sequential number used to identify the consumer.

  The main while() loop continues until production_complete is set.

  We obtain a unique_lock from q_mutex and invoke wait_for() on cv_consumer , with a timeout and a lambda that tests if the deque is empty. We need the timeout because it's possible for the producer threads to finish while some of the consumer threads are still running, leaving the deque empty.

  Once we have a non-empty deque , we can print ( consume ) an item and pop it off the deque .

• In main() , we use async() to spawn the producer and consumer threads. async() conforms to the RAII ( Resource Acquisition Is Initialization ) pattern, so I'll usually prefer it over thread , where possible. async() returns a future object, so we'll keep a list of future<void> objects for process management:

  int main() {     list<future<void>> producers;     list<future<void>> consumers;      for (size_t i {}; i < num_producers; ++i) {         producers.emplace_back(async(producer, i));     }      for (size_t i {}; i < num_consumers; ++i) {         consumers.emplace_back(async(consumer, i));     }     ...

  We use for loops to create producer and consumer threads.

• Finally, we use a list of future objects to determine when our producer and consumer threads are complete:

  for (auto& f : producers) f.wait(); production_complete = true; println("producers done.");  for (auto& f : consumers) f.wait(); println("consumers done.");

  We loop through our producers container to wait() for the producer threads to complete. Then we can set the production_complete flag. We likewise loop through the consumers container to wait() for the consumer threads to complete. We could perform any final analysis or completion processes here.

• The output is a bit long to show in its entirety:

  cid 0: pid 0, qs  0, item 01 cid 0: pid 0, qs  1, item 02 cid 0: pid 0, qs  2, item 03 cid 0: pid 0, qs  3, item 04 cid 0: pid 0, qs  4, item 05 ... cid 4: pid 2, qs  0, item 12 cid 4: pid 2, qs  0, item 13 cid 3: pid 2, qs  0, item 14 cid 0: pid 2, qs  0, item 15 producers done. consumers done.

How it works

The heart of this recipe is in the use of two condition_variable objects to control the producer and consumer threads asynchronously:

condition_variable cv_producer{}; condition_variable cv_consumer{};

In the producer() function, the cv_producer object obtains a unique_lock , waits for the deque to be available, and notifies the cv_consumer object when an item has been produced:

void producer(const size_t id) {     for (size_t i {}; i < num_items; ++i) {         this_thread::sleep_for(delay_time * id);         unique_lock<mutex> lock(q_mutex);         cv_producer.wait(lock, [&]{             return qs.size() < queue_limit;         });         qs.push_back(format("pid {}, qs  {}, item {:02}\n",             id, qs.size(), i + 1));         cv_consumer.notify_all();     } }

Conversely, in the consumer() function, the cv_consumer object obtains a unique_lock , waits for the deque to have items, and notifies the cv_producer object when an item has been consumed:

void consumer(const size_t id) {     while (!production_complete) {         unique_lock<mutex> lock(q_mutex);         cv_consumer.wait_for(lock, consumer_wait, [&]{             return !qs.empty();         });         if (!qs.empty()) {             print("cid {}: {}", id, qs.front());             qs.pop_front();         }         cv_producer.notify_all();     } }

These complementary locks, waits, and notifications constitute the balance of coordination between multiple producers and consumers.

Suspend and resume execution with coroutines

Introduced with C++20, coroutines were not yet implemented consistently across platforms in time for the previous edition of this book.

Coroutines allow functions to suspend and resume execution without blocking. This provides support for true, language-level asynchronous programming, without reliance on threads, callbacks, or explicit state management.

Today, all major compilers support coroutines reliably. Libraries have stabilized and debugging and performance have improved. This means that coroutines are practical and portable across platforms.

This recipe demonstrates a simple cooperative scheduler built with coroutines. Multiple tasks run on a single thread and explicitly yield control, allowing them to progress without preemption or synchronization.

How to do it

In this recipe we create a non-blocking cooperative scheduler using a coroutine to execute tasks.

• Every coroutine requires a promise_type object to manage the lifecycle and behavior of the coroutine. The promise_type is encapsulated in a task class for use by our coroutine:

  class task { public:     class promise_type; // forward declaration     using co_h = std::coroutine_handle<promise_type>;     co_h handle;      class promise_type {     public:         task get_return_object() {             return task{                 co_h::from_promise(*this)             };         }          std::suspend_always initial_suspend() noexcept           { return {}; }         std::suspend_always final_suspend() noexcept           { return {}; }         void return_void() noexcept {}         void unhandled_exception() { std::terminate(); }     };      explicit task(co_h h)         : handle(h) {}      // move only, no copying     task(task&& other) noexcept : handle(other.handle)         { other.handle = {}; }     task(consttask&)=delete;     task& operator=(consttask&)=delete;     task& operator=(task&&)=delete;      ~task() {         if (handle) handle.destroy();     } };

  The task class encapsulates the promise_type , which provides hooks for the compiler to call when it encounters a coroutine statement. These hooks provide behaviors for the coroutine statements: co_await , co_yield , and co_return .

  The task class is move-only, so we delete the copy constructor and copy operators.

• The scheduler itself is in a class called scheduler :

  class scheduler {     std::deque<std::coroutine_handle<>> ready;  public:     void schedule(std::coroutine_handle<> h) {         ready.push_back(h);     }      void run() {         while (!ready.empty()) {             auto h = ready.front();             ready.pop_front();             h.resume();         }     } };

  The scheduler class manages a deque of coroutine_handle objects in a FIFO (first-in, first-out) queue, providing predictable ordered behavior. The schedule() function pushes coroutine handles onto the back of the queue and the run() function pops them off the front and resumes each coroutine in turn.

• The yield class provides the awaitable protocol for use with the co_await statement in the coroutine:

  class yield { scheduler&y_sched; public:     explicit yield(scheduler& s) noexcept : y_sched(s) {}     bool await_ready() const noexcept { return false; }     void await_resume() const noexcept {}        void await_suspend(std::coroutine_handle<> h) const noexcept {         y_sched.schedule(h);     } };

  The constructor initializes a reference to the scheduler so that a single schedule queue may be properly managed. The member functions, await_ready() , await_resume() , and await_suspend() , provide hooks for the co_await coroutine statement.

• The coroutine itself is called worker() :

  task worker(scheduler& sched, int id) {     for (int i = 0; i < 3; ++i) {         println("worker {} step {}", id, i);         co_await yield {sched};     } }

  The coroutine itself is rather simple, just a loop and a co_await expression.

  co_await is syntactically an operator and yield {sched} is the operand. The compiler uses the yield object for its awaitable protocol to run the scheduler.

• Now we can invoke our coroutine in the main() function:

  int main() {     scheduler sched {};      // Create tasks     auto t1 = worker(sched, 1);     auto t2 = worker(sched, 2);     auto t3 = worker(sched, 3);      // Schedule initial execution     sched.schedule(t1.handle);     sched.schedule(t2.handle);     sched.schedule(t3.handle);      // Run event loop     sched.run(); }

  We create a scheduler object, sched , and call worker(sched) three times. We take the coroutine handles from the three distinct task instances and pass those to the scheduler, then run it with sched.run() .

  Our output:

  worker 1 step 0 worker 2 step 0 worker 3 step 0 worker 1 step 1 worker 2 step 1 worker 3 step 1 worker 1 step 2 worker 2 step 2 worker 3 step 2

  We see that our coroutine tasks are executed by the scheduler in the expeted sequential order.

How it works

The bare minimum requirement for a functional coroutine is a coroutine function, which must contain a coroutine keyword and an associated promise type. The promise type is associated with the coroutine function's return type. For example:

say_hello hello() {     println("before suspension");     co_await std::suspend_always{};     println("after resumption"); }

Coroutine keywords include, co_await , co_yield , and co_return . These keywords are associated with functions within the promise_type class.

In this minimal example, the compiler will look for say_hello::promise_type to define how the coroutine is created, suspended, resumed, and destroyed. This means we need a class called say_hello that encapsulates a promise_type class:

class say_hello { public:     class promise_type;     using co_h = std::coroutine_handle<promise_type>;     co_h h;      class promise_type {     public:         say_hello get_return_object() {             return say_hello{                 co_h::from_promise(*this)             };         }          std::suspend_never         initial_suspend() noexcept { return {}; }                std::suspend_always         final_suspend() noexcept { return {}; }                void return_void() noexcept {}         void unhandled_exception() {}     };      explicit say_hello(co_h h) : h(h) {}        say_hello(say_hello&& other) noexcept : handle(other.handle)            { other.handle = {}; }     say_hello(constsay_hello&)=delete;     say_hello& operator=(constsay_hello&)=delete;      ~say_hello() {         if (h) h.destroy();     } };

The say_hello class provides ownership and manages the lifetime of the coroutine. This includes the promise_type required for coroutine operation.

Our hello() function uses the coroutine statement:

co_await std::suspend_always{};

The compiler then uses the awaitable protocol defined by std::suspend_always to determine whether execution should suspend. The promise_type class defines suspension points for coroutine creation and completion. In this example, promise_type::final_suspend() returns std::suspend_always , causing the coroutine to suspend when it reaches the end of execution:

std::suspend_always say_hello::promise_type::final_suspend() { return {}; }

The coroutine in our scheduler example contains a few additional moving parts, often included in coroutine implementations:

• The task class provides ownership and manages the lifetime of the coroutine frame. This is the class that includes the promise_type .
• A scheduler class maintains the deque of coroutine handles.
• The yield class implements the awaitable protocol to hand off execution to the scheduler. The awaitable protocol is implemented by member functions that operate as hooks:

  class yield {     scheduler& y_sched; public:     explicit yield(scheduler& s) noexcept : y_sched(s) {}     bool await_ready() const noexcept { return false; }     void await_resume() const noexcept {}        void await_suspend(std::coroutine_handle<> h) const noexcept {         y_sched.schedule(h);     } };

  The await_ready() function returns false , indicating that the coroutine should always suspend when this awaitable is awaited.

  The await_suspend() function is called when suspension occurs. It passes the coroutine handle to the scheduler. This hands control over scheduling decisions to the scheduler without blocking the current thread.

  The await_resume() function is invoked when the coroutine is later resumed.

  The worker() function represents the coroutine itself and returns the task object.

This example provides a functional introduction to implementing a coroutine.

Use coroutines for a producer-consumer solution

Earlier in this chapter we used threads for a producer-consumer solution. We've also seen how to implement a scheduler with coroutines. In this recipe, we'll put all our knowledge to work to solve the producer-consumer problem with coroutines.

How to do it

In this recipe we use two coroutines, one each for the producer and the consumer. These are managed with a common scheduler.

• The task class is consistent with the same class in our scheduler recipe. In fact, this will often be the same code for different solutions:

  class task { public:     class promise_type;     using co_h = std::coroutine_handle<promise_type>;     co_h handle;      class promise_type {     public:         task get_return_object() {             return task{                 co_h::from_promise(*this)             };         }          std::suspend_always             initial_suspend() noexcept { return {}; }         std::suspend_always final_suspend()             noexcept { return {}; }         void return_void() noexcept {}         void unhandled_exception() { std::terminate(); }     };      explicit task(co_h h) : handle(h) {}     task(task&& other) noexcept : handle(other.handle)         { other.handle = {}; }     task(consttask&)=delete;     task& operator=(consttask&)=delete;     task& operator=(task&&)=delete;     ~task() { if (handle) handle.destroy(); } };

  The task class encapsulates the promise_type which provides hooks for the coroutine.

• The scheduler class has some additional functions for managing resources that are shared between the producer and consumer

  class scheduler {     std::deque<std::coroutine_handle<>> ready;     bool f_done {false};     std::deque<size_t> queue {}; public:     void schedule(std::coroutine_handle<> h) {         ready.push_back(h);     }      void run() {         while (!ready.empty()) {             auto h = ready.front();             ready.pop_front();             h.resume();         }     }      void done(bool f) { f_done = f; }     bool done() const { return f_done; }        void q_push(const size_t& n) { queue.push_back(n); }     bool q_empty() const { return queue.empty(); }     size_t q_pop() {         size_t n = queue.front();         queue.pop_front();         return n;     } };

  We've added bool f_done as a flag for the producer to tell the consumer that it's completed. We also added a deque to hold a queue of the producer's work to be consumed by the consumer. These shared resources have proper encapsulation to coordinate their usage: a setter and getter for the flag and push, pop, and empty functions for the queue.

• The yield class is also unchanged from our scheduler recipe

  class yield {     scheduler& y_sched; public:     explicit yield(scheduler& s) noexcept : y_sched(s) {}     bool await_ready() const noexcept { return false; }     void await_resume() const noexcept {}        void await_suspend(std::coroutine_handle<> h) const noexcept {         y_sched.schedule(h);     } };

  The yield class provides the awaitable protocol for use by the co_await statement in the coroutine.

• The producer coroutine is simply a for loop that pushes data into the production queue:

  task producer(scheduler& sched) {     for (size_t i = 0; i < num_items; ++i) {         sched.q_push(i);         println("produced {}", i);         co_await yield{sched};     }     sched.done(true); }

  It uses sched.q_push() to push data onto the queue. The co_await statement uses the yield awaitable to pass control to the scheduler. We call sched.done(true) when the loop completes.

• The consumer coroutine runs a loop until it sees an empty queue and the sched.done() flag is set

  task consumer(scheduler& sched) {     while (!sched.q_empty() || !sched.done()) {         if (!sched.q_empty()) {             println("consumed {}", sched.q_pop());         }         co_await yield{sched};     } }

  The || condition in the while loop may seem counterintuitive. In the producer-consumer idiom, it's possible for the queue to be empty while more work may still arrive. Conversely, because the coroutines run asynchronously, it's also possible for the done flag to be set while items remain in the queue. For these reasons the consumer must continue running until the queue is empty and production is complete.

• The main() function calls the coroutine functions and passes the handles to a scheduler object

  int main() {     scheduler sched{};      auto p = producer(sched);     auto c = consumer(sched);      sched.schedule(p.handle);     sched.schedule(c.handle);      sched.run(); }

  We take the coroutine handles from the two distinct coroutines and pass those to the scheduler. And then call sched.run() .

  Our output:

  produced 0 consumed 0 produced 1 consumed 1 produced 2 consumed 2 produced 3 consumed 3 ...

How it works

This recipe builds on our scheduler recipe by reusing the scheduler and task classes while adding producer-consumer coordination. The key distinction is in the shared data between the producer and consumer coroutine functions.

The producer must create product and the consumer must have access to that product. This is accomplished by sharing data in the scheduler class.

The scheduler class has three data members:

std::deque<std::coroutine_handle<>> ready; bool f_done{false}; std::deque<size_t> queue {};

The deque<std::coroutine_handle<>> ready queue is carried over from the previous recipe. This is simply a queue of coroutine handles to be run by the scheduler.

The f_done flag is shared with a setter and getter:

void done(bool f) { f_done = f; } bool done() const { return f_done; }

This maintains good encapsulation while allowing the producer to tell the consumer when the work is complete.

The std::deque<size_t> queue contains the work of the producer. This is shared by a set of convenience methods:

void q_push(const size_t& n) { queue.push_back(n); } bool q_empty() const { return queue.empty(); } size_t q_pop() {     size_t n = queue.front();     queue.pop_front();     return n; }

The producer can add product to the queue with q_push() , and the consumer can check the state of the queue with q_empty() or pop items from the queue with q_pop() . This adds convenience over the deque interface while maintaining the safety of encapsulation.

Coroutines can often simplify concurrency by eliminating the need for locks and condition variables. Because coroutines suspend and resume on a single thread, it's easier to share data without the complexity and risk of synchronization primitives.

While coroutines provide efficient cooperative concurrency, unlike threads they do not inherently provide parallel execution across multiple processor cores. If optimal performance on multiple cores is critical, coroutines may be combined with threads or thread pools.

Subscribe to Deep Engineering

Join thousands of developers and architects who want to understand how software is changing, deepen their expertise, and build systems that last.

Deep Engineering is a weekly expert-led newsletter for experienced practitioners, featuring original analysis, technical interviews, and curated insights on architecture, system design, and modern programming practice.

Scan the QR or visit the link to subscribe for free.