thread_pool_executor.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include "task.h"
 
#include <atomic>
#include <coroutine>
#include <exception>
#include <functional>
#include <future>
#include <memory>
#include <optional>
#include <thread>
#include <type_traits>
 
#ifdef KCENON_HAS_COMMON_SYSTEM
#include <kcenon/common/interfaces/executor_interface.h>
#include <kcenon/common/interfaces/thread_pool_interface.h>
#endif
 
namespace kcenon::container::async
{
#ifdef KCENON_HAS_COMMON_SYSTEM
    using executor_ptr = std::shared_ptr<kcenon::common::interfaces::IExecutor>;
#else
    using executor_ptr = std::nullptr_t;
#endif
 
    namespace detail
    {
        template<typename T>
        struct executor_state
        {
            std::function<T()> work_;
            std::optional<T> result_;
            std::exception_ptr exception_;
            std::atomic<bool> ready_{false};
            std::coroutine_handle<> continuation_;
 
            explicit executor_state(std::function<T()> work)
                : work_(std::move(work)) {}
 
            executor_state(const executor_state&) = delete;
            executor_state& operator=(const executor_state&) = delete;
            executor_state(executor_state&&) = delete;
            executor_state& operator=(executor_state&&) = delete;
        };
 
        template<typename T>
        struct executor_awaitable
        {
            std::shared_ptr<executor_state<T>> state_;
            executor_ptr executor_;
 
            executor_awaitable(std::function<T()> work, executor_ptr executor)
                : state_(std::make_shared<executor_state<T>>(std::move(work)))
                , executor_(std::move(executor))
            {
            }
 
            executor_awaitable(executor_awaitable&&) noexcept = default;
            executor_awaitable& operator=(executor_awaitable&&) noexcept = default;
 
            executor_awaitable(const executor_awaitable&) = delete;
            executor_awaitable& operator=(const executor_awaitable&) = delete;
 
            [[nodiscard]] bool await_ready() const noexcept
            {
                return false;
            }
 
            void await_suspend(std::coroutine_handle<> handle)
            {
                auto state = state_;
                state->continuation_ = handle;
 
#ifdef KCENON_HAS_COMMON_SYSTEM
                if (executor_ && executor_->is_running())
                {
                    // Create a job that executes the work
                    class work_job : public kcenon::common::interfaces::IJob
                    {
                    public:
                        work_job(std::shared_ptr<executor_state<T>> s)
                            : state_(std::move(s)) {}
 
                        kcenon::common::VoidResult execute() override
                        {
                            try {
                                state_->result_.emplace(state_->work_());
                            } catch (...) {
                                state_->exception_ = std::current_exception();
                            }
                            state_->ready_.store(true, std::memory_order_release);
                            if (state_->continuation_) {
                                state_->continuation_.resume();
                            }
                            return kcenon::common::ok();
                        }
 
                        std::string get_name() const override {
                            return "async_container_work";
                        }
 
                    private:
                        std::shared_ptr<executor_state<T>> state_;
                    };
 
                    auto job = std::make_unique<work_job>(state);
                    auto result = executor_->execute(std::move(job));
                    if (!result.is_ok()) {
                        // Fallback to thread if executor fails
                        run_in_thread(state, handle);
                    }
                    return;
                }
#endif
                // Fallback: run in a detached thread
                run_in_thread(state, handle);
            }
 
            T await_resume()
            {
                while (!state_->ready_.load(std::memory_order_acquire)) {
                    // Spin until ready
                }
                if (state_->exception_) {
                    std::rethrow_exception(state_->exception_);
                }
                return std::move(*state_->result_);
            }
 
        private:
            static void run_in_thread(
                std::shared_ptr<executor_state<T>> state,
                std::coroutine_handle<> handle)
            {
                std::thread([state, handle]() mutable {
                    try {
                        state->result_.emplace(state->work_());
                    } catch (...) {
                        state->exception_ = std::current_exception();
                    }
                    state->ready_.store(true, std::memory_order_release);
                    handle.resume();
                }).detach();
            }
        };
 
        template<typename F>
        auto make_executor_awaitable(F&& func, executor_ptr executor)
            -> executor_awaitable<std::invoke_result_t<F>>
        {
            using result_type = std::invoke_result_t<F>;
            return executor_awaitable<result_type>(
                std::forward<F>(func), std::move(executor));
        }
 
    } // namespace detail
 
    class async_executor_context
    {
    public:
        static async_executor_context& instance()
        {
            static async_executor_context instance;
            return instance;
        }
 
        void set_executor(executor_ptr executor)
        {
            executor_ = std::move(executor);
        }
 
        [[nodiscard]] executor_ptr get_executor() const
        {
            return executor_;
        }
 
        [[nodiscard]] bool has_executor() const
        {
#ifdef KCENON_HAS_COMMON_SYSTEM
            return executor_ != nullptr;
#else
            return false;
#endif
        }
 
        void clear_executor()
        {
#ifdef KCENON_HAS_COMMON_SYSTEM
            executor_ = nullptr;
#endif
        }
 
    private:
        async_executor_context() = default;
        executor_ptr executor_{};
    };
 
    class executor_context_guard
    {
    public:
        explicit executor_context_guard(executor_ptr executor)
            : previous_(async_executor_context::instance().get_executor())
        {
            async_executor_context::instance().set_executor(std::move(executor));
        }
 
        ~executor_context_guard()
        {
            async_executor_context::instance().set_executor(std::move(previous_));
        }
 
        executor_context_guard(const executor_context_guard&) = delete;
        executor_context_guard& operator=(const executor_context_guard&) = delete;
 
    private:
        executor_ptr previous_;
    };
 
} // namespace kcenon::container::async