async_writer.h

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.
 
#pragma once
 
#include "queued_writer_base.h"
#include <queue>
#include <thread>
#include <condition_variable>
#include <iostream>
 
namespace kcenon::logger {
 
class async_writer : public queued_writer_base<std::queue<log_entry>> {
    using base_type = queued_writer_base<std::queue<log_entry>>;
 
public:
    explicit async_writer(std::unique_ptr<log_writer_interface> wrapped_writer,
                         std::size_t queue_size = 10000,
                         std::chrono::seconds flush_timeout = std::chrono::seconds(5))
        : base_type(std::move(wrapped_writer), queue_size, "async")
        , flush_timeout_(flush_timeout)
        , running_(false) {
    }
 
    ~async_writer() override {
        stop();
    }
 
    void start() {
        // Use compare_exchange to safely check and set running_ flag
        bool expected = false;
        if (!running_.compare_exchange_strong(expected, true)) {
            return; // Already running
        }
 
        // Try to create the worker thread with proper error handling
        try {
            worker_thread_ = std::thread([this]() {
                process_messages();
            });
        } catch (const std::exception&) {
            // Thread creation failed, rollback the running_ flag
            running_.store(false);
            throw; // Re-throw to notify caller
        }
    }
 
    void stop(bool force_flush = true) {
        if (!running_.exchange(false)) {
            return; // Already stopped
        }
 
        // Process remaining messages if requested
        if (force_flush) {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            size_t remaining_count = queue_.size();
            if (remaining_count > 0) {
                std::cerr << "[async_writer] Info: Processing " << remaining_count
                          << " remaining messages before shutdown.\n";
            }
        }
 
        // Signal the worker thread to stop
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            queue_cv_.notify_all();
        }
 
        // Wait for the worker thread to finish (infinite wait - shutdown prioritizes safety)
        if (worker_thread_.joinable()) {
            worker_thread_.join();
        }
 
        // Verify all messages were processed
        if (!queue_.empty()) {
            std::cerr << "[async_writer] Warning: " << queue_.size()
                      << " messages were not processed during shutdown.\n";
        }
    }
 
    common::VoidResult write(const log_entry& entry) override {
        if (!running_) {
            // If not running, write directly
            return wrapped().write(entry);
        }
 
        return try_enqueue(entry);
    }
 
    common::VoidResult flush() override {
        if (!running_) {
            return wrapped().flush();
        }
 
        // Wait for the queue to be empty with a timeout to prevent indefinite blocking
        std::unique_lock<std::mutex> lock(queue_mutex_);
        bool flushed = flush_cv_.wait_for(lock, flush_timeout_, [this]() {
            return queue_.empty();
        });
 
        if (!flushed) {
            // Timeout occurred - worker thread may have exited or is blocked
            return make_logger_void_result(logger_error_code::flush_timeout,
                                           "Flush operation timed out after " +
                                           std::to_string(flush_timeout_.count()) + " seconds");
        }
 
        // Flush the wrapped writer
        return wrapped().flush();
    }
 
    bool is_healthy() const override {
        return wrapped().is_healthy() && running_;
    }
 
    std::string get_name() const override {
        return "async_" + wrapped().get_name();
    }
 
protected:
    common::VoidResult handle_overflow(const log_entry& /*entry*/) override {
        return make_logger_void_result(logger_error_code::queue_full, "Async writer queue is full");
    }
 
    void on_entry_enqueued() override {
        queue_cv_.notify_one();
    }
 
private:
    void process_messages() {
        while (running_) {
            std::unique_lock<std::mutex> lock(queue_mutex_);
 
            // Wait for messages or stop signal
            queue_cv_.wait(lock, [this]() {
                return !queue_.empty() || !running_;
            });
 
            // Process all available messages
            while (!queue_.empty()) {
                auto entry = std::move(queue_.front());
                queue_.pop();
 
                // Unlock while writing
                lock.unlock();
                wrapped().write(entry);
                lock.lock();
            }
 
            // Notify flush waiters
            flush_cv_.notify_all();
        }
    }
 
    void flush_remaining() {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        while (!queue_.empty()) {
            auto entry = std::move(queue_.front());
            queue_.pop();
            wrapped().write(entry);
        }
        wrapped().flush();
    }
 
    std::chrono::seconds flush_timeout_;  // Configurable flush timeout
 
    std::condition_variable queue_cv_;
    std::condition_variable flush_cv_;
 
    std::atomic<bool> running_;
    std::thread worker_thread_;
};
 
} // namespace kcenon::logger