ThreadPool.h

Go to the documentation of this file.

 
// SPDX-License-Identifier: LGPL-2.1-or-later
 
#ifndef GUL17_THREADPOOL_H_
#define GUL17_THREADPOOL_H_
 
#include <chrono>
#include <condition_variable>
#include <functional>
#include <future>
#include <limits>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <thread>
#include <vector>
 
#include <gul17/cat.h>
#include <gul17/traits.h>
 
namespace gul17 {
 
class ThreadPool;
 
namespace detail {
 
GUL_EXPORT
std::shared_ptr<ThreadPool> lock_pool_or_throw(std::weak_ptr<ThreadPool> pool);
 
} // namespace detail
 
enum class TaskState
{
    pending,  
    running,  
    complete, 
    canceled  
};
 
class ThreadPool : public std::enable_shared_from_this<ThreadPool>
{
public:
    using TaskId = std::uint64_t;
 
    using ThreadId = std::vector<std::thread>::size_type;
 
    template <typename T>
    class TaskHandle
    {
    public:
        TaskHandle()
        {}
 
        TaskHandle(TaskId id, std::future<T> future, std::shared_ptr<ThreadPool> pool)
            : future_{ std::move(future) }
            , id_{ id }
            , pool_{ std::move(pool) }
        {}
 
        bool cancel()
        {
            future_ = {};
            return detail::lock_pool_or_throw(pool_)->cancel_pending_task(id_);
        }
 
        T get_result()
        {
            if (not future_.valid())
                throw std::logic_error("Canceled task has no result");
            return future_.get();
        }
 
        bool is_complete() const
        {
            if (not future_.valid())
                return false;
 
            return future_.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
        }
 
        TaskState get_state() const
        {
            const auto state = detail::lock_pool_or_throw(pool_)->get_task_state(id_);
 
            if (state == InternalTaskState::unknown)
            {
                if (is_complete())
                    return TaskState::complete;
                else
                    return TaskState::canceled;
            }
            return static_cast<TaskState>(state);
        }
 
    private:
        std::future<T> future_;
        TaskId id_{ 0 };
        std::weak_ptr<ThreadPool> pool_;
    };
 
 
    using TimePoint = std::chrono::time_point<std::chrono::system_clock>;
    using Duration = TimePoint::duration;
 
    constexpr static std::size_t default_capacity{ 200 };
 
    constexpr static std::size_t max_capacity{ 10'000'000 };
 
    constexpr static std::size_t max_threads{ 10'000 };
 
    ~ThreadPool();
 
    template <typename Function>
    TaskHandle<std::invoke_result_t<Function, ThreadPool&>>
    add_task(Function fct, TimePoint start_time = {}, std::string name = {})
    {
        static_assert(
            std::is_invocable<Function, ThreadPool&>::value
            || std::is_invocable<Function>::value,
            "Invalid function signature: Must be T fct() or T fct(ThreadPool&)");
 
        using Result = std::invoke_result_t<Function, ThreadPool&>;
 
        TaskHandle<Result> task_handle = [this, &fct, start_time, &name]()
            {
                std::lock_guard<std::mutex> lock(mutex_);
 
                if (is_full_i())
                {
                    throw std::runtime_error(cat(
                        "Cannot add task: Pending queue has reached capacity (",
                        pending_tasks_.size(), ')'));
                }
 
                using PackagedTask = std::packaged_task<Result(ThreadPool&)>;
 
                auto named_task_ptr = std::make_unique<NamedTaskImpl<PackagedTask>>(
                    PackagedTask{ std::move(fct) }, std::move(name));
 
                TaskHandle<Result> handle{
                    next_task_id_, named_task_ptr->fct_.get_future(), shared_from_this() };
 
                pending_tasks_.emplace_back(
                    next_task_id_, std::move(named_task_ptr), start_time);
 
                ++next_task_id_;
 
                return handle;
            }();
 
        cv_.notify_one();
 
        return task_handle;
    }
 
    template <typename Function,
        std::enable_if_t<std::is_invocable<Function>::value, bool> = true>
    TaskHandle<std::invoke_result_t<Function>>
    add_task(Function fct, TimePoint start_time = {}, std::string name = {})
    {
        return add_task(
            [f = std::move(fct)](ThreadPool&) mutable { return f(); },
            start_time, std::move(name));
    }
 
    template <typename Function,
        std::enable_if_t<std::is_invocable<Function, ThreadPool&>::value, bool> = true>
    TaskHandle<std::invoke_result_t<Function, ThreadPool&>>
    add_task(Function fct, Duration delay_before_start, std::string name = {})
    {
        return add_task(std::move(fct),
            std::chrono::system_clock::now() + delay_before_start, std::move(name));
    }
 
    template <typename Function,
        std::enable_if_t<std::is_invocable<Function>::value, bool> = true>
    TaskHandle<std::invoke_result_t<Function>>
    add_task(Function fct, Duration delay_before_start, std::string name = {})
    {
        return add_task(std::move(fct),
            std::chrono::system_clock::now() + delay_before_start, std::move(name));
    }
 
    template <typename Function,
        std::enable_if_t<std::is_invocable<Function, ThreadPool&>::value, bool> = true>
    TaskHandle<std::invoke_result_t<Function, ThreadPool&>>
    add_task(Function fct, std::string name)
    {
        return add_task(std::move(fct), TimePoint{}, std::move(name));
    }
 
    template <typename Function,
        std::enable_if_t<std::is_invocable<Function>::value, bool> = true>
    TaskHandle<std::invoke_result_t<Function>>
    add_task(Function fct, std::string name)
    {
        return add_task(std::move(fct), TimePoint{}, std::move(name));
    }
 
    GUL_EXPORT
    std::size_t cancel_pending_tasks();
 
    GUL_EXPORT
    std::size_t capacity() const noexcept { return capacity_; }
 
    GUL_EXPORT
    std::size_t count_pending() const;
 
    GUL_EXPORT
    std::size_t count_threads() const noexcept;
 
    GUL_EXPORT
    std::vector<std::string> get_pending_task_names() const;
 
    GUL_EXPORT
    std::vector<std::string> get_running_task_names() const;
 
    GUL_EXPORT
    ThreadId get_thread_id() const;
 
    GUL_EXPORT
    bool is_full() const noexcept;
 
    GUL_EXPORT
    bool is_idle() const;
 
    GUL_EXPORT
    bool is_shutdown_requested() const;
 
    GUL_EXPORT
    static std::shared_ptr<ThreadPool> make_shared(
        std::size_t num_threads, std::size_t capacity = default_capacity);
 
private:
    enum class InternalTaskState
    {
        pending = static_cast<int>(TaskState::pending),
        running = static_cast<int>(TaskState::running),
        unknown
    };
 
    struct NamedTask
    {
        NamedTask(std::string name)
        : name_{ std::move(name) }
        {}
 
        virtual ~NamedTask() = default;
        virtual void operator()(ThreadPool& pool) = 0;
 
        std::string name_;
    };
 
    template <typename FunctionType>
    struct NamedTaskImpl : public NamedTask
    {
    public:
        NamedTaskImpl(FunctionType fct, std::string name)
        : NamedTask{ std::move(name) }
        , fct_{ std::move(fct) }
        {}
 
        void operator()(ThreadPool& pool) override { fct_(pool); }
 
        FunctionType fct_;
    };
 
    struct Task
    {
        TaskId id_{};
        std::unique_ptr<NamedTask> named_task_;
        TimePoint start_time_{}; // When the task is to be started (at least no earlier)
 
        Task() = default;
 
        Task(TaskId task_id, std::unique_ptr<NamedTask> named_task, TimePoint start_time)
        : id_{ task_id }
        , named_task_{ std::move(named_task) }
        , start_time_{ start_time }
        {}
    };
 
    std::size_t capacity_{ 0 };
 
    std::vector<std::thread> threads_;
 
    thread_local static ThreadId thread_id_;
 
    std::condition_variable cv_;
 
    mutable std::mutex mutex_; // Protects the following variables
    std::vector<Task> pending_tasks_;
    std::vector<TaskId> running_task_ids_;
    std::vector<std::string> running_task_names_;
    TaskId next_task_id_ = 0;
    bool shutdown_requested_{ false };
 
 
    ThreadPool(std::size_t num_threads, std::size_t capacity);
 
    GUL_EXPORT
    bool cancel_pending_task(TaskId task_id);
 
    GUL_EXPORT
    InternalTaskState get_task_state(TaskId task_id) const;
 
    GUL_EXPORT
    bool is_full_i() const noexcept;
 
    void perform_work(std::size_t thread_index);
};
 
inline std::shared_ptr<ThreadPool> make_thread_pool(
    std::size_t num_threads, std::size_t capacity = ThreadPool::default_capacity)
{
    return ThreadPool::make_shared(num_threads, capacity);
}
 
 
} // namespace gul17
 
#endif // GUL17_THREADPOOL_H_