database::integrated::adapters::backends::fallback_thread_backend Class Reference

Simple thread pool using std::thread. More...

#include <fallback_thread_backend.h>

Inheritance diagram for database::integrated::adapters::backends::fallback_thread_backend:

Collaboration diagram for database::integrated::adapters::backends::fallback_thread_backend:

Public Member Functions
	fallback_thread_backend (const db_thread_config &config)
	~fallback_thread_backend () override
	fallback_thread_backend (const fallback_thread_backend &)=delete
fallback_thread_backend &	operator= (const fallback_thread_backend &)=delete
	fallback_thread_backend (fallback_thread_backend &&)=delete
fallback_thread_backend &	operator= (fallback_thread_backend &&)=delete
common::VoidResult	initialize () override
	Initialize the thread backend.
common::VoidResult	shutdown () override
	Shutdown the thread backend gracefully.
bool	is_initialized () const override
	Check if backend is initialized.
common::VoidResult	execute (std::function< void()> task) override
	Execute a task (fire-and-forget)
void	wait_for_completion () override
	Wait for all pending tasks to complete.
bool	wait_for_completion_timeout (std::chrono::milliseconds timeout) override
	Wait for completion with timeout.
std::size_t	worker_count () const override
	Get number of worker threads.
std::size_t	queue_size () const override
	Get current queue size.
bool	is_idle () const override
	Check if thread pool is idle.
Public Member Functions inherited from database::integrated::adapters::backends::thread_backend
virtual	~thread_backend ()=default

Private Member Functions
void	worker_thread ()
	Worker thread function.

Private Attributes
const db_thread_config &	config_
bool	initialized_
std::atomic< bool >	shutdown_requested_
std::vector< std::thread >	workers_
std::queue< std::function< void()> >	task_queue_
std::mutex	queue_mutex_
std::condition_variable	queue_cv_
std::atomic< std::size_t >	active_tasks_
std::mutex	completion_mutex_
std::condition_variable	completion_cv_

Detailed Description

Simple thread pool using std::thread.

This backend provides basic thread pooling when thread_system is unavailable. Uses std::thread and std::queue for task management.

Definition at line 46 of file fallback_thread_backend.h.

Constructor & Destructor Documentation

fallback_thread_backend() [1/3]

database::integrated::adapters::backends::fallback_thread_backend::fallback_thread_backend ( const db_thread_config & config )

explicit

Definition at line 26 of file fallback_thread_backend.cpp.

    : config_(config)
    , initialized_(false)
    , shutdown_requested_(false)
    , active_tasks_(0)
{
}

~fallback_thread_backend()

database::integrated::adapters::backends::fallback_thread_backend::~fallback_thread_backend ( )

override

Definition at line 34 of file fallback_thread_backend.cpp.

{
    if (initialized_)
    {
        shutdown();
    }
}

References initialized_, and shutdown().

Here is the call graph for this function:

fallback_thread_backend() [2/3]

database::integrated::adapters::backends::fallback_thread_backend::fallback_thread_backend ( const fallback_thread_backend & )

delete

fallback_thread_backend() [3/3]

database::integrated::adapters::backends::fallback_thread_backend::fallback_thread_backend ( fallback_thread_backend && )

delete

Member Function Documentation

execute()

common::VoidResult database::integrated::adapters::backends::fallback_thread_backend::execute ( std::function< void()> task )

overridevirtual

Execute a task (fire-and-forget)

Parameters

task	Task to execute

Returns

VoidResult::ok() on successful submission

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 127 of file fallback_thread_backend.cpp.

{
    if (!initialized_)
    {
        return make_error("Thread pool not initialized");
    }
 
    if (!task)
    {
        return make_error("Invalid task");
    }
 
    if (shutdown_requested_)
    {
        return make_error("Thread pool is shutting down");
    }
 
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
 
        // Check queue size limit
        if (config_.max_queue_size > 0 && task_queue_.size() >= config_.max_queue_size)
        {
            return make_error("Task queue full");
        }
 
        task_queue_.push(std::move(task));
    }
 
    queue_cv_.notify_one();
    return common::ok();
}

References config_, initialized_, database::integrated::db_thread_config::max_queue_size, common::ok(), queue_cv_, queue_mutex_, shutdown_requested_, and task_queue_.

Here is the call graph for this function:

initialize()

common::VoidResult database::integrated::adapters::backends::fallback_thread_backend::initialize ( )

overridevirtual

Initialize the thread backend.

Returns

VoidResult::ok() on success, error on failure

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 42 of file fallback_thread_backend.cpp.

{
    if (initialized_)
    {
        return common::ok();
    }
 
    try
    {
        shutdown_requested_ = false;
        active_tasks_ = 0;
 
        // Determine worker count
        std::size_t worker_count = config_.thread_count;
        if (worker_count == 0)
        {
            worker_count = std::max(1u, std::thread::hardware_concurrency());
        }
 
        // Create worker threads
        workers_.reserve(worker_count);
        for (std::size_t i = 0; i < worker_count; ++i)
        {
            workers_.emplace_back(&fallback_thread_backend::worker_thread, this);
        }
 
        initialized_ = true;
        return common::ok();
    }
    catch (const std::exception& e)
    {
        return make_error(std::string("Thread pool initialization failed: ") + e.what());
    }
}

References active_tasks_, config_, initialized_, common::ok(), shutdown_requested_, database::integrated::db_thread_config::thread_count, worker_count(), worker_thread(), and workers_.

Here is the call graph for this function:

is_idle()

bool database::integrated::adapters::backends::fallback_thread_backend::is_idle ( ) const

overridevirtual

Check if thread pool is idle.

Returns

true if no pending or running tasks

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 199 of file fallback_thread_backend.cpp.

{
    std::lock_guard<std::mutex> lock(queue_mutex_);
    return task_queue_.empty() && active_tasks_ == 0;
}

References active_tasks_, queue_mutex_, and task_queue_.

is_initialized()

bool database::integrated::adapters::backends::fallback_thread_backend::is_initialized ( ) const

overridevirtual

Check if backend is initialized.

Returns

true if initialized and ready

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 122 of file fallback_thread_backend.cpp.

{
    return initialized_;
}

References initialized_.

operator=() [1/2]

fallback_thread_backend & database::integrated::adapters::backends::fallback_thread_backend::operator= ( const fallback_thread_backend & )

delete

operator=() [2/2]

fallback_thread_backend & database::integrated::adapters::backends::fallback_thread_backend::operator= ( fallback_thread_backend && )

delete

queue_size()

std::size_t database::integrated::adapters::backends::fallback_thread_backend::queue_size ( ) const

overridevirtual

Get current queue size.

Returns

Number of pending tasks

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 193 of file fallback_thread_backend.cpp.

{
    std::lock_guard<std::mutex> lock(queue_mutex_);
    return task_queue_.size();
}

References queue_mutex_, and task_queue_.

shutdown()

common::VoidResult database::integrated::adapters::backends::fallback_thread_backend::shutdown ( )

overridevirtual

Shutdown the thread backend gracefully.

Returns

VoidResult::ok() on success, error on failure

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 77 of file fallback_thread_backend.cpp.

{
    if (!initialized_)
    {
        return common::ok();
    }
 
    try
    {
        // Signal shutdown
        shutdown_requested_ = true;
 
        // Wake up all workers
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            queue_cv_.notify_all();
        }
 
        // Wait for all workers to finish
        for (auto& worker : workers_)
        {
            if (worker.joinable())
            {
                worker.join();
            }
        }
 
        workers_.clear();
 
        // Clear remaining tasks
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            std::queue<std::function<void()>> empty;
            task_queue_.swap(empty);
        }
 
        initialized_ = false;
        return common::ok();
    }
    catch (const std::exception& e)
    {
        return make_error(std::string("Thread pool shutdown failed: ") + e.what());
    }
}

References initialized_, common::ok(), queue_cv_, queue_mutex_, shutdown_requested_, task_queue_, worker(), and workers_.

Referenced by ~fallback_thread_backend().

Here is the call graph for this function:

Here is the caller graph for this function:

wait_for_completion()

void database::integrated::adapters::backends::fallback_thread_backend::wait_for_completion ( )

overridevirtual

Wait for all pending tasks to complete.

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 160 of file fallback_thread_backend.cpp.

{
    if (!initialized_)
    {
        return;
    }
 
    std::unique_lock<std::mutex> lock(completion_mutex_);
    completion_cv_.wait(lock, [this]() {
        std::lock_guard<std::mutex> queue_lock(queue_mutex_);
        return task_queue_.empty() && active_tasks_ == 0;
    });
}

References active_tasks_, completion_cv_, completion_mutex_, initialized_, queue_mutex_, and task_queue_.

wait_for_completion_timeout()

bool database::integrated::adapters::backends::fallback_thread_backend::wait_for_completion_timeout ( std::chrono::milliseconds timeout )

overridevirtual

Wait for completion with timeout.

Parameters

timeout

Maximum wait time

Returns

true if all tasks completed, false if timeout

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 174 of file fallback_thread_backend.cpp.

{
    if (!initialized_)
    {
        return true;
    }
 
    std::unique_lock<std::mutex> lock(completion_mutex_);
    return completion_cv_.wait_for(lock, timeout, [this]() {
        std::lock_guard<std::mutex> queue_lock(queue_mutex_);
        return task_queue_.empty() && active_tasks_ == 0;
    });
}

References active_tasks_, completion_cv_, completion_mutex_, initialized_, queue_mutex_, task_queue_, and database::timeout.

worker_count()

std::size_t database::integrated::adapters::backends::fallback_thread_backend::worker_count ( ) const

overridevirtual

Get number of worker threads.

Returns

Worker count

Implements database::integrated::adapters::backends::thread_backend.

Definition at line 188 of file fallback_thread_backend.cpp.

{
    return workers_.size();
}

References workers_.

Referenced by initialize().

Here is the caller graph for this function:

worker_thread()

void database::integrated::adapters::backends::fallback_thread_backend::worker_thread ( )

private

Worker thread function.

Definition at line 205 of file fallback_thread_backend.cpp.

{
    while (true)
    {
        std::function<void()> task;
 
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
 
            // Wait for task or shutdown
            queue_cv_.wait(lock, [this]() {
                return shutdown_requested_ || !task_queue_.empty();
            });
 
            // Check shutdown
            if (shutdown_requested_ && task_queue_.empty())
            {
                break;
            }
 
            // Get next task
            if (!task_queue_.empty())
            {
                task = std::move(task_queue_.front());
                task_queue_.pop();
            }
        }
 
        // Execute task
        if (task)
        {
            ++active_tasks_;
 
            try
            {
                task();
            }
            catch (...)
            {
                // Swallow exceptions to prevent worker thread from terminating
            }
 
            --active_tasks_;
 
            // Notify completion waiters
            {
                std::lock_guard<std::mutex> lock(completion_mutex_);
                completion_cv_.notify_all();
            }
        }
    }
}

References active_tasks_, completion_cv_, completion_mutex_, queue_cv_, queue_mutex_, shutdown_requested_, and task_queue_.

Referenced by initialize().

Here is the caller graph for this function:

Member Data Documentation

active_tasks_

std::atomic<std::size_t> database::integrated::adapters::backends::fallback_thread_backend::active_tasks_

private

Definition at line 86 of file fallback_thread_backend.h.

Referenced by initialize(), is_idle(), wait_for_completion(), wait_for_completion_timeout(), and worker_thread().

completion_cv_

std::condition_variable database::integrated::adapters::backends::fallback_thread_backend::completion_cv_

private

Definition at line 88 of file fallback_thread_backend.h.

Referenced by wait_for_completion(), wait_for_completion_timeout(), and worker_thread().

completion_mutex_

std::mutex database::integrated::adapters::backends::fallback_thread_backend::completion_mutex_

mutableprivate

Definition at line 87 of file fallback_thread_backend.h.

Referenced by wait_for_completion(), wait_for_completion_timeout(), and worker_thread().

config_

const db_thread_config& database::integrated::adapters::backends::fallback_thread_backend::config_

private

Definition at line 76 of file fallback_thread_backend.h.

Referenced by execute(), and initialize().

initialized_

bool database::integrated::adapters::backends::fallback_thread_backend::initialized_

private

Definition at line 77 of file fallback_thread_backend.h.

Referenced by execute(), initialize(), is_initialized(), shutdown(), wait_for_completion(), wait_for_completion_timeout(), and ~fallback_thread_backend().

queue_cv_

std::condition_variable database::integrated::adapters::backends::fallback_thread_backend::queue_cv_

private

Definition at line 84 of file fallback_thread_backend.h.

Referenced by execute(), shutdown(), and worker_thread().

queue_mutex_

std::mutex database::integrated::adapters::backends::fallback_thread_backend::queue_mutex_

mutableprivate

Definition at line 83 of file fallback_thread_backend.h.

Referenced by execute(), is_idle(), queue_size(), shutdown(), wait_for_completion(), wait_for_completion_timeout(), and worker_thread().

shutdown_requested_

std::atomic<bool> database::integrated::adapters::backends::fallback_thread_backend::shutdown_requested_

private

Definition at line 78 of file fallback_thread_backend.h.

Referenced by execute(), initialize(), shutdown(), and worker_thread().

task_queue_

std::queue<std::function<void()> > database::integrated::adapters::backends::fallback_thread_backend::task_queue_

private

Definition at line 81 of file fallback_thread_backend.h.

Referenced by execute(), is_idle(), queue_size(), shutdown(), wait_for_completion(), wait_for_completion_timeout(), and worker_thread().

workers_

std::vector<std::thread> database::integrated::adapters::backends::fallback_thread_backend::workers_

private

Definition at line 80 of file fallback_thread_backend.h.

Referenced by initialize(), shutdown(), and worker_count().

---

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

• database/integrated/adapters/backends/fallback_thread_backend.h
• database/integrated/adapters/backends/fallback_thread_backend.cpp