high_performance_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 "lockfree_queue.h"
#include "batch_processor.h"
#include "../memory/object_pool.h"
#include "../memory/log_entry_pool.h"
#include <kcenon/logger/writers/base_writer.h>
#include <kcenon/logger/interfaces/log_entry.h>
#include <kcenon/logger/core/error_codes.h>
#include <memory>
#include <thread>
#include <atomic>
#include <chrono>
#include <functional>  // Added for std::function usage
 
namespace kcenon::logger::async {
 
class high_performance_async_writer : public base_writer {
public:
    struct config {
        // Queue configuration
        size_t queue_size{8192};              
 
        // Batch processing configuration
        batch_processor::config batch_config; 
 
        // Memory pool configuration
        memory::object_pool<memory::log_entry_pool::pooled_log_entry>::config pool_config;
 
        // Performance tuning
        bool enable_memory_pooling{true};     
        bool enable_batch_processing{true};   
 
        std::chrono::microseconds flush_timeout{1000}; 
 
        config() {
            // Optimize batch config for high performance
            batch_config.initial_batch_size = 50;
            batch_config.max_batch_size = 500;
            batch_config.min_batch_size = 10;
            batch_config.max_wait_time = std::chrono::milliseconds{100};
            batch_config.enable_dynamic_sizing = true;
            batch_config.enable_back_pressure = true;
 
            // Optimize pool config
            pool_config.initial_size = 200;
            pool_config.max_size = 2000;
            pool_config.allow_growth = true;
        }
    };
 
    struct performance_stats {
        std::atomic<uint64_t> total_writes{0};
        std::atomic<uint64_t> successful_writes{0};
        std::atomic<uint64_t> dropped_writes{0};
        std::atomic<uint64_t> queue_full_events{0};
 
        std::atomic<double> average_latency_us{0.0};
        std::atomic<double> throughput_per_second{0.0};
 
        std::chrono::steady_clock::time_point start_time{std::chrono::steady_clock::now()};
 
        double success_ratio() const {
            const auto total = total_writes.load();
            return total > 0 ? static_cast<double>(successful_writes.load()) / total : 0.0;
        }
 
        void reset() {
            total_writes = 0;
            successful_writes = 0;
            dropped_writes = 0;
            queue_full_events = 0;
            average_latency_us = 0.0;
            throughput_per_second = 0.0;
            start_time = std::chrono::steady_clock::now();
        }
    };
 
    explicit high_performance_async_writer(
        log_writer_ptr wrapped_writer,
        const config& cfg = config{});
 
    ~high_performance_async_writer() override;
 
    bool start();
 
    void stop(bool flush_remaining = true);
 
    common::VoidResult write(const log_entry& entry) override;
 
    common::VoidResult flush() override;
 
    bool is_healthy() const override;
 
    std::string get_name() const override;
 
    void set_use_color(bool use_color) override;
 
    const performance_stats& get_stats() const { return stats_; }
 
    void reset_stats() { stats_.reset(); }
 
    double get_queue_utilization() const;
 
    const batch_processor::processing_stats* get_batch_stats() const;
 
private:
    struct queued_log_entry {
        common::interfaces::log_level level;
        std::string message;
        std::string file;
        int line;
        std::string function;
        std::chrono::system_clock::time_point timestamp;
        std::chrono::steady_clock::time_point enqueue_time; 
 
        queued_log_entry() = default;
 
        queued_log_entry(common::interfaces::log_level lvl,
                        std::string msg,
                        std::string f,
                        int l,
                        std::string func,
                        std::chrono::system_clock::time_point ts)
            : level(lvl)
            , message(std::move(msg))
            , file(std::move(f))
            , line(l)
            , function(std::move(func))
            , timestamp(ts)
            , enqueue_time(std::chrono::steady_clock::now()) {}
    };
 
    common::VoidResult write_direct(const log_entry& entry);
 
    void update_stats(bool success, std::chrono::nanoseconds latency);
 
    batch_processor::batch_entry to_batch_entry(const queued_log_entry& entry) const;
 
    // Configuration
    config config_;
 
    // Underlying writer
    log_writer_ptr wrapped_writer_;
 
    // High-performance components
    std::unique_ptr<batch_processor> batch_processor_;
    std::unique_ptr<memory::object_pool<memory::log_entry_pool::pooled_log_entry>> memory_pool_;
 
    // State management
    std::atomic<bool> running_{false};
 
    // Performance tracking
    mutable performance_stats stats_;
};
 
std::unique_ptr<high_performance_async_writer> make_high_performance_async_writer(
    log_writer_ptr writer,
    const high_performance_async_writer::config& cfg = high_performance_async_writer::config{});
 
} // namespace kcenon::logger::async