kcenon::logger::async::async_worker Class Reference

Standalone async worker with jthread compatibility. More...

#include <async_worker.h>

Collaboration diagram for kcenon::logger::async::async_worker:

Public Member Functions
	async_worker (std::size_t queue_size=8192)
	Constructor with configurable queue size.
	~async_worker ()
	Destructor - ensures graceful shutdown.
	async_worker (const async_worker &)=delete
async_worker &	operator= (const async_worker &)=delete
	async_worker (async_worker &&)=delete
async_worker &	operator= (async_worker &&)=delete
void	start ()
	Start the worker thread.
void	stop ()
	Stop the worker thread gracefully.
bool	enqueue (task_type task)
	Enqueue a task for async processing.
void	flush ()
	Flush all pending tasks.
bool	is_running () const noexcept
	Check if the worker is currently running.
std::size_t	pending_count () const
	Get the current queue size.
std::size_t	capacity () const noexcept
	Get the queue capacity.
std::uint64_t	dropped_count () const noexcept
	Get the number of dropped tasks due to queue overflow.

Private Member Functions
void	worker_loop (simple_stop_source &stop)
	Worker thread main loop (fallback version)
void	drain_queue ()
	Process all remaining tasks in the queue.

Private Attributes
const std::size_t	queue_size_
	Maximum queue capacity.
std::queue< task_type >	queue_
	Task queue.
std::mutex	queue_mutex_
	Protects queue_ access.
std::condition_variable	queue_cv_
	Signals new tasks or stop (fallback)
compat_jthread	worker_thread_
	Background worker thread.
std::atomic< bool >	running_ {false}
	Worker state flag.
std::atomic< std::uint64_t >	dropped_count_ {0}
	Overflow counter.

Detailed Description

Standalone async worker with jthread compatibility.

This worker provides a background thread for processing logging tasks asynchronously. It uses std::jthread's cooperative cancellation mechanism via std::stop_token for graceful shutdown when available, or falls back to manual stop mechanism for environments without jthread support.

Thread Safety:

• enqueue() is thread-safe for multiple producers
• The worker thread is the single consumer
• flush() blocks until all pending tasks are processed

Performance Characteristics:

• Lock-free enqueue path when queue is not full
• Batch processing reduces context switching overhead
• Configurable queue size for different workloads

Usage:

async_worker worker(8192);
worker.start();
 
worker.enqueue([&logger, message]() {
    logger.write(message);
});
 
worker.flush();  // Wait for all pending tasks
worker.stop();   // Graceful shutdown

Since

1.3.0

Definition at line 82 of file async_worker.h.

Constructor & Destructor Documentation

async_worker() [1/3]

kcenon::logger::async::async_worker::async_worker ( std::size_t queue_size = 8192 )

explicit

Constructor with configurable queue size.

Parameters

queue_size

Maximum number of pending tasks (default: 8192)

The queue size should be chosen based on expected logging throughput:

• Higher values reduce the chance of dropped messages under burst load
• Lower values reduce memory footprint

Note

Memory usage is approximately queue_size * sizeof(task_type)

Definition at line 12 of file async_worker.cpp.

    : queue_size_(queue_size)
{
}

~async_worker()

kcenon::logger::async::async_worker::~async_worker

(

)

Destructor - ensures graceful shutdown.

Automatically calls stop() if the worker is still running. The destructor will block until all pending tasks are processed and the worker thread has terminated.

Definition at line 17 of file async_worker.cpp.

                            {
    stop();
}

References stop().

Here is the call graph for this function:

async_worker() [2/3]

kcenon::logger::async::async_worker::async_worker ( const async_worker & )

delete

async_worker() [3/3]

kcenon::logger::async::async_worker::async_worker ( async_worker && )

delete

Member Function Documentation

capacity()

std::size_t kcenon::logger::async::async_worker::capacity ( ) const

nodiscardnoexcept

Get the queue capacity.

Returns

Maximum number of tasks the queue can hold

Definition at line 122 of file async_worker.cpp.

                                                {
    return queue_size_;
}

References queue_size_.

drain_queue()

void kcenon::logger::async::async_worker::drain_queue ( )

private

Process all remaining tasks in the queue.

Called during shutdown to ensure no tasks are lost.

Definition at line 209 of file async_worker.cpp.

                               {
    std::queue<task_type> remaining;
 
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        std::swap(remaining, queue_);
    }
 
    // Process remaining tasks
    while (!remaining.empty()) {
        auto task = std::move(remaining.front());
        remaining.pop();
 
        if (task) {
            try {
                task();
            } catch (...) {
                // Swallow exceptions during drain
            }
        }
    }
}

References queue_, and queue_mutex_.

Referenced by flush(), and stop().

Here is the caller graph for this function:

dropped_count()

std::uint64_t kcenon::logger::async::async_worker::dropped_count ( ) const

nodiscardnoexcept

Get the number of dropped tasks due to queue overflow.

Returns

Total number of tasks that were not enqueued

Definition at line 126 of file async_worker.cpp.

                                                       {
    return dropped_count_.load(std::memory_order_relaxed);
}

References dropped_count_.

enqueue()

bool kcenon::logger::async::async_worker::enqueue

(

task_type

task

)

Enqueue a task for async processing.

Parameters

task	The task to execute (callable with no arguments)

Returns

true if task was enqueued, false if queue is full

If the queue is full, the task is dropped and false is returned. The caller should handle this case (e.g., log synchronously or retry).

Thread-safe: Multiple threads can call enqueue() concurrently.

if (!worker.enqueue(task)) {
    // Queue full - handle overflow
    task();  // Execute synchronously as fallback
}

Definition at line 62 of file async_worker.cpp.

                                         {
    if (!task || !running_.load(std::memory_order_acquire)) {
        return false;
    }
 
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
 
        // Check queue capacity
        if (queue_.size() >= queue_size_) {
            // Track dropped task
            std::uint64_t dropped = dropped_count_.fetch_add(1, std::memory_order_relaxed) + 1;
 
            // Log warning periodically (every 100 drops) to avoid log spam
            if (dropped % 100 == 1) {
                std::fprintf(stderr,
                    "[WARNING] Async worker queue full: %llu tasks dropped total\n",
                    static_cast<unsigned long long>(dropped));
            }
            return false;
        }
 
        queue_.push(std::move(task));
    }
 
    // Notify worker thread
    queue_cv_.notify_one();
    return true;
}

References dropped_count_, queue_, queue_cv_, queue_mutex_, queue_size_, and running_.

flush()

void kcenon::logger::async::async_worker::flush

(

)

Flush all pending tasks.

Blocks until all tasks currently in the queue have been processed. New tasks enqueued during flush() may or may not be included.

Thread-safe: Can be called from any thread.

Note

This method may block for an extended period if the queue contains many tasks or tasks are slow to execute.

Definition at line 92 of file async_worker.cpp.

                         {
    if (!running_.load(std::memory_order_acquire)) {
        // Not running - just drain the queue directly
        drain_queue();
        return;
    }
 
    // Wait until queue is empty
    // Use a polling approach with short waits to avoid deadlock
    while (true) {
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            if (queue_.empty()) {
                break;
            }
        }
        // Brief yield to allow worker to process
        std::this_thread::sleep_for(std::chrono::microseconds(100));
    }
}

References drain_queue(), queue_, queue_mutex_, and running_.

Here is the call graph for this function:

is_running()

bool kcenon::logger::async::async_worker::is_running ( ) const

nodiscardnoexcept

Check if the worker is currently running.

Returns

true if the worker thread is active

Definition at line 113 of file async_worker.cpp.

                                             {
    return running_.load(std::memory_order_acquire);
}

References running_.

operator=() [1/2]

async_worker & kcenon::logger::async::async_worker::operator= ( async_worker && )

delete

operator=() [2/2]

async_worker & kcenon::logger::async::async_worker::operator= ( const async_worker & )

delete

pending_count()

std::size_t kcenon::logger::async::async_worker::pending_count ( ) const

nodiscard

Get the current queue size.

Returns

Number of pending tasks in the queue

Note

This is an approximate value due to concurrent access.

Definition at line 117 of file async_worker.cpp.

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

References queue_, and queue_mutex_.

start()

void kcenon::logger::async::async_worker::start

(

)

Start the worker thread.

Creates a background thread that processes tasks from the queue. The thread will run until stop() is called or the worker is destroyed.

Thread-safe: Can be called from any thread. Idempotent: Multiple calls are safe (subsequent calls are no-ops).

Definition at line 21 of file async_worker.cpp.

                         {
    // Check if already running (idempotent)
    if (running_.exchange(true, std::memory_order_acq_rel)) {
        return;
    }
 
#if LOGGER_HAS_JTHREAD
    // Create worker thread with std::jthread
    // The stop_token is automatically provided by jthread
    worker_thread_ = compat_jthread([this](std::stop_token stop_token) {
        worker_loop(stop_token);
    });
#else
    // Create worker thread with manual stop source
    // Note: We use the stop_source created by compat_jthread to ensure
    // request_stop() correctly signals the worker loop
    worker_thread_ = compat_jthread([this](simple_stop_source& stop) {
        worker_loop(stop);
    });
#endif
}

References running_, stop(), worker_loop(), and worker_thread_.

Here is the call graph for this function:

stop()

void kcenon::logger::async::async_worker::stop

(

)

Stop the worker thread gracefully.

Signals the worker thread to stop and waits for it to complete processing remaining tasks. This method:

1. Requests stop via stop mechanism
2. Wakes up the worker thread
3. Waits for the thread to join
4. Processes any remaining tasks in the queue

Thread-safe: Can be called from any thread. Idempotent: Multiple calls are safe (subsequent calls are no-ops).

Note

This method blocks until the worker thread terminates.

Definition at line 43 of file async_worker.cpp.

                        {
    // Check if not running (idempotent)
    if (!running_.exchange(false, std::memory_order_acq_rel)) {
        return;
    }
 
    // Request stop
    worker_thread_.request_stop();
 
    // Wake up the worker thread
    queue_cv_.notify_all();
 
    // Wait for thread to finish
    worker_thread_.join();
 
    // Process any remaining tasks
    drain_queue();
}

References drain_queue(), kcenon::logger::async::compat_jthread::join(), queue_cv_, kcenon::logger::async::compat_jthread::request_stop(), running_, and worker_thread_.

Referenced by start(), worker_loop(), and ~async_worker().

Here is the call graph for this function:

Here is the caller graph for this function:

worker_loop()

void kcenon::logger::async::async_worker::worker_loop ( simple_stop_source & stop )

private

Worker thread main loop (fallback version)

Parameters

stop	Stop source for manual cancellation

Processes tasks from the queue until stop is requested. Uses condition variable for efficient waiting.

Definition at line 170 of file async_worker.cpp.

                                                       {
    while (!stop.stop_requested()) {
        task_type task;
 
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
 
            // Wait for task or stop signal
            queue_cv_.wait(lock, [this, &stop]() {
                return stop.stop_requested() || !queue_.empty();
            });
 
            // Check if we should stop
            if (stop.stop_requested()) {
                break;
            }
 
            // Check if we have work
            if (queue_.empty()) {
                continue;
            }
 
            // Get task from queue
            task = std::move(queue_.front());
            queue_.pop();
        }
 
        // Execute task outside the lock
        if (task) {
            try {
                task();
            } catch (...) {
                // Swallow exceptions to prevent thread termination
            }
        }
    }
}

References queue_, queue_cv_, queue_mutex_, and stop().

Referenced by start().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

dropped_count_

std::atomic<std::uint64_t> kcenon::logger::async::async_worker::dropped_count_ {0}

private

Overflow counter.

Definition at line 238 of file async_worker.h.

{0};  

Referenced by dropped_count(), and enqueue().

queue_

std::queue<task_type> kcenon::logger::async::async_worker::queue_

private

Task queue.

Definition at line 227 of file async_worker.h.

Referenced by drain_queue(), enqueue(), flush(), pending_count(), and worker_loop().

queue_cv_

std::condition_variable kcenon::logger::async::async_worker::queue_cv_

private

Signals new tasks or stop (fallback)

Definition at line 233 of file async_worker.h.

Referenced by enqueue(), stop(), and worker_loop().

queue_mutex_

std::mutex kcenon::logger::async::async_worker::queue_mutex_

mutableprivate

Protects queue_ access.

Definition at line 228 of file async_worker.h.

Referenced by drain_queue(), enqueue(), flush(), pending_count(), and worker_loop().

queue_size_

const std::size_t kcenon::logger::async::async_worker::queue_size_

private

Maximum queue capacity.

Definition at line 226 of file async_worker.h.

Referenced by capacity(), and enqueue().

running_

std::atomic<bool> kcenon::logger::async::async_worker::running_ {false}

private

Worker state flag.

Definition at line 237 of file async_worker.h.

{false};      

Referenced by enqueue(), flush(), is_running(), start(), and stop().

worker_thread_

compat_jthread kcenon::logger::async::async_worker::worker_thread_

private

Background worker thread.

Definition at line 236 of file async_worker.h.

Referenced by start(), and stop().

---

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

• src/impl/async/async_worker.h
• src/impl/async/async_worker.cpp