common_executor_adapter.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
 
#if __has_include(<kcenon/common/interfaces/executor_interface.h>)
#include <kcenon/common/interfaces/executor_interface.h>
#include <kcenon/common/patterns/result.h>
#elif __has_include(<common/interfaces/executor_interface.h>)
#include <common/interfaces/executor_interface.h>
#include <common/patterns/result.h>
#ifndef KCENON_COMMON_EXECUTOR_FALLBACK_DEFINED
#define KCENON_COMMON_EXECUTOR_FALLBACK_DEFINED
namespace kcenon {
namespace common {
using ::common::Result;
using ::common::VoidResult;
namespace interfaces {
using IExecutor = ::common::interfaces::IExecutor;
}
} // namespace common
} // namespace kcenon
#endif
#else
#error "Unable to locate common executor interface header."
#endif
 
#include <kcenon/thread/core/callback_job.h>
#include <kcenon/thread/core/error_handling.h>
#include <kcenon/thread/core/thread_pool.h>
 
#include <atomic>
#include <chrono>
#include <exception>
#include <functional>
#include <future>
#include <memory>
#include <optional>
#include <sstream>
#include <string>
#include <thread>
#include <utility>
 
namespace kcenon::thread::adapters {
 
namespace detail {
 
inline common::error_info make_error_info(int code, std::string message, std::string module = "thread_system") {
    return common::error_info{code, std::move(message), std::move(module)};
}
 
inline common::error_info make_error_info(kcenon::thread::error_code code, std::string message) {
    return common::error_info{
        static_cast<int>(code),
        std::move(message),
        "thread_system"
    };
}
 
inline std::exception_ptr to_exception(const common::error_info& info) {
    std::ostringstream ss;
    ss << "[" << info.module << "] " << info.message << " (code=" << info.code << ")";
    if (info.details) {
        ss << ": " << *info.details;
    }
    return std::make_exception_ptr(std::runtime_error(ss.str()));
}
 
inline common::VoidResult make_error(const common::error_info& info) {
    return common::VoidResult(info);
}
 
inline common::VoidResult make_error(kcenon::thread::error_code code, std::string message) {
    return common::VoidResult(make_error_info(code, std::move(message)));
}
 
inline common::error_info unexpected_pool_error() {
    return make_error_info(-1, "Thread pool unavailable");
}
 
inline common::VoidResult wrap_user_task(const std::function<void()>& task) {
    try {
        task();
        return common::ok();
    } catch (const std::exception& ex) {
        return make_error(kcenon::thread::error_code::job_execution_failed, ex.what());
    } catch (...) {
        return make_error(kcenon::thread::error_code::job_execution_failed,
                          "Unknown exception while executing task");
    }
}
 
inline std::optional<common::error_info> enqueue_job(
    const std::shared_ptr<kcenon::thread::thread_pool>& pool,
    const std::shared_ptr<std::promise<void>>& promise,
    std::function<common::VoidResult()> body) {
    if (!pool) {
        auto info = unexpected_pool_error();
        promise->set_exception(to_exception(info));
        return info;
    }
 
    auto completion_once = std::make_shared<std::once_flag>();
    // Capture promise by value so it stays alive until the lambda is
    // destroyed by the worker thread (after do_work() returns and the
    // callback_job is cleaned up).  The worker goes through several
    // steps between set_value()/set_exception() and lambda destruction,
    // which gives future::get()'s condition_variable::wait ample time
    // to complete – avoiding the data-race that early p.reset() caused.
    auto captured_promise = promise;
 
    auto job = std::make_unique<kcenon::thread::callback_job>(
        [captured_promise, completion_once, body = std::move(body)]() -> common::VoidResult {
            try {
                auto result = body();
                if (result.is_err()) {
                    auto info = result.error();
                    std::call_once(*completion_once, [&]() {
                        captured_promise->set_exception(to_exception(info));
                    });
                    return result;
                }
                std::call_once(*completion_once, [&]() {
                    captured_promise->set_value();
                });
                return result;
            } catch (const std::exception& ex) {
                auto info = make_error_info(
                    kcenon::thread::error_code::job_execution_failed,
                    ex.what()
                );
                std::call_once(*completion_once, [&]() {
                    captured_promise->set_exception(to_exception(info));
                });
                return make_error(info);
            } catch (...) {
                auto info = make_error_info(
                    kcenon::thread::error_code::job_execution_failed,
                    "Unhandled exception while executing job"
                );
                std::call_once(*completion_once, [&]() {
                    captured_promise->set_exception(to_exception(info));
                });
                return make_error(info);
            }
        });
 
    auto enqueue_result = pool->enqueue(std::move(job));
    if (enqueue_result.is_err()) {
        const auto& info = enqueue_result.error();
        std::call_once(*completion_once, [&]() {
            promise->set_exception(to_exception(info));
        });
        return info;
    }
 
    return std::nullopt;
}
 
inline common::Result<std::future<void>> schedule_task(
    const std::shared_ptr<kcenon::thread::thread_pool>& pool,
    std::function<common::VoidResult()> body) {
    auto promise = std::make_shared<std::promise<void>>();
    auto future = promise->get_future();
 
    if (auto error = enqueue_job(pool, promise, std::move(body))) {
        return common::Result<std::future<void>>(*error);
    }
 
    return common::Result<std::future<void>>::ok(std::move(future));
}
 
inline void schedule_task_async(
    std::shared_ptr<kcenon::thread::thread_pool> pool,
    std::shared_ptr<std::promise<void>> promise,
    std::function<common::VoidResult()> body,
    std::chrono::milliseconds delay) {
    // Use the thread pool itself to handle delayed execution
    // This avoids creating detached threads or extra std::async threads
    if (!pool) {
        promise->set_exception(to_exception(unexpected_pool_error()));
        return;
    }
 
    // Create once_flag to protect against race between delayed_job exception and enqueue failure
    auto completion_once = std::make_shared<std::once_flag>();
 
    // Capture promise by value (same approach as enqueue_job – no early reset)
    auto captured_promise = promise;
 
    // Create a wrapper job that handles delay and then enqueues the actual task
    auto delayed_job = std::make_unique<kcenon::thread::callback_job>(
        [pool, captured_promise, completion_once, body = std::move(body), delay]() -> common::VoidResult {
            try {
                if (delay.count() > 0) {
                    std::this_thread::sleep_for(delay);
                }
                // Enqueue the actual job after the delay
                // Note: enqueue_job has its own once_flag protection internally
                (void)enqueue_job(pool, captured_promise, std::move(body));
                return common::ok();
            } catch (...) {
                std::call_once(*completion_once, [&]() {
                    captured_promise->set_exception(std::current_exception());
                });
                return make_error(kcenon::thread::error_code::job_execution_failed,
                                  "Exception during delayed task scheduling");
            }
        });
 
    // Enqueue the delayed job to the pool
    auto enqueue_result = pool->enqueue(std::move(delayed_job));
    if (enqueue_result.is_err()) {
        std::call_once(*completion_once, [&]() {
            promise->set_exception(to_exception(enqueue_result.error()));
        });
    }
}
 
} // namespace detail
 
class thread_pool_executor_adapter : public common::interfaces::IExecutor {
public:
    explicit thread_pool_executor_adapter(std::shared_ptr<kcenon::thread::thread_pool> pool)
        : pool_(std::move(pool)) {}
 
    std::future<void> submit(std::function<void()> task) {
        auto result = detail::schedule_task(pool_, [task = std::move(task)]() mutable {
            return detail::wrap_user_task(task);
        });
 
        if (result.is_ok()) {
            return std::move(result.unwrap());
        }
 
        std::promise<void> failed;
        failed.set_exception(detail::to_exception(result.error()));
        return failed.get_future();
    }
 
    std::future<void> submit_delayed(std::function<void()> task,
                                     std::chrono::milliseconds delay) {
        auto promise = std::make_shared<std::promise<void>>();
        auto future = promise->get_future();
 
        detail::schedule_task_async(pool_, promise,
            [task = std::move(task)]() mutable {
                return detail::wrap_user_task(task);
            },
            delay);
 
        return future;
    }
 
    common::Result<std::future<void>> execute(std::unique_ptr<common::interfaces::IJob>&& job) override {
        auto shared_job = std::shared_ptr<common::interfaces::IJob>(std::move(job));
        return detail::schedule_task(pool_,
            [shared_job]() mutable -> common::VoidResult {
                try {
                    auto result = shared_job->execute();
                    if (result.is_err()) {
                        return detail::make_error(result.error());
                    }
                    return common::ok();
                } catch (const std::exception& ex) {
                    return detail::make_error(
                        kcenon::thread::error_code::job_execution_failed,
                        ex.what());
                } catch (...) {
                    return detail::make_error(
                        kcenon::thread::error_code::job_execution_failed,
                        "Unknown exception while executing common job");
                }
            });
    }
 
    common::Result<std::future<void>> execute_delayed(
        std::unique_ptr<common::interfaces::IJob>&& job,
        std::chrono::milliseconds delay) override {
        auto promise = std::make_shared<std::promise<void>>();
        auto future = promise->get_future();
        auto shared_job = std::shared_ptr<common::interfaces::IJob>(std::move(job));
 
        detail::schedule_task_async(pool_, promise,
            [shared_job]() mutable -> common::VoidResult {
                try {
                    auto result = shared_job->execute();
                    if (result.is_err()) {
                        return detail::make_error(result.error());
                    }
                    return common::ok();
                } catch (const std::exception& ex) {
                    return detail::make_error(
                        kcenon::thread::error_code::job_execution_failed,
                        ex.what());
                } catch (...) {
                    return detail::make_error(
                        kcenon::thread::error_code::job_execution_failed,
                        "Unknown exception while executing common job");
                }
            },
            delay);
 
        return common::Result<std::future<void>>::ok(std::move(future));
    }
 
    size_t worker_count() const override {
        return pool_ ? pool_->get_active_worker_count() : 0U;
    }
 
    bool is_running() const override {
        return pool_ && pool_->is_running();
    }
 
    size_t pending_tasks() const override {
        return pool_ ? pool_->get_pending_task_count() : 0U;
    }
 
    void shutdown(bool wait_for_completion = true) override {
        if (!pool_) {
            return;
        }
 
        auto stop_result = pool_->stop(!wait_for_completion);
        if (stop_result.is_err()) {
            // Best effort: surface error via exception to aid debugging.
            const auto& err = stop_result.error();
            throw std::runtime_error(err.message);
        }
    }
 
    std::shared_ptr<kcenon::thread::thread_pool> get_thread_pool() const {
        return pool_;
    }
 
private:
    std::shared_ptr<kcenon::thread::thread_pool> pool_;
};
 
class common_executor_factory {
public:
    static std::shared_ptr<common::interfaces::IExecutor> create_from_thread_pool(
        std::shared_ptr<kcenon::thread::thread_pool> pool) {
        return std::make_shared<thread_pool_executor_adapter>(std::move(pool));
    }
};
 
} // namespace kcenon::thread::adapters