concurrencpp, the C++ concurrency library

concurrencpp brings the power of concurrent tasks to the C++ world, allowing developers to write highly concurrent applications easily and safely by using tasks, executors and coroutines. By using concurrencpp applications can break down big procedures that need to be processed asynchronously into smaller tasks that run concurrently and work in a co-operative manner to achieve the wanted result. concurrencpp also allows applications to write parallel algorithms easily by using parallel coroutines.

concurrencpp main advantages are: * Writing modern concurrency code using higher level tasks instead of low level primitives like std::thread and std::mutex. * Writing highly concurrent and parallel applications that scale automatically to use all hardware resources, as needed. * Achieving non-blocking, synchronous-like code easily by using C++20 coroutines and the co_await keyword. * Reducing the possibility of race conditions, data races and deadlocks by using high-level objects with built-in synchronization. * concurrencpp provides various types of commonly used executors with a complete coroutine integration. * Applications can extend the library by implementing their own provided executors. * concurrencpp is mature and well tested on various platforms and operating systems.

### Table of contents * concurrencpp overview * Tasks * concurrencpp coroutines * Executors * executor API * Executor types * Using executors * thread_pool_executor API * manual_executor API * Result objects * result type * result API * lazy_result type * lazy_result API * Parallel coroutines * Parallel Fibonacci example * Result-promises * result_promise API * result_promise example * Shared result objects * shared_result API * shared_result example * Termination in concurrencpp * Resume executors * Utility functions * make_ready_result * make_exceptional_result * when_all * when_any * resume_on * Timers and Timer queues * timer_queue API * timer API * Regular timer example * Oneshot timers * Oneshot timer example * Delay objects * Delay object example * Generators
* generator API * generator example * Asynchronous locks
* async_lock API * scoped_async_lock API * async_lock example * Asynchronous condition variable
* async_condition_variable API * async_condition_variable example * The runtime object * runtime API * Thread creation and termination monitoring * Creating user-defined executors * task objects * task API * Writing a user-defined executor example * Supported platforms and tools * Building, installing and testing

concurrencpp overview

concurrencpp is built around the concept of concurrent tasks. A task is an asynchronous operation. Tasks offer a higher level of abstraction for concurrent code than traditional thread-centric approaches. Tasks can be chained together, meaning that tasks pass their asynchronous result from one to another, where the result of one task is used as if it were a parameter or an intermediate value of another ongoing task. Tasks allow applications to utilize available hardware resources better and scale much more than using raw threads, since tasks can be suspended, awaiting another task to produce a result, without blocking underlying OS-threads. Tasks bring much more productivity to developers by allowing them to focus more on business-logic and less on low-level concepts like thread management and inter-thread synchronization.

While tasks specify what actions have to be executed, executors are worker-objects that specify where and how to execute tasks. Executors spare applications the tedious management of thread pools and task queues. Executors also decouple those concepts away from application code, by providing a unified API for creating and scheduling tasks.

Tasks communicate with each other using result objects. A result object is an asynchronous pipe that pass the asynchronous result of one task to another ongoing-task. Results can be awaited and resolved in a non-blocking manner.

These three concepts - the task, the executor and the associated result are the building blocks of concurrencpp. Executors run tasks that communicate with each other by sending results through result-objects. Tasks, executors and result objects work together symbiotically to produce concurrent code which is fast and clean.

concurrencpp is built around the RAII concept. In order to use tasks and executors, applications create a runtime instance in the beginning of the main function. The runtime is then used to acquire existing executors and register new user-defined executors. Executors are used to create and schedule tasks to run, and they might return a result object that can be used to pass the asynchronous result to another task that acts as its consumer. When the runtime is destroyed, it iterates over every stored executor and calls its shutdown method. Every executor then exits gracefully. Unscheduled tasks are destroyed, and attempts to create new tasks will throw an exception.

"Hello world" program using concurrencpp:

#include "concurrencpp/concurrencpp.h"
#include <iostream>

int main() {
    concurrencpp::runtime runtime;
    auto result = runtime.thread_executor()->submit([] {
        std::cout << "hello world" << std::endl;
    });

    result.get();
    return 0;
}

In this basic example, we created a runtime object, then we acquired the thread executor from the runtime. We used submit to pass a lambda as our given callable. This lambda returns void, hence, the executor returns a result<void> object that passes the asynchronous result back to the caller. main calls get which blocks the main thread until the result becomes ready. If no exception was thrown, get returns void. If an exception was thrown, get re-throws it. Asynchronously, thread_executor launches a new thread of execution and runs the given lambda. It implicitly co_return void and the task is finished. main is then unblocked.

Concurrent even-number counting:

#include "concurrencpp/concurrencpp.h"

#include <iostream>
#include <vector>
#include <algorithm>

#include <ctime>

using namespace concurrencpp;

std::vector<int> make_random_vector() {
    std::vector<int> vec(64 * 1'024);

    std::srand(std::time(nullptr));
    for (auto& i : vec) {
        i = ::rand();
    }

    return vec;
}

result<size_t> count_even(std::shared_ptr<thread_pool_executor> tpe, const std::vector<int>& vector) {
    const auto vecor_size = vector.size();
    const auto concurrency_level = tpe->max_concurrency_level();
    const auto chunk_size = vecor_size / concurrency_level;

    std::vector<result<size_t>> chunk_count;

    for (auto i = 0; i < concurrency_level; i++) {
        const auto chunk_begin = i * chunk_size;
        const auto chunk_end = chunk_begin + chunk_size;
        auto result = tpe->submit([&vector, chunk_begin, chunk_end]() -> size_t {
            return std::count_if(vector.begin() + chunk_begin, vector.begin() + chunk_end, [](auto i) {
                return i % 2 == 0;
            });
        });

        chunk_count.emplace_back(std::move(result));
    }

    size_t total_count = 0;

    for (auto& result : chunk_count) {
        total_count += co_await result;
    }

    co_return total_count;
}

int main() {
    concurrencpp::runtime runtime;
    const auto vector = make_random_vector();
    auto result = count_even(runtime.thread_pool_executor(), vector);
    const auto total_count = result.get();
    std::cout << "there are " << total_count << " even numbers in the vector" << std::endl;
    return 0;
}

In this example, we start the program by creating a runtime object. We create a vector filled with random numbers, then we acquire the thread_pool_executor from the runtime and call count_even. count_even is a coroutine that spawns more tasks and co_awaits for them to finish inside. max_concurrency_level returns the maximum amount of workers that the executor supports, In the threadpool executor case, the number of workers is calculated from the number of cores. We then partition the array to match the number of workers and send every chunk to be processed in its own task. Asynchronously, the workers count how many even numbers each chunk contains, and co_return the result. count_even sums every result by pulling the count using co_await, the final result is then co_returned. The main thread, which was blocked by calling get is unblocked and the total count is returned. main prints the number of even numbers and the program terminates gracefully.

Tasks

Every big or complex operation can be broken down to smaller and chainable steps. Tasks are asynchronous operations implementing those computational steps. Tasks can run anywhere with the help of executors. While tasks can be created from regular callables (such as functors and lambdas), Tasks are mostly used with coroutines, which allow smooth suspension and resumption. In concurrencpp, the task concept is represented by the concurrencpp::task class. Although the task concept is central to concurrenpp, applications will rarely have to create and manipulate task objects themselves, as task objects are created and scheduled by the runtime with no external help.

concurrencpp coroutines

concurrencpp allows applications to produce and consume coroutines as the main way of creating tasks. concurrencpp supports both eager and lazy tasks.

Eager tasks start to run the moment they are invoked. This type of execution is recommended when applications need to fire an asynchronous action and consume its result later on (fire and consume later), or completely ignore the asynchronous result (fire and forget).

Eager tasks can return result or null_result. result return type tells the coroutine to pass the returned value or the thrown exception (fire and consume later) while null_result return type tells the coroutine to drop and ignore any of them (fire and forget).

Eager coroutines can start to run synchronously, in the caller thread. This kind of coroutines is called "regular coroutines". Concurrencpp eager coroutines can also start to run in parallel, inside a given executor, this kind of coroutines is called "parallel coroutines".

Lazy tasks, on the other hand, start to run only when co_awaited. This type of tasks is recommended when the result of the task is meant to be consumed immediately after creating the task. Lazy tasks, being deferred, are a bit more optimized for the case of immediate-consumption, as they do not need special thread-synchronization in order to pass the asynchronous result back to its consumer. The compiler might also optimize away some memory allocations needed to form the underlying coroutine promise. It is not possible to fire a lazy task and execute something else meanwhile - the firing of a lazy-callee coroutine necessarily means the suspension of the caller-coroutine. The caller coroutine will only be resumed when the lazy-callee coroutine completes. Lazy tasks can only return lazy_result.

Lazy tasks can be converted to eager tasks by calling lazy_result::run. This method runs the lazy task inline and returns a result object that monitors the newly started task. If developers are unsure which result type to use, they are encouraged to use lazy results, as they can be converted to regular (eager) results if needed.

When a function returns any of lazy_result, result or null_resultand contains at least one co_await or co_return in its body, the function is a concurrencpp coroutine. Every valid concurrencpp coroutine is a valid task. In our count-even example above, count_even is such a coroutine. We first spawned count_even, then inside it the threadpool executor spawned more child tasks (that are created from regular callables), that were eventually joined using co_await.

Executors

A concurrencpp executor is an object that is able to schedule and run tasks. Executors simplify the work of managing resources such as threads, thread pools and task queues by decoupling them away from application cod