High-performance asynchronous executor using thread_system. More...

#include <async_operations.h>

Collaboration diagram for database::async::async_executor:

Public Member Functions
	async_executor (size_t thread_count=std::thread::hardware_concurrency(), const thread_context_type &=thread_context_type())
	Constructs an async executor with specified thread count.

	~async_executor ()

	async_executor (const async_executor &)=delete

async_executor &	operator= (const async_executor &)=delete

	async_executor (async_executor &&)=delete

async_executor &	operator= (async_executor &&)=delete

template<typename F , typename... Args> requires concepts::SubmittableTask<F, Args...>
auto	submit (F &&func, Args &&... args) -> std::future< std::invoke_result_t< F, Args... > >
	Submits a task for asynchronous execution.

void	shutdown ()
	Gracefully shuts down the executor.

void	wait_for_completion ()
	Waits for all pending tasks to complete.

size_t	pending_tasks () const
	Returns the number of pending tasks.

size_t	thread_count () const
	Returns the number of worker threads.

constexpr bool	is_using_thread_system () const
	Checks if using thread_system implementation.

Private Member Functions
void	worker_thread ()

Private Attributes
std::vector< std::thread >	workers_

std::queue< std::function< void()> >	tasks_

std::mutex	queue_mutex_

std::condition_variable	condition_

std::atomic< bool >	stop_

size_t	thread_count_

Detailed Description

High-performance asynchronous executor using thread_system.

This executor leverages thread_system's advanced features when available:

Adaptive job queue (mutex ↔ lock-free automatic switching)
Sub-microsecond latency (77ns job scheduling)
1.16M+ jobs/second throughput
Integrated monitoring and logging

When USE_THREAD_SYSTEM is not defined, falls back to std::thread implementation.

Thread Safety

All methods are thread-safe and can be called from multiple threads.

Performance

Throughput: 1.16M+ jobs/s (vs ~50K with std::thread)
Latency: 77ns scheduling (vs 2-5μs with std::thread)
Scalability: Linear scaling up to hardware concurrency

Examples: async_executor_demo.cpp, and async_operations_demo.cpp.

Definition at line 328 of file async_operations.h.

Constructor & Destructor Documentation

◆ async_executor() [1/3]

database::async::async_executor::async_executor	(	size_t	thread_count = std::thread::hardware_concurrency(),
		const thread_context_type &	= thread_context_type() )

inlineexplicit

Constructs an async executor with specified thread count.

Parameters

thread_count	Number of worker threads (defaults to hardware concurrency)
context	Thread context for logging/monitoring (optional)

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 362 of file async_operations.h.

            : thread_count_(thread_count)
            , stop_(false)
        {
            workers_.reserve(thread_count_);
            for (size_t i = 0; i < thread_count_; ++i) {
                workers_.emplace_back([this] { worker_thread(); });
            }
        }

References thread_count_, worker_thread(), and workers_.

Here is the call graph for this function:

◆ ~async_executor()

database::async::async_executor::~async_executor ( )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 375 of file async_operations.h.

                          {
            shutdown();
        }

References shutdown().

Here is the call graph for this function:

◆ async_executor() [2/3]

database::async::async_executor::async_executor ( const async_executor & )

delete

◆ async_executor() [3/3]

database::async::async_executor::async_executor ( async_executor && )

delete

Member Function Documentation

◆ is_using_thread_system()

bool database::async::async_executor::is_using_thread_system ( ) const

inlineconstexpr

Checks if using thread_system implementation.

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h, and async_executor_demo.cpp.

Definition at line 510 of file async_operations.h.

                                                      {
            return using_thread_system;
        }

References database::async::using_thread_system.

Referenced by demonstrate_basic_usage(), and demonstrate_high_throughput().

Here is the caller graph for this function:

◆ operator=() [1/2]

async_executor & database::async::async_executor::operator= ( async_executor && )

delete

◆ operator=() [2/2]

async_executor & database::async::async_executor::operator= ( const async_executor & )

delete

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

◆ pending_tasks()

size_t database::async::async_executor::pending_tasks ( ) const

inline

Returns the number of pending tasks.

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 488 of file async_operations.h.

                                     {
#ifdef USE_THREAD_SYSTEM
            if (pool_) {
                return pool_->get_job_queue()->size();
            }
            return 0;
#else
            std::unique_lock<std::mutex> lock(queue_mutex_);
            return tasks_.size();
#endif
        }

References queue_mutex_, and tasks_.

Referenced by demonstrate_shutdown().

Here is the caller graph for this function:

◆ shutdown()

void database::async::async_executor::shutdown ( )

inline

Gracefully shuts down the executor.

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 441 of file async_operations.h.

                        {
#ifdef USE_THREAD_SYSTEM
            if (pool_) {
                pool_->stop(false);
            }
#else
            {
                std::unique_lock<std::mutex> lock(queue_mutex_);
                stop_ = true;
            }
            condition_.notify_all();
 
            for (auto& worker : workers_) {
                if (worker.joinable()) {
                    worker.join();
                }
            }
            workers_.clear();
#endif
        }

References condition_, queue_mutex_, stop_, worker(), and workers_.

Referenced by demonstrate_shutdown(), TEST_F(), and ~async_executor().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ submit()

template<typename F , typename... Args>
requires concepts::SubmittableTask<F, Args...>

auto database::async::async_executor::submit	(	F &&	func,
		Args &&...	args ) -> std::future<std::invoke_result_t<F, Args...>>

inline

Submits a task for asynchronous execution.

Template Parameters

F	Callable type - constrained by SubmittableTask concept
Args	Argument types

Parameters

func	The callable to execute
args	Arguments to pass to the callable

Returns: std::future with the result of the callable

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h, and async_executor_demo.cpp.

Definition at line 395 of file async_operations.h.

        {
            using return_type = std::invoke_result_t<F, Args...>;
 
#ifdef USE_THREAD_SYSTEM
            auto task = std::make_shared<std::packaged_task<return_type()>>(
                std::bind(std::forward<F>(func), std::forward<Args>(args)...)
            );
 
            auto future = task->get_future();
 
            auto job = std::make_unique<lambda_job>(
                [task]() { (*task)(); },
                "async_task"
            );
 
            auto result = pool_->enqueue(std::move(job));
            if (result.is_err()) {
                throw std::runtime_error("Failed to enqueue job: " +
                    result.error().message);
            }
 
            return future;
#else
            auto task = std::make_shared<std::packaged_task<return_type()>>(
                std::bind(std::forward<F>(func), std::forward<Args>(args)...)
            );
 
            auto future = task->get_future();
 
            {
                std::unique_lock<std::mutex> lock(queue_mutex_);
                if (stop_) {
                    throw std::runtime_error("Cannot submit task to stopped executor");
                }
                tasks_.emplace([task]() { (*task)(); });
            }
 
            condition_.notify_one();
            return future;
#endif
        }

References condition_, queue_mutex_, stop_, and tasks_.

Referenced by compare_with_legacy(), demonstrate_basic_usage(), demonstrate_error_handling(), demonstrate_high_throughput(), and demonstrate_shutdown().

Here is the caller graph for this function:

◆ thread_count()

size_t database::async::async_executor::thread_count ( ) const

inline

Returns the number of worker threads.

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h, and async_executor_demo.cpp.

Definition at line 503 of file async_operations.h.

                                    {
            return thread_count_;
        }

References thread_count_.

Referenced by demonstrate_basic_usage(), and TEST_F().

Here is the caller graph for this function:

◆ wait_for_completion()

void database::async::async_executor::wait_for_completion ( )

inline

Waits for all pending tasks to complete.

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 465 of file async_operations.h.

                                   {
#ifdef USE_THREAD_SYSTEM
            if (pool_) {
                while (pool_->get_job_queue()->size() > 0) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(10));
                }
            }
#else
            while (true) {
                {
                    std::unique_lock<std::mutex> lock(queue_mutex_);
                    if (tasks_.empty()) {
                        break;
                    }
                }
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
#endif
        }

References queue_mutex_, and tasks_.

Referenced by demonstrate_high_throughput().

Here is the caller graph for this function:

◆ worker_thread()

void database::async::async_executor::worker_thread ( )

inlineprivate

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 519 of file async_operations.h.

                             {
            while (true) {
                std::function<void()> task;
 
                {
                    std::unique_lock<std::mutex> lock(queue_mutex_);
                    condition_.wait(lock, [this] {
                        return stop_ || !tasks_.empty();
                    });
 
                    if (stop_ && tasks_.empty()) {
                        return;
                    }
 
                    if (!tasks_.empty()) {
                        task = std::move(tasks_.front());
                        tasks_.pop();
                    }
                }
 
                if (task) {
                    task();
                }
            }
        }