executor_adapter.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <kcenon/common/interfaces/executor_interface.h>
#include <kcenon/common/patterns/result.h>
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
 
namespace kcenon::thread {
class thread_pool;
}  // namespace kcenon::thread
 
namespace kcenon::pacs::integration {
 
// Forward declaration
class thread_pool_interface;
 
class lambda_job : public kcenon::common::interfaces::IJob {
public:
    using job_function = std::function<kcenon::common::VoidResult()>;
 
    explicit lambda_job(
        job_function func,
        std::string name = "lambda_job",
        int priority = 0)
        : func_(std::move(func))
        , name_(std::move(name))
        , priority_(priority) {}
 
    template <typename F>
        requires std::is_void_v<std::invoke_result_t<F>>
    explicit lambda_job(F&& func, std::string name = "lambda_job", int priority = 0)
        : func_([f = std::forward<F>(func)]() -> kcenon::common::VoidResult {
              f();
              return kcenon::common::VoidResult(std::monostate{});
          })
        , name_(std::move(name))
        , priority_(priority) {}
 
    kcenon::common::VoidResult execute() override {
        if (!func_) {
            return kcenon::common::VoidResult(
                kcenon::common::error_info{-1, "No function provided", "executor"});
        }
        return func_();
    }
 
    [[nodiscard]] std::string get_name() const override { return name_; }
    [[nodiscard]] int get_priority() const override { return priority_; }
 
private:
    job_function func_;
    std::string name_;
    int priority_;
};
 
class thread_pool_executor_adapter : public kcenon::common::interfaces::IExecutor {
public:
    explicit thread_pool_executor_adapter(
        std::shared_ptr<kcenon::thread::thread_pool> pool);
 
    explicit thread_pool_executor_adapter(std::size_t worker_count);
 
    ~thread_pool_executor_adapter() override;
 
    // Non-copyable, non-movable
    thread_pool_executor_adapter(const thread_pool_executor_adapter&) = delete;
    thread_pool_executor_adapter& operator=(const thread_pool_executor_adapter&) = delete;
    thread_pool_executor_adapter(thread_pool_executor_adapter&&) = delete;
    thread_pool_executor_adapter& operator=(thread_pool_executor_adapter&&) = delete;
 
    // =========================================================================
    // IExecutor Implementation
    // =========================================================================
 
    [[nodiscard]] kcenon::common::Result<std::future<void>> execute(
        std::unique_ptr<kcenon::common::interfaces::IJob>&& job) override;
 
    [[nodiscard]] kcenon::common::Result<std::future<void>> execute_delayed(
        std::unique_ptr<kcenon::common::interfaces::IJob>&& job,
        std::chrono::milliseconds delay) override;
 
    [[nodiscard]] std::size_t worker_count() const override;
 
    [[nodiscard]] bool is_running() const override;
 
    [[nodiscard]] std::size_t pending_tasks() const override;
 
    void shutdown(bool wait_for_completion = true) override;
 
    // =========================================================================
    // Convenience Methods
    // =========================================================================
 
    template <typename F>
        requires std::invocable<F>
    [[nodiscard]] auto submit(F&& func, std::string name = "submitted_job")
        -> kcenon::common::Result<std::future<void>> {
        auto job = std::make_unique<lambda_job>(std::forward<F>(func), std::move(name));
        return execute(std::move(job));
    }
 
    [[nodiscard]] auto get_underlying_pool() const
        -> std::shared_ptr<kcenon::thread::thread_pool>;
 
private:
    std::shared_ptr<kcenon::thread::thread_pool> pool_;
    std::atomic<bool> running_{true};
    std::atomic<std::size_t> pending_count_{0};
 
    // For delayed execution
    std::thread delay_thread_;
    std::mutex delay_mutex_;
    std::condition_variable delay_cv_;
    std::atomic<bool> shutdown_requested_{false};
 
    struct delayed_task {
        std::chrono::steady_clock::time_point execute_at;
        std::function<void()> task;
 
        bool operator>(const delayed_task& other) const {
            return execute_at > other.execute_at;
        }
    };
    std::priority_queue<delayed_task, std::vector<delayed_task>,
                        std::greater<delayed_task>> delayed_tasks_;
};
 
[[nodiscard]] std::shared_ptr<kcenon::common::interfaces::IExecutor>
make_executor(std::shared_ptr<thread_pool_interface> pool_interface);
 
}  // namespace kcenon::pacs::integration