dynamicThreadPool.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_DYNAMICTHREADPOOL_HPP
#define RTCTK_COMPONENTFRAMEWORK_DYNAMICTHREADPOOL_HPP
 
#include <atomic>
#include <cassert>
#include <forward_list>
#include <memory>
#include <mutex>
#include <stop_token>
#include <thread>
 
#include <functional>
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
#include <utility>
 
namespace rtctk::componentFramework {

class DynamicThreadPool {
public:
    explicit DynamicThreadPool(std::string name, numapp::NumaPolicies policies = {})
        : m_name(std::move(name)), m_policies(std::move(policies)) {
    }
 
    // neither copyable nor movable
    DynamicThreadPool(const DynamicThreadPool&) = delete;
    DynamicThreadPool(DynamicThreadPool&&) = delete;
    DynamicThreadPool& operator=(const DynamicThreadPool&) = delete;
    DynamicThreadPool& operator=(DynamicThreadPool&&) = delete;
 
    ~DynamicThreadPool() {
        std::scoped_lock lock(m_worker_mutex);
        for (auto& w : m_workers) {
            w.m_thread.request_stop();
        }
        m_workers.clear();  // jthreads auto-join
    }

    template <class Func, class... Args>
    void Submit(Func&& func, Args&&... args) {
        std::scoped_lock lock(m_worker_mutex);
        bool stored = false;
        auto prev = m_workers.before_begin();
        auto cur = m_workers.begin();
 
        while (cur != m_workers.end()) {
            if (not cur->m_alive) {
                if (not stored) {
                    // false linter error
                    // NOLINTBEGIN(bugprone-use-after-move)
                    cur->Run(
                        m_name, m_policies, std::forward<Func>(func), std::forward<Args>(args)...);
                    // NOLINTEND(bugprone-use-after-move)
                    stored = true;
                    prev = cur;
                    ++cur;
                } else {
                    // we already have a place, remove old thread
                    cur = m_workers.erase_after(prev);
                }
            } else {
                prev = cur;
                ++cur;
            }
        }
        // no spot free
        if (not stored) {
            m_workers.emplace_after(prev)->Run(
                m_name, m_policies, std::forward<Func>(func), std::forward<Args>(args)...);
        }
    }
 
private:
    class Worker;
 
    std::string m_name;
    numapp::NumaPolicies m_policies;
    std::mutex m_worker_mutex;
    std::forward_list<Worker> m_workers;
 
    class Worker {
    public:
        // no copy because of thread, no move because of atomic
        Worker(const Worker&) = delete;
        Worker(Worker&&) = delete;
        Worker& operator=(const Worker&) = delete;
        Worker& operator=(Worker&&) = delete;
        Worker() = default;
        template <class Func, class... Args>
        void Run(std::string_view name,
                 const numapp::NumaPolicies& policies,
                 Func&& func,
                 Args&&... args) {
            bool before = m_alive.exchange(true);
            // The following line is needed to avoid compiler warnings in release builds,
            // where assert becomes a no-op.
            (void)before;
            assert(not before);
            m_thread = numapp::MakeJthread(
                name,
                policies,
                [alive_guard = AliveGuard{&m_alive, &AliveGuardFunction},
                 func = std::forward<Func>(func)](std::stop_token st, auto&&... args) mutable {
                    if constexpr (std::is_invocable_v<std::decay_t<Func>,
                                                      std::stop_token,
                                                      std::decay_t<Args>...>) {
                        std::invoke(func, st, std::move(args)...);
                    } else {
                        std::invoke(func, move(args)...);
                    }
                },
                std::forward<Args>(args)...);
        }
        std::jthread m_thread;
        std::atomic<bool> m_alive{false};
 
    private:
        static void AliveGuardFunction(std::atomic<bool>* alive_ptr) {
            alive_ptr->store(false, std::memory_order_release);
        }
        using AliveGuard = std::unique_ptr<std::atomic<bool>, decltype(&AliveGuardFunction)>;
    };
};
}  // namespace rtctk::componentFramework
 
#endif  // RTCTK_COMPONENTFRAMEWORK_DYNAMICTHREADPOOL_HPP