async_worker.cpp

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#include "async_worker.h"
 
#include <cstdio>
#include <utility>
 
namespace kcenon::logger::async {
 
async_worker::async_worker(std::size_t queue_size)
    : queue_size_(queue_size)
{
}
 
async_worker::~async_worker() {
    stop();
}
 
void async_worker::start() {
    // 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
}
 
void async_worker::stop() {
    // 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();
}
 
bool async_worker::enqueue(task_type task) {
    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;
}
 
void async_worker::flush() {
    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));
    }
}
 
bool async_worker::is_running() const noexcept {
    return running_.load(std::memory_order_acquire);
}
 
std::size_t async_worker::pending_count() const {
    std::lock_guard<std::mutex> lock(queue_mutex_);
    return queue_.size();
}
 
std::size_t async_worker::capacity() const noexcept {
    return queue_size_;
}
 
std::uint64_t async_worker::dropped_count() const noexcept {
    return dropped_count_.load(std::memory_order_relaxed);
}
 
#if LOGGER_HAS_JTHREAD
void async_worker::worker_loop(std::stop_token stop_token) {
    while (!stop_token.stop_requested()) {
        task_type task;
 
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
 
            // Wait for task or stop signal
            // condition_variable_any works with stop_token
            bool has_work = queue_cv_.wait(lock, stop_token, [this]() {
                return !queue_.empty();
            });
 
            // Check if we should stop
            if (stop_token.stop_requested()) {
                break;
            }
 
            // Check if we have work (not spurious wakeup)
            if (!has_work || 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
            }
        }
    }
}
#else
void async_worker::worker_loop(simple_stop_source& stop) {
    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
            }
        }
    }
}
#endif
 
void async_worker::drain_queue() {
    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
            }
        }
    }
}
 
} // namespace kcenon::logger::async