Awaitable that runs work using an executor or fallback thread. More...

#include <thread_pool_executor.h>

Collaboration diagram for kcenon::container::async::detail::executor_awaitable< T >:

Public Member Functions
	executor_awaitable (std::function< T()> work, executor_ptr 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

bool	await_ready () const noexcept

void	await_suspend (std::coroutine_handle<> handle)

T	await_resume ()

Public Attributes
std::shared_ptr< executor_state< T > >	state_

executor_ptr	executor_

Static Private Member Functions
static void	run_in_thread (std::shared_ptr< executor_state< T > > state, std::coroutine_handle<> handle)

Detailed Description

template<typename T>
struct kcenon::container::async::detail::executor_awaitable< T >

Awaitable that runs work using an executor or fallback thread.

When an executor is provided, submits work to the executor. Otherwise, falls back to std::thread-based execution.

Definition at line 91 of file thread_pool_executor.h.

Constructor & Destructor Documentation

◆ executor_awaitable() [1/3]

template<typename T >

kcenon::container::async::detail::executor_awaitable< T >::executor_awaitable	(	std::function< T()>	work,
		executor_ptr	executor )

inline

Definition at line 96 of file thread_pool_executor.h.

                : state_(std::make_shared<executor_state<T>>(std::move(work)))
                , executor_(std::move(executor))
            {
            }

◆ executor_awaitable() [2/3]

template<typename T >

kcenon::container::async::detail::executor_awaitable< T >::executor_awaitable ( executor_awaitable< T > && )

defaultnoexcept

◆ executor_awaitable() [3/3]

template<typename T >

kcenon::container::async::detail::executor_awaitable< T >::executor_awaitable ( const executor_awaitable< T > & )

delete

Member Function Documentation

◆ await_ready()

template<typename T >

bool kcenon::container::async::detail::executor_awaitable< T >::await_ready ( ) const

inlinenodiscardnoexcept

Definition at line 108 of file thread_pool_executor.h.

            {
                return false;
            }

◆ await_resume()

template<typename T >

T kcenon::container::async::detail::executor_awaitable< T >::await_resume ( )

inline

Definition at line 163 of file thread_pool_executor.h.

            {
                while (!state_->ready_.load(std::memory_order_acquire)) {
                    // Spin until ready
                }
                if (state_->exception_) {
                    std::rethrow_exception(state_->exception_);
                }
                return std::move(*state_->result_);
            }

References kcenon::container::async::detail::executor_awaitable< T >::state_.

◆ await_suspend()

template<typename T >

void kcenon::container::async::detail::executor_awaitable< T >::await_suspend ( std::coroutine_handle<> handle )

inline

Definition at line 113 of file thread_pool_executor.h.

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

References kcenon::container::async::detail::executor_awaitable< T >::executor_, kcenon::container::async::detail::executor_awaitable< T >::run_in_thread(), and kcenon::container::async::detail::executor_awaitable< T >::state_.

Here is the call graph for this function:

◆ operator=() [1/2]

template<typename T >

executor_awaitable & kcenon::container::async::detail::executor_awaitable< T >::operator= ( const executor_awaitable< T > & )

delete

◆ operator=() [2/2]

template<typename T >

executor_awaitable & kcenon::container::async::detail::executor_awaitable< T >::operator= ( executor_awaitable< T > && )

defaultnoexcept

◆ run_in_thread()

template<typename T >

static void kcenon::container::async::detail::executor_awaitable< T >::run_in_thread	(	std::shared_ptr< executor_state< T > >	state,
		std::coroutine_handle<>	handle )

inlinestaticprivate

Definition at line 175 of file thread_pool_executor.h.

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