concurrencpp Documentation

You can monitor when concurrencpp workers create or terminate threads by setting callbacks in concurrencpp::runtime_options. These callbacks are executed from within the created or terminating thread, so std::this_thread::get_id() will return the relevant thread ID.

Callbacks must be copiable because they are copied to each worker. The callback signature is void callback(std::string_view thread_name), where thread_name is a non-unique title used for logging and debugging.

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

int main() {
    concurrencpp::runtime_options options;
    options.thread_started_callback = [](std::string_view thread_name) {
        std::cout << "A new thread is starting to run, name: " << thread_name << ", thread id: " << std::this_thread::get_id()
                  << std::endl;
    };

    options.thread_terminated_callback = [](std::string_view thread_name) {
        std::cout << "A thread is terminating, name: " << thread_name << ", thread id: " << std::this_thread::get_id() << std::endl;
    };

    concurrencpp::runtime runtime(options);
    // ... rest of application
}

The runtime object is the central agent for managing executors and the global timer queue.

Lifecycle Management

• Creation: Create a runtime object as a value type at the very beginning of your main function. This ensures no tasks are processed before the runtime exists.
• Destruction (RAII): When the runtime object goes out of scope, it automatically shuts down all registered executors and the timer queue.
• Termination: Ongoing tasks should exit as soon as possible during shutdown. If a task tries to use an executor after the runtime has begun shutting down, a concurrencpp::runtime_shutdown exception will be thrown.

Executor Acquisition

Use the runtime object to obtain standard executors:

• thread_pool_executor(): A standard thread pool.
• inline_executor(): Executes tasks immediately on the current thread.
• background_executor(): A background thread pool.
• thread_executor(): A single-threaded executor.

Custom Executors

You can register custom executors with the runtime using:

• make_worker_thread_executor()
• make_manual_executor()
• make_executor<executor_type>(args...)

To implement a custom executor, inherit from concurrencpp::derivable_executor<T>.

Key Requirements:

1. Thread Safety: Executors are used from multiple threads; all implemented methods must be thread-safe.
2. Instantiation: Use runtime::make_executor<T>(args...) to create new executors. Do not use std::make_shared, new, or attempt to re-instantiate built-in executors like thread_pool_executor.
3. Shutdown Handling:
   • shutdown(): Must signal underlying threads to quit and then join them. It must handle multiple calls by ignoring subsequent calls after the first.
   • shutdown_requested(): Should monitor the executor state.
   • enqueue(): Must throw concurrencpp::errors::runtime_shutdown if shutdown() has already been called.
4. Task Storage: Implementations are responsible for storing concurrencpp::task objects received via enqueue and executing them according to the executor's mechanism.

#include "concurrencpp/concurrencpp.h"
#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>

class logging_executor : public concurrencpp::derivable_executor<logging_executor> {
private:
    mutable std::mutex _lock;
    std::queue<concurrencpp::task> _queue;
    std::condition_variable _condition;
    bool _shutdown_requested;
    std::thread _thread;
    const std::string _prefix;

    void work_loop() {
        while (true) {
            std::unique_lock<std::mutex> lock(_lock);
            if (_shutdown_requested) return;

            if (!_queue.empty()) {
                auto task = std::move(_queue.front());
                _queue.pop();
                lock.unlock();
                std::cout << _prefix << " A task is being executed" << std::endl;
                task();
                continue;
            }

            _condition.wait(lock, [this] { return !_queue.empty() || _shutdown_requested; });
        }
    }

public:
    logging_executor(std::string_view prefix) :
        derivable_executor<logging_executor>("logging_executor"),
        _shutdown_requested(false),
        _prefix(prefix) {
        _thread = std::thread([this] { work_loop(); });
    }

    void enqueue(concurrencpp::task task) override {
        std::cout << _prefix << " A task is being enqueued!" << std::endl;
        std::unique_lock<std::mutex> lock(_lock);
        if (_shutdown_requested) {
            throw concurrencpp::errors::runtime_shutdown("logging executor - executor was shutdown.");
        }
        _queue.emplace(std::move(task));
        _condition.notify_one();
    }

    void enqueue(std::span<concurrencpp::task> tasks) override {
        std::cout << _prefix << tasks.size() << " tasks are being enqueued!" << std::endl;
        std::unique_lock<std::mutex> lock(_lock);
        if (_shutdown_requested) {
            throw concurrencpp::errors::runtime_shutdown("logging executor - executor was shutdown.");
        }
        for (auto& task : tasks) {
            _queue.emplace(std::move(task));
        }
        _condition.notify_one();
    }

    int max_concurrency_level() const noexcept override { return 1; }

    bool shutdown_requested() const noexcept override {
        std::unique_lock<std::mutex> lock(_lock);
        return _shutdown_requested;
    }

    void shutdown() noexcept override {
        std::cout << _prefix << " shutdown requested" << std::endl;
        std::unique_lock<std::mutex> lock(_lock);
        if (_shutdown_requested) return;
        _shutdown_requested = true;
        lock.unlock();
        _condition.notify_one();
        _thread.join();
    }
};

int main() {
    concurrencpp::runtime runtime;
    auto logging_ex = runtime.make_executor<logging_executor>("Session #1234");

    for (size_t i = 0; i < 10; i++) {
        logging_ex->post([] { std::cout << "hello world" << std::endl; });
    }

    std::getchar();
    return 0;
}