kcenon::pacs::integration::thread_pool_executor_adapter Class Reference

IExecutor implementation using kcenon::thread::thread_pool. More...

#include <executor_adapter.h>

Inheritance diagram for kcenon::pacs::integration::thread_pool_executor_adapter:

Collaboration diagram for kcenon::pacs::integration::thread_pool_executor_adapter:

Classes
struct	delayed_task

Public Member Functions
	thread_pool_executor_adapter (std::shared_ptr< kcenon::thread::thread_pool > pool)
	Construct adapter with thread pool.
	thread_pool_executor_adapter (std::size_t worker_count)
	Construct adapter with worker count.
	~thread_pool_executor_adapter () override
	Destructor - ensures graceful shutdown.
	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
kcenon::common::Result< std::future< void > >	execute (std::unique_ptr< kcenon::common::interfaces::IJob > &&job) override
	Execute a job.
kcenon::common::Result< std::future< void > >	execute_delayed (std::unique_ptr< kcenon::common::interfaces::IJob > &&job, std::chrono::milliseconds delay) override
	Execute a job with delay.
std::size_t	worker_count () const override
	Get the number of worker threads.
bool	is_running () const override
	Check if the executor is running.
std::size_t	pending_tasks () const override
	Get the number of pending tasks.
void	shutdown (bool wait_for_completion=true) override
	Shutdown the executor.
template<typename F > requires std::invocable<F>
auto	submit (F &&func, std::string name="submitted_job") -> kcenon::common::Result< std::future< void > >
	Submit a void-returning callable directly.
auto	get_underlying_pool () const -> std::shared_ptr< kcenon::thread::thread_pool >
	Get the underlying thread pool.

Private Attributes
std::shared_ptr< kcenon::thread::thread_pool >	pool_
std::atomic< bool >	running_ {true}
std::atomic< std::size_t >	pending_count_ {0}
std::thread	delay_thread_
std::mutex	delay_mutex_
std::condition_variable	delay_cv_
std::atomic< bool >	shutdown_requested_ {false}
std::priority_queue< delayed_task, std::vector< delayed_task >, std::greater< delayed_task > >	delayed_tasks_

Detailed Description

IExecutor implementation using kcenon::thread::thread_pool.

This class adapts kcenon::thread::thread_pool to the IExecutor interface, enabling standardized task execution across the pacs_system workflow modules.

Thread Safety: All public methods are thread-safe.

Definition at line 144 of file executor_adapter.h.

Constructor & Destructor Documentation

thread_pool_executor_adapter() [1/4]

kcenon::pacs::integration::thread_pool_executor_adapter::thread_pool_executor_adapter ( std::shared_ptr< kcenon::thread::thread_pool > pool )

explicit

Construct adapter with thread pool.

Parameters

pool	Existing thread pool to use

Definition at line 30 of file executor_adapter.cpp.

    : pool_(std::move(pool)) {
    if (!pool_) {
        throw std::invalid_argument("Thread pool cannot be null");
    }
}

References pool_.

thread_pool_executor_adapter() [2/4]

kcenon::pacs::integration::thread_pool_executor_adapter::thread_pool_executor_adapter ( std::size_t worker_count )

explicit

Construct adapter with worker count.

Creates a new thread pool with the specified number of workers.

Parameters

worker_count

Number of worker threads

Definition at line 38 of file executor_adapter.cpp.

                                                                                 {
    kcenon::thread::thread_context context;
    pool_ = std::make_shared<kcenon::thread::thread_pool>("executor_pool", context);
 
    // Create and enqueue workers
    std::vector<std::unique_ptr<kcenon::thread::thread_worker>> workers;
    workers.reserve(worker_count);
 
    for (std::size_t i = 0; i < worker_count; ++i) {
        workers.push_back(std::make_unique<kcenon::thread::thread_worker>(false, context));
    }
 
    auto enqueue_result = pool_->enqueue_batch(std::move(workers));
    if (enqueue_result.is_err()) {
        throw std::runtime_error("Failed to enqueue workers to thread pool");
    }
 
    auto start_result = pool_->start();
    if (start_result.is_err()) {
        throw std::runtime_error("Failed to start thread pool");
    }
}

References pool_, and worker_count().

Here is the call graph for this function:

~thread_pool_executor_adapter()

kcenon::pacs::integration::thread_pool_executor_adapter::~thread_pool_executor_adapter ( )

override

Destructor - ensures graceful shutdown.

Definition at line 61 of file executor_adapter.cpp.

                                                            {
    shutdown(true);
}

References shutdown().

Here is the call graph for this function:

thread_pool_executor_adapter() [3/4]

kcenon::pacs::integration::thread_pool_executor_adapter::thread_pool_executor_adapter ( const thread_pool_executor_adapter & )

delete

thread_pool_executor_adapter() [4/4]

kcenon::pacs::integration::thread_pool_executor_adapter::thread_pool_executor_adapter ( thread_pool_executor_adapter && )

delete

Member Function Documentation

execute()

kcenon::common::Result< std::future< void > > kcenon::pacs::integration::thread_pool_executor_adapter::execute ( std::unique_ptr< kcenon::common::interfaces::IJob > && job )

nodiscardoverride

Execute a job.

Submits the job to the thread pool for asynchronous execution.

Parameters

job	The job to execute

Returns

Result containing future or error

Definition at line 65 of file executor_adapter.cpp.

                                                     {
 
    if (!running_.load()) {
        return kcenon::common::Result<std::future<void>>(
            kcenon::common::error_info{-1, "Executor is not running", "executor"});
    }
 
    if (!job) {
        return kcenon::common::Result<std::future<void>>(
            kcenon::common::error_info{-2, "Job cannot be null", "executor"});
    }
 
    auto promise = std::make_shared<std::promise<void>>();
    auto future = promise->get_future();
 
    // Capture job by shared_ptr for thread safety
    auto shared_job = std::shared_ptr<kcenon::common::interfaces::IJob>(std::move(job));
 
    ++pending_count_;
 
    try {
        auto job_obj = kcenon::thread::job_builder()
            .name("executor_task")
            .work([this, shared_job, promise]() -> kcenon::common::VoidResult {
                try {
                    auto result = shared_job->execute();
                    if (result.is_ok()) {
                        promise->set_value();
                    } else {
                        promise->set_exception(std::make_exception_ptr(
                            std::runtime_error(result.error().message)));
                    }
                } catch (...) {
                    promise->set_exception(std::current_exception());
                }
                --pending_count_;
                return kcenon::common::ok();
            })
            .build();
 
        auto enqueue_result = pool_->enqueue(std::move(job_obj));
        if (enqueue_result.is_err()) {
            --pending_count_;
            return kcenon::common::Result<std::future<void>>(
                kcenon::common::error_info{-3, "Failed to enqueue task to thread pool", "executor"});
        }
    } catch (const std::exception&) {
        --pending_count_;
        return kcenon::common::Result<std::future<void>>(
            kcenon::common::error_info{-3, "Failed to enqueue task to thread pool", "executor"});
    }
 
    return kcenon::common::Result<std::future<void>>::ok(std::move(future));
}

References pending_count_, pool_, and running_.

execute_delayed()

kcenon::common::Result< std::future< void > > kcenon::pacs::integration::thread_pool_executor_adapter::execute_delayed ( std::unique_ptr< kcenon::common::interfaces::IJob > && job, std::chrono::milliseconds delay )

nodiscardoverride

Execute a job with delay.

Parameters

job	The job to execute
delay	The delay before execution

Returns

Result containing future or error

Definition at line 121 of file executor_adapter.cpp.

                                 {
 
    if (!running_.load()) {
        return kcenon::common::Result<std::future<void>>(
            kcenon::common::error_info{-1, "Executor is not running", "executor"});
    }
 
    if (!job) {
        return kcenon::common::Result<std::future<void>>(
            kcenon::common::error_info{-2, "Job cannot be null", "executor"});
    }
 
    auto promise = std::make_shared<std::promise<void>>();
    auto future = promise->get_future();
 
    // Capture job and schedule for later execution
    auto shared_job = std::shared_ptr<kcenon::common::interfaces::IJob>(std::move(job));
    auto execute_at = std::chrono::steady_clock::now() + delay;
 
    ++pending_count_;
 
    {
        std::lock_guard<std::mutex> lock(delay_mutex_);
        delayed_tasks_.push({execute_at,
            [this, shared_job, promise]() mutable {
                try {
                    auto result = shared_job->execute();
                    if (result.is_ok()) {
                        promise->set_value();
                    } else {
                        promise->set_exception(std::make_exception_ptr(
                            std::runtime_error(result.error().message)));
                    }
                } catch (...) {
                    promise->set_exception(std::current_exception());
                }
                --pending_count_;
            }});
    }
 
    // Start delay thread if not already running
    if (!delay_thread_.joinable()) {
        delay_thread_ = std::thread([this]() {
            while (!shutdown_requested_.load()) {
                std::function<void()> task;
 
                {
                    std::unique_lock<std::mutex> lock(delay_mutex_);
 
                    if (delayed_tasks_.empty()) {
                        delay_cv_.wait_for(lock, std::chrono::milliseconds(100));
                        continue;
                    }
 
                    auto now = std::chrono::steady_clock::now();
                    const auto& top_task = delayed_tasks_.top();
 
                    if (top_task.execute_at > now) {
                        delay_cv_.wait_until(lock, top_task.execute_at);
                        continue;
                    }
 
                    task = top_task.task;
                    delayed_tasks_.pop();
                }
 
                if (task && pool_ && pool_->is_running()) {
                    try {
                        auto job_obj = kcenon::thread::job_builder()
                            .name("delayed_task")
                            .work([task = std::move(task)]() -> kcenon::common::VoidResult {
                                task();
                                return kcenon::common::ok();
                            })
                            .build();
 
                        (void)pool_->enqueue(std::move(job_obj));
                    } catch (...) {
                        // Ignore enqueue failures in delayed task scheduler
                    }
                }
            }
        });
    }
 
    delay_cv_.notify_one();
 
    return kcenon::common::Result<std::future<void>>::ok(std::move(future));
}

References delay_cv_, delay_mutex_, delay_thread_, delayed_tasks_, pending_count_, pool_, running_, and shutdown_requested_.

get_underlying_pool()

std::shared_ptr< kcenon::thread::thread_pool > kcenon::pacs::integration::thread_pool_executor_adapter::get_underlying_pool ( ) const -> std::shared_ptr<kcenon::thread::thread_pool>

nodiscard

Get the underlying thread pool.

Definition at line 242 of file executor_adapter.cpp.

                                                        {
    return pool_;
}

References pool_.

is_running()

bool kcenon::pacs::integration::thread_pool_executor_adapter::is_running ( ) const

nodiscardoverride

Check if the executor is running.

Definition at line 217 of file executor_adapter.cpp.

                                                    {
    return running_.load() && pool_ && pool_->is_running();
}

References pool_, and running_.

operator=() [1/2]

thread_pool_executor_adapter & kcenon::pacs::integration::thread_pool_executor_adapter::operator= ( const thread_pool_executor_adapter & )

delete

operator=() [2/2]

thread_pool_executor_adapter & kcenon::pacs::integration::thread_pool_executor_adapter::operator= ( thread_pool_executor_adapter && )

delete

pending_tasks()

std::size_t kcenon::pacs::integration::thread_pool_executor_adapter::pending_tasks ( ) const

nodiscardoverride

Get the number of pending tasks.

Definition at line 221 of file executor_adapter.cpp.

                                                            {
    std::size_t pool_pending = pool_ ? pool_->get_pending_task_count() : 0;
    return pool_pending + pending_count_.load();
}

References pending_count_, and pool_.

shutdown()

void kcenon::pacs::integration::thread_pool_executor_adapter::shutdown ( bool wait_for_completion = true )

override

Shutdown the executor.

Parameters

wait_for_completion

Wait for all pending tasks to complete

Definition at line 226 of file executor_adapter.cpp.

                                                                    {
    running_.store(false);
    shutdown_requested_.store(true);
 
    delay_cv_.notify_all();
 
    if (delay_thread_.joinable()) {
        delay_thread_.join();
    }
 
    if (pool_) {
        pool_->stop(!wait_for_completion);
    }
}

References delay_cv_, delay_thread_, pool_, running_, and shutdown_requested_.

Referenced by ~thread_pool_executor_adapter().

Here is the caller graph for this function:

submit()

template<typename F >
requires std::invocable<F>

auto kcenon::pacs::integration::thread_pool_executor_adapter::submit ( F && func, std::string name = "submitted_job" ) -> kcenon::common::Result<std::future<void>>

inlinenodiscard

Submit a void-returning callable directly.

This is a convenience method that wraps the callable in a lambda_job.

Template Parameters

F	Callable type

Parameters

func	The function to execute
name	Job name for logging

Returns

Result containing future or error

Definition at line 238 of file executor_adapter.h.

                                               {
        auto job = std::make_unique<lambda_job>(std::forward<F>(func), std::move(name));
        return execute(std::move(job));
    }

References name.

worker_count()

std::size_t kcenon::pacs::integration::thread_pool_executor_adapter::worker_count ( ) const

nodiscardoverride

Get the number of worker threads.

Definition at line 213 of file executor_adapter.cpp.

                                                           {
    return pool_ ? pool_->get_active_worker_count() : 0;
}

References pool_.

Referenced by thread_pool_executor_adapter().

Here is the caller graph for this function:

Member Data Documentation

delay_cv_

std::condition_variable kcenon::pacs::integration::thread_pool_executor_adapter::delay_cv_

private

Definition at line 258 of file executor_adapter.h.

Referenced by execute_delayed(), and shutdown().

delay_mutex_

std::mutex kcenon::pacs::integration::thread_pool_executor_adapter::delay_mutex_

private

Definition at line 257 of file executor_adapter.h.

Referenced by execute_delayed().

delay_thread_

std::thread kcenon::pacs::integration::thread_pool_executor_adapter::delay_thread_

private

Definition at line 256 of file executor_adapter.h.

Referenced by execute_delayed(), and shutdown().

delayed_tasks_

std::priority_queue<delayed_task, std::vector<delayed_task>, std::greater<delayed_task> > kcenon::pacs::integration::thread_pool_executor_adapter::delayed_tasks_

private

Definition at line 270 of file executor_adapter.h.

Referenced by execute_delayed().

pending_count_

std::atomic<std::size_t> kcenon::pacs::integration::thread_pool_executor_adapter::pending_count_ {0}

private

Definition at line 253 of file executor_adapter.h.

{0};

Referenced by execute(), execute_delayed(), and pending_tasks().

pool_

std::shared_ptr<kcenon::thread::thread_pool> kcenon::pacs::integration::thread_pool_executor_adapter::pool_

private

Definition at line 251 of file executor_adapter.h.

Referenced by execute(), execute_delayed(), get_underlying_pool(), is_running(), pending_tasks(), shutdown(), thread_pool_executor_adapter(), thread_pool_executor_adapter(), and worker_count().

running_

std::atomic<bool> kcenon::pacs::integration::thread_pool_executor_adapter::running_ {true}

private

Definition at line 252 of file executor_adapter.h.

{true};

Referenced by execute(), execute_delayed(), is_running(), and shutdown().

shutdown_requested_

std::atomic<bool> kcenon::pacs::integration::thread_pool_executor_adapter::shutdown_requested_ {false}

private

Definition at line 259 of file executor_adapter.h.

{false};

Referenced by execute_delayed(), and shutdown().

---

The documentation for this class was generated from the following files:

• include/kcenon/pacs/integration/executor_adapter.h
• src/integration/executor_adapter.cpp