batch_processor.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 <kcenon/logger/interfaces/log_entry.h>
#include <kcenon/logger/writers/base_writer.h>
#include <kcenon/logger/core/error_codes.h>
#include <kcenon/common/interfaces/logger_interface.h>
#include <vector>
#include <chrono>
#include <atomic>
#include <memory>
#include <functional>
#include <mutex>              // Added for synchronization primitives
#include <condition_variable> // Added for threading coordination
 
namespace kcenon::logger::async {
 
// Forward declaration for processing worker (std::jthread-based)
class batch_processing_jthread_worker;
 
class batch_processor {
public:
    struct config {
        size_t initial_batch_size{100};      
        size_t min_batch_size{10};           
        size_t max_batch_size{1000};         
 
        std::chrono::milliseconds max_wait_time{1000};    
        std::chrono::milliseconds min_wait_time{10};      
 
        bool enable_dynamic_sizing{true};     
        bool enable_back_pressure{true};      
 
        double size_increase_factor{1.5};     
        double size_decrease_factor{0.8};     
 
        size_t back_pressure_threshold{5000}; 
        std::chrono::microseconds back_pressure_delay{100}; 
 
        config() noexcept {}
    };
 
    struct batch_entry {
        common::interfaces::log_level level;
        std::string message;
        std::string file;
        int line;
        std::string function;
        std::chrono::system_clock::time_point timestamp;
 
        batch_entry() = default;
 
        batch_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) {}
    };
 
    struct processing_stats {
        std::atomic<uint64_t> total_batches{0};
        std::atomic<uint64_t> total_entries{0};
        std::atomic<uint64_t> dropped_entries{0};
        std::atomic<uint64_t> back_pressure_events{0};
        std::atomic<uint64_t> dynamic_size_adjustments{0};
 
        std::atomic<uint64_t> flush_by_size{0};
        std::atomic<uint64_t> flush_by_time{0};
        std::atomic<uint64_t> flush_by_manual{0};
 
        std::atomic<double> average_batch_size{0.0};
        std::atomic<double> average_processing_time_ms{0.0};
 
        void reset() {
            total_batches = 0;
            total_entries = 0;
            dropped_entries = 0;
            back_pressure_events = 0;
            dynamic_size_adjustments = 0;
            flush_by_size = 0;
            flush_by_time = 0;
            flush_by_manual = 0;
            average_batch_size = 0.0;
            average_processing_time_ms = 0.0;
        }
    };
 
    explicit batch_processor(log_writer_ptr writer,
                           const config& cfg = config{});
 
    ~batch_processor();
 
    bool start();
 
    void stop(bool flush_remaining = true);
 
    bool add_entry(batch_entry&& entry);
 
    bool add_entry(const batch_entry& entry);
 
    void flush();
 
    bool is_healthy() const;
 
    const processing_stats& get_stats() const { return stats_; }
 
    void reset_stats() { stats_.reset(); }
 
    size_t get_queue_size() const;
 
    size_t get_current_batch_size() const {
        return current_batch_size_.load(std::memory_order_relaxed);
    }
 
private:
    void process_loop_iteration();
 
    size_t process_batch(std::vector<batch_entry>& batch);
 
    size_t collect_entries(std::vector<batch_entry>& batch,
                          size_t max_entries,
                          std::chrono::steady_clock::time_point deadline);
 
    void adjust_batch_size();
 
    bool handle_back_pressure();
 
    bool should_flush_by_time(std::chrono::steady_clock::time_point last_flush_time) const;
 
    void update_stats(size_t batch_size,
                     std::chrono::nanoseconds processing_time,
                     const std::string& flush_reason);
 
    // Configuration
    config config_;
 
    // Writer
    log_writer_ptr writer_;
 
    // Queue
    static constexpr size_t queue_size = 8192;  // Must be power of 2
    std::unique_ptr<lockfree_spsc_queue<batch_entry, queue_size>> queue_;
 
    // Processing worker (using std::jthread)
    std::unique_ptr<batch_processing_jthread_worker> processing_worker_;
    std::atomic<bool> running_{false};
    std::atomic<bool> should_stop_{false};
 
    // Condition variable for waking the worker thread when new entries arrive
    std::mutex notify_mutex_;
    std::condition_variable notify_cv_;
 
    // Dynamic sizing
    std::atomic<size_t> current_batch_size_;
    std::atomic<std::chrono::milliseconds> current_wait_time_;
 
    // Statistics
    mutable processing_stats stats_;
 
    // Performance metrics for dynamic adjustment
    std::atomic<double> recent_processing_time_ms_{0.0};
    std::atomic<size_t> recent_queue_size_{0};
    std::chrono::steady_clock::time_point last_adjustment_time_;
};
 
std::unique_ptr<batch_processor> make_batch_processor(
    log_writer_ptr writer,
    const batch_processor::config& cfg = batch_processor::config{});
 
} // namespace kcenon::logger::async