kcenon::thread::adapters::detail Namespace Reference

Functions
common::error_info	make_error_info (int code, std::string message, std::string module="thread_system")
common::error_info	make_error_info (kcenon::thread::error_code code, std::string message)
std::exception_ptr	to_exception (const common::error_info &info)
common::VoidResult	make_error (const common::error_info &info)
common::VoidResult	make_error (kcenon::thread::error_code code, std::string message)
common::error_info	unexpected_pool_error ()
common::VoidResult	wrap_user_task (const std::function< void()> &task)
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)
common::Result< std::future< void > >	schedule_task (const std::shared_ptr< kcenon::thread::thread_pool > &pool, std::function< common::VoidResult()> body)
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)

Function Documentation

enqueue_job()

std::optional< common::error_info > kcenon::thread::adapters::detail::enqueue_job ( const std::shared_ptr< kcenon::thread::thread_pool > & pool, const std::shared_ptr< std::promise< void > > & promise, std::function< common::VoidResult()> body )

inline

Definition at line 126 of file common_executor_adapter.h.

                                          {
    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;
}

References kcenon::thread::info, kcenon::thread::job_execution_failed, make_error(), make_error_info(), to_exception(), and unexpected_pool_error().

Referenced by schedule_task(), and schedule_task_async().

Here is the call graph for this function:

Here is the caller graph for this function:

make_error() [1/2]

common::VoidResult kcenon::thread::adapters::detail::make_error ( const common::error_info & info )

inline

Definition at line 102 of file common_executor_adapter.h.

                                                                 {
    return common::VoidResult(info);
}

References kcenon::thread::info.

Referenced by enqueue_job(), kcenon::thread::adapters::thread_pool_executor_adapter::execute(), kcenon::thread::adapters::thread_pool_executor_adapter::execute_delayed(), schedule_task_async(), and wrap_user_task().

Here is the caller graph for this function:

make_error() [2/2]

common::VoidResult kcenon::thread::adapters::detail::make_error ( kcenon::thread::error_code code, std::string message )

inline

Definition at line 106 of file common_executor_adapter.h.

                                                                                   {
    return common::VoidResult(make_error_info(code, std::move(message)));
}

References make_error_info().

Here is the call graph for this function:

make_error_info() [1/2]

common::error_info kcenon::thread::adapters::detail::make_error_info ( int code, std::string message, std::string module = "thread_system" )

inline

Definition at line 81 of file common_executor_adapter.h.

                                                                                                         {
    return common::error_info{code, std::move(message), std::move(module)};
}

Referenced by enqueue_job(), make_error(), and unexpected_pool_error().

Here is the caller graph for this function:

make_error_info() [2/2]

common::error_info kcenon::thread::adapters::detail::make_error_info ( kcenon::thread::error_code code, std::string message )

inline

Definition at line 85 of file common_executor_adapter.h.

                                                                                        {
    return common::error_info{
        static_cast<int>(code),
        std::move(message),
        "thread_system"
    };
}

schedule_task()

common::Result< std::future< void > > kcenon::thread::adapters::detail::schedule_task ( const std::shared_ptr< kcenon::thread::thread_pool > & pool, std::function< common::VoidResult()> body )

inline

Definition at line 193 of file common_executor_adapter.h.

                                          {
    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));
}

References enqueue_job().

Referenced by kcenon::thread::adapters::thread_pool_executor_adapter::execute(), and kcenon::thread::adapters::thread_pool_executor_adapter::submit().

Here is the call graph for this function:

Here is the caller graph for this function:

schedule_task_async()

void kcenon::thread::adapters::detail::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 )

inline

Definition at line 206 of file common_executor_adapter.h.

                                 {
    // 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()));
        });
    }
}

References kcenon::thread::delay, enqueue_job(), kcenon::thread::job_execution_failed, make_error(), to_exception(), and unexpected_pool_error().

Referenced by kcenon::thread::adapters::thread_pool_executor_adapter::execute_delayed(), and kcenon::thread::adapters::thread_pool_executor_adapter::submit_delayed().

Here is the call graph for this function:

Here is the caller graph for this function:

to_exception()

std::exception_ptr kcenon::thread::adapters::detail::to_exception ( const common::error_info & info )

inline

Definition at line 93 of file common_executor_adapter.h.

                                                                   {
    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()));
}

References kcenon::thread::info.

Referenced by enqueue_job(), schedule_task_async(), and kcenon::thread::adapters::thread_pool_executor_adapter::submit().

Here is the caller graph for this function:

unexpected_pool_error()

common::error_info kcenon::thread::adapters::detail::unexpected_pool_error ( )

inline

Definition at line 110 of file common_executor_adapter.h.

                                                {
    return make_error_info(-1, "Thread pool unavailable");
}

References make_error_info().

Referenced by enqueue_job(), and schedule_task_async().

Here is the call graph for this function:

Here is the caller graph for this function:

wrap_user_task()

common::VoidResult kcenon::thread::adapters::detail::wrap_user_task ( const std::function< void()> & task )

inline

Definition at line 114 of file common_executor_adapter.h.

                                                                        {
    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");
    }
}

References kcenon::thread::job_execution_failed, and make_error().

Referenced by kcenon::thread::adapters::thread_pool_executor_adapter::submit(), and kcenon::thread::adapters::thread_pool_executor_adapter::submit_delayed().

Here is the call graph for this function:

Here is the caller graph for this function: