kcenon::logger::async::batch_processor Class Reference

Advanced batch processor with dynamic sizing and back-pressure handling. More...

#include <batch_processor.h>

Collaboration diagram for kcenon::logger::async::batch_processor:

Classes
struct	batch_entry
	Batch entry structure. More...
struct	config
	Configuration for batch processor. More...
struct	processing_stats
	Processing statistics. More...

Public Member Functions
	batch_processor (log_writer_ptr writer, const config &cfg=config{})
	Constructor.
	~batch_processor ()
	Destructor.
bool	start ()
	Start the batch processor.
void	stop (bool flush_remaining=true)
	Stop the batch processor.
bool	add_entry (batch_entry &&entry)
	Add entry to the batch queue.
bool	add_entry (const batch_entry &entry)
	Add entry to the batch queue (copy version)
void	flush ()
	Force flush current batch.
bool	is_healthy () const
	Check if processor is healthy.
const processing_stats &	get_stats () const
	Get current processing statistics.
void	reset_stats ()
	Reset statistics.
size_t	get_queue_size () const
	Get current queue size.
size_t	get_current_batch_size () const
	Get current batch size setting.

Private Member Functions
void	process_loop_iteration ()
	Single iteration of processing loop.
size_t	process_batch (std::vector< batch_entry > &batch)
	Process current batch.
size_t	collect_entries (std::vector< batch_entry > &batch, size_t max_entries, std::chrono::steady_clock::time_point deadline)
	Collect entries for batch processing.
void	adjust_batch_size ()
	Adjust batch size based on performance metrics.
bool	handle_back_pressure ()
	Handle back-pressure conditions.
bool	should_flush_by_time (std::chrono::steady_clock::time_point last_flush_time) const
	Check if batch should be flushed based on time.
void	update_stats (size_t batch_size, std::chrono::nanoseconds processing_time, const std::string &flush_reason)
	Update processing statistics.

Private Attributes
config	config_
log_writer_ptr	writer_
std::unique_ptr< lockfree_spsc_queue< batch_entry, queue_size > >	queue_
std::unique_ptr< batch_processing_jthread_worker >	processing_worker_
std::atomic< bool >	running_ {false}
std::atomic< bool >	should_stop_ {false}
std::mutex	notify_mutex_
std::condition_variable	notify_cv_
std::atomic< size_t >	current_batch_size_
std::atomic< std::chrono::milliseconds >	current_wait_time_
processing_stats	stats_
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_

Static Private Attributes
static constexpr size_t	queue_size = 8192

Detailed Description

Advanced batch processor with dynamic sizing and back-pressure handling.

Definition at line 36 of file batch_processor.h.

Constructor & Destructor Documentation

batch_processor()

kcenon::logger::async::batch_processor::batch_processor ( log_writer_ptr writer, const config & cfg = config{} )

explicit

Constructor.

Parameters

writer	Target writer for batch output
cfg	Configuration

Definition at line 119 of file batch_processor.cpp.

    : config_(cfg)
    , writer_(std::move(writer))
    , queue_(std::make_unique<lockfree_spsc_queue<batch_entry, queue_size>>())
    , current_batch_size_(cfg.initial_batch_size)
    , current_wait_time_(cfg.max_wait_time)
    , last_adjustment_time_(std::chrono::steady_clock::now()) {
 
    if (!writer_) {
        throw std::invalid_argument("Writer cannot be null");
    }
 
    // Validate configuration
    if (cfg.min_batch_size > cfg.max_batch_size) {
        throw std::invalid_argument("min_batch_size cannot be greater than max_batch_size");
    }
 
    if (cfg.initial_batch_size < cfg.min_batch_size || cfg.initial_batch_size > cfg.max_batch_size) {
        throw std::invalid_argument("initial_batch_size must be within min/max range");
    }
}

References kcenon::logger::async::batch_processor::config::initial_batch_size, kcenon::logger::async::batch_processor::config::max_batch_size, kcenon::logger::async::batch_processor::config::min_batch_size, and writer_.

~batch_processor()

kcenon::logger::async::batch_processor::~batch_processor

(

)

Destructor.

Definition at line 141 of file batch_processor.cpp.

                                  {
    stop(true);
}

References stop().

Here is the call graph for this function:

Member Function Documentation

add_entry() [1/2]

bool kcenon::logger::async::batch_processor::add_entry

(

batch_entry &&

entry

)

Add entry to the batch queue.

Parameters

entry

Log entry to add

Returns

true if added successfully, false if queue is full

Definition at line 193 of file batch_processor.cpp.

                                                   {
    if (!running_.load(std::memory_order_relaxed)) {
        return false;
    }
 
    if (!queue_->enqueue(std::move(entry))) {
        stats_.dropped_entries.fetch_add(1, std::memory_order_relaxed);
        return false;
    }
 
    // Wake the worker thread immediately
    notify_cv_.notify_one();
 
    return true;
}

References kcenon::logger::async::batch_processor::processing_stats::dropped_entries, notify_cv_, queue_, running_, and stats_.

Referenced by add_entry().

Here is the caller graph for this function:

add_entry() [2/2]

bool kcenon::logger::async::batch_processor::add_entry

(

const batch_entry &

entry

)

Add entry to the batch queue (copy version)

Parameters

entry

Log entry to add

Returns

true if added successfully, false if queue is full

Definition at line 209 of file batch_processor.cpp.

                                                        {
    return add_entry(batch_entry(entry));
}

References add_entry().

Here is the call graph for this function:

adjust_batch_size()

void kcenon::logger::async::batch_processor::adjust_batch_size ( )

private

Adjust batch size based on performance metrics.

Definition at line 336 of file batch_processor.cpp.

                                        {
    const auto current_size = current_batch_size_.load(std::memory_order_relaxed);
    const auto queue_size = get_queue_size();
    const auto recent_time = recent_processing_time_ms_.load(std::memory_order_relaxed);
 
    size_t new_size = current_size;
 
    // Increase batch size if queue is building up and processing is fast
    if (queue_size > current_size * 2 && recent_time < 10.0) {
        new_size = std::min(config_.max_batch_size,
                           static_cast<size_t>(current_size * config_.size_increase_factor));
    }
    // Decrease batch size if processing is slow or queue is small
    else if (recent_time > 100.0 || queue_size < current_size / 4) {
        new_size = std::max(config_.min_batch_size,
                           static_cast<size_t>(current_size * config_.size_decrease_factor));
    }
 
    if (new_size != current_size) {
        current_batch_size_.store(new_size, std::memory_order_relaxed);
        stats_.dynamic_size_adjustments.fetch_add(1, std::memory_order_relaxed);
    }
}

References config_, current_batch_size_, kcenon::logger::async::batch_processor::processing_stats::dynamic_size_adjustments, get_queue_size(), kcenon::logger::async::batch_processor::config::max_batch_size, kcenon::logger::async::batch_processor::config::min_batch_size, queue_size, recent_processing_time_ms_, kcenon::logger::async::batch_processor::config::size_decrease_factor, kcenon::logger::async::batch_processor::config::size_increase_factor, and stats_.

Referenced by process_loop_iteration().

Here is the call graph for this function:

Here is the caller graph for this function:

collect_entries()

size_t kcenon::logger::async::batch_processor::collect_entries ( std::vector< batch_entry > & batch, size_t max_entries, std::chrono::steady_clock::time_point deadline )

private

Collect entries for batch processing.

Parameters

batch	Vector to fill with entries
max_entries	Maximum entries to collect
deadline	Time deadline for collection

Returns

Number of entries collected

Definition at line 279 of file batch_processor.cpp.

                                                                                 {
    size_t collected = 0;
    batch_entry entry;
 
    while (collected < max_entries && std::chrono::steady_clock::now() < deadline) {
        if (queue_->dequeue(entry)) {
            batch.push_back(std::move(entry));
            ++collected;
        } else {
            // Queue is empty, wait for notification or remaining deadline
            auto remaining = deadline - std::chrono::steady_clock::now();
            if (remaining <= std::chrono::steady_clock::duration::zero()) {
                break;
            }
            std::unique_lock<std::mutex> lock(notify_mutex_);
            notify_cv_.wait_for(lock, remaining,
                                [this]{ return queue_->size() > 0 || should_stop_.load(std::memory_order_relaxed); });
        }
    }
 
    // Try to collect more entries if we have time left and space
    while (collected < max_entries && queue_->dequeue(entry)) {
        batch.push_back(std::move(entry));
        ++collected;
    }
 
    return collected;
}

References notify_cv_, notify_mutex_, queue_, and should_stop_.

Referenced by process_loop_iteration().

Here is the caller graph for this function:

flush()

void kcenon::logger::async::batch_processor::flush

(

)

Force flush current batch.

Definition at line 213 of file batch_processor.cpp.

                            {
    // This is handled by the processing loop
    // We could add a flush signal mechanism here if needed
}

get_current_batch_size()

size_t kcenon::logger::async::batch_processor::get_current_batch_size ( ) const

inline

Get current batch size setting.

Returns

Current batch size

Definition at line 190 of file batch_processor.h.

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

References current_batch_size_.

get_queue_size()

size_t kcenon::logger::async::batch_processor::get_queue_size

(

)

const

Get current queue size.

Returns

Approximate queue size

Definition at line 223 of file batch_processor.cpp.

                                             {
    return queue_->size();
}

References queue_.

Referenced by adjust_batch_size(), handle_back_pressure(), and update_stats().

Here is the caller graph for this function:

get_stats()

const processing_stats & kcenon::logger::async::batch_processor::get_stats ( ) const

inline

Get current processing statistics.

Returns

Reference to current stats

Definition at line 173 of file batch_processor.h.

{ return stats_; }

References stats_.

handle_back_pressure()

bool kcenon::logger::async::batch_processor::handle_back_pressure ( )

private

Handle back-pressure conditions.

Returns

true if should continue processing

Definition at line 360 of file batch_processor.cpp.

                                           {
    const auto queue_size = get_queue_size();
 
    if (queue_size > config_.back_pressure_threshold) {
        stats_.back_pressure_events.fetch_add(1, std::memory_order_relaxed);
 
        // Apply back-pressure delay
        std::this_thread::sleep_for(config_.back_pressure_delay);
 
        return queue_size < queue_size * 1.5; // Continue if queue isn't growing too fast
    }
 
    return true;
}

References kcenon::logger::async::batch_processor::config::back_pressure_delay, kcenon::logger::async::batch_processor::processing_stats::back_pressure_events, kcenon::logger::async::batch_processor::config::back_pressure_threshold, config_, get_queue_size(), queue_size, and stats_.

Referenced by process_loop_iteration().

Here is the call graph for this function:

Here is the caller graph for this function:

is_healthy()

bool kcenon::logger::async::batch_processor::is_healthy

(

)

const

Check if processor is healthy.

Returns

true if running and writer is healthy

Definition at line 218 of file batch_processor.cpp.

                                       {
    return running_.load(std::memory_order_relaxed) &&
           writer_ && writer_->is_healthy();
}

References running_, and writer_.

process_batch()

size_t kcenon::logger::async::batch_processor::process_batch ( std::vector< batch_entry > & batch )

private

Process current batch.

Parameters

batch

Vector of entries to process

Returns

Number of entries processed

Definition at line 310 of file batch_processor.cpp.

                                                                   {
    if (batch.empty() || !writer_) {
        return 0;
    }
 
    size_t processed = 0;
    for (const auto& entry : batch) {
        // Create log_entry from batch_entry
        log_entry log_ent(entry.level, entry.message, entry.file,
                         entry.line, entry.function, entry.timestamp);
 
        auto result = writer_->write(log_ent);
        if (result.is_ok()) {
            ++processed;
        }
    }
 
    // Flush after batch processing
    writer_->flush();
 
    stats_.total_batches.fetch_add(1, std::memory_order_relaxed);
    stats_.total_entries.fetch_add(processed, std::memory_order_relaxed);
 
    return processed;
}

References stats_, kcenon::logger::async::batch_processor::processing_stats::total_batches, kcenon::logger::async::batch_processor::processing_stats::total_entries, and writer_.

Referenced by process_loop_iteration(), and stop().

Here is the caller graph for this function:

process_loop_iteration()

void kcenon::logger::async::batch_processor::process_loop_iteration ( )

private

Single iteration of processing loop.

Definition at line 227 of file batch_processor.cpp.

                                             {
    static thread_local std::chrono::steady_clock::time_point last_flush_time = std::chrono::steady_clock::now();
    static thread_local std::chrono::steady_clock::time_point last_adjustment_time = std::chrono::steady_clock::now();
 
    const auto batch_size = current_batch_size_.load(std::memory_order_relaxed);
    const auto wait_time = current_wait_time_.load(std::memory_order_relaxed);
 
    std::vector<batch_entry> current_batch;
    current_batch.reserve(batch_size);
 
    const auto deadline = std::chrono::steady_clock::now() + wait_time;
    const auto entries_collected = collect_entries(current_batch, batch_size, deadline);
 
    if (entries_collected > 0) {
        const auto process_start = std::chrono::steady_clock::now();
        const auto processed = process_batch(current_batch);
        const auto process_end = std::chrono::steady_clock::now();
 
        const auto processing_time = process_end - process_start;
        const bool flushed_by_size = (entries_collected >= batch_size);
        const bool flushed_by_time = should_flush_by_time(last_flush_time);
 
        std::string flush_reason;
        if (flushed_by_size) {
            flush_reason = "size";
            stats_.flush_by_size.fetch_add(1, std::memory_order_relaxed);
        } else if (flushed_by_time) {
            flush_reason = "time";
            stats_.flush_by_time.fetch_add(1, std::memory_order_relaxed);
        } else {
            flush_reason = "partial";
        }
 
        update_stats(processed, processing_time, flush_reason);
        last_flush_time = process_end;
 
        // Handle back-pressure
        if (config_.enable_back_pressure) {
            handle_back_pressure();
        }
 
        // Dynamic batch size adjustment
        if (config_.enable_dynamic_sizing) {
            const auto now = std::chrono::steady_clock::now();
            if (now - last_adjustment_time > std::chrono::seconds(5)) {
                adjust_batch_size();
                last_adjustment_time = now;
            }
        }
    }
}

References adjust_batch_size(), collect_entries(), config_, current_batch_size_, current_wait_time_, kcenon::logger::async::batch_processor::config::enable_back_pressure, kcenon::logger::async::batch_processor::config::enable_dynamic_sizing, kcenon::logger::async::batch_processor::processing_stats::flush_by_size, kcenon::logger::async::batch_processor::processing_stats::flush_by_time, handle_back_pressure(), process_batch(), should_flush_by_time(), stats_, and update_stats().

Referenced by start().

Here is the call graph for this function:

Here is the caller graph for this function:

reset_stats()

void kcenon::logger::async::batch_processor::reset_stats ( )

inline

Reset statistics.

Definition at line 178 of file batch_processor.h.

{ stats_.reset(); }

References kcenon::logger::async::batch_processor::processing_stats::reset(), and stats_.

Here is the call graph for this function:

should_flush_by_time()

bool kcenon::logger::async::batch_processor::should_flush_by_time ( std::chrono::steady_clock::time_point last_flush_time ) const

private

Check if batch should be flushed based on time.

Parameters

last_flush_time

Time of last flush

Returns

true if timeout reached

Definition at line 375 of file batch_processor.cpp.

                                                                                                {
    const auto now = std::chrono::steady_clock::now();
    const auto elapsed = now - last_flush_time;
    const auto current_wait = current_wait_time_.load(std::memory_order_relaxed);
 
    return elapsed >= current_wait;
}

References current_wait_time_.

Referenced by process_loop_iteration().

Here is the caller graph for this function:

start()

bool kcenon::logger::async::batch_processor::start

(

)

Start the batch processor.

Returns

true if started successfully

Definition at line 145 of file batch_processor.cpp.

                            {
    bool expected = false;
    if (!running_.compare_exchange_strong(expected, true)) {
        return false; // Already running
    }
 
    should_stop_ = false;
    processing_worker_ = std::make_unique<batch_processing_jthread_worker>(
        [this] { process_loop_iteration(); }, notify_mutex_, notify_cv_);
    processing_worker_->start();
    return true;
}

References notify_cv_, notify_mutex_, process_loop_iteration(), processing_worker_, running_, and should_stop_.

Here is the call graph for this function:

stop()

void kcenon::logger::async::batch_processor::stop

(

bool

flush_remaining = true

)

Stop the batch processor.

Parameters

flush_remaining

Whether to flush remaining entries

Definition at line 158 of file batch_processor.cpp.

                                               {
    if (!running_.exchange(false)) {
        return; // Already stopped
    }
 
    should_stop_ = true;
 
    // Wake the worker so it can observe should_stop_ and exit
    notify_cv_.notify_one();
 
    if (processing_worker_) {
        processing_worker_->stop();
        processing_worker_.reset();
    }
 
    if (flush_remaining) {
        // Process any remaining entries
        batch_entry entry;
        std::vector<batch_entry> final_batch;
        final_batch.reserve(queue_size);
 
        while (queue_->dequeue(entry)) {
            final_batch.push_back(std::move(entry));
        }
 
        if (!final_batch.empty()) {
            process_batch(final_batch);
        }
 
        if (writer_) {
            writer_->flush();
        }
    }
}

References notify_cv_, process_batch(), processing_worker_, queue_, queue_size, running_, should_stop_, and writer_.

Referenced by ~batch_processor().

Here is the call graph for this function:

Here is the caller graph for this function:

update_stats()

void kcenon::logger::async::batch_processor::update_stats ( size_t batch_size, std::chrono::nanoseconds processing_time, const std::string & flush_reason )

private

Update processing statistics.

Parameters

batch_size	Size of processed batch
processing_time	Time taken to process
flush_reason	Reason for flush

Definition at line 383 of file batch_processor.cpp.

                                                                 {
    (void)batch_size;  // Suppress unused parameter warning
    (void)flush_reason;  // Suppress unused parameter warning
    const double processing_time_ms =
        std::chrono::duration_cast<std::chrono::microseconds>(processing_time).count() / 1000.0;
 
    // Update recent processing time (exponential moving average)
    const double alpha = 0.1;
    const double current_time = recent_processing_time_ms_.load(std::memory_order_relaxed);
    const double new_time = alpha * processing_time_ms + (1.0 - alpha) * current_time;
    recent_processing_time_ms_.store(new_time, std::memory_order_relaxed);
 
    // Update average batch size
    const auto total_batches = stats_.total_batches.load(std::memory_order_relaxed);
    if (total_batches > 0) {
        const auto total_entries = stats_.total_entries.load(std::memory_order_relaxed);
        const double avg_size = static_cast<double>(total_entries) / total_batches;
        stats_.average_batch_size.store(avg_size, std::memory_order_relaxed);
    }
 
    // Update average processing time
    stats_.average_processing_time_ms.store(
        recent_processing_time_ms_.load(std::memory_order_relaxed),
        std::memory_order_relaxed);
 
    // Update recent queue size for adjustment algorithm
    recent_queue_size_.store(get_queue_size(), std::memory_order_relaxed);
}

References kcenon::logger::async::batch_processor::processing_stats::average_batch_size, kcenon::logger::async::batch_processor::processing_stats::average_processing_time_ms, get_queue_size(), recent_processing_time_ms_, recent_queue_size_, stats_, kcenon::logger::async::batch_processor::processing_stats::total_batches, and kcenon::logger::async::batch_processor::processing_stats::total_entries.

Referenced by process_loop_iteration().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

config_

config kcenon::logger::async::batch_processor::config_

private

Definition at line 247 of file batch_processor.h.

Referenced by adjust_batch_size(), handle_back_pressure(), and process_loop_iteration().

current_batch_size_

std::atomic<size_t> kcenon::logger::async::batch_processor::current_batch_size_

private

Definition at line 266 of file batch_processor.h.

Referenced by adjust_batch_size(), get_current_batch_size(), and process_loop_iteration().

current_wait_time_

std::atomic<std::chrono::milliseconds> kcenon::logger::async::batch_processor::current_wait_time_

private

Definition at line 267 of file batch_processor.h.

Referenced by process_loop_iteration(), and should_flush_by_time().

last_adjustment_time_

std::chrono::steady_clock::time_point kcenon::logger::async::batch_processor::last_adjustment_time_

private

Definition at line 275 of file batch_processor.h.

notify_cv_

std::condition_variable kcenon::logger::async::batch_processor::notify_cv_

private

Definition at line 263 of file batch_processor.h.

Referenced by add_entry(), collect_entries(), start(), and stop().

notify_mutex_

std::mutex kcenon::logger::async::batch_processor::notify_mutex_

private

Definition at line 262 of file batch_processor.h.

Referenced by collect_entries(), and start().

processing_worker_

std::unique_ptr<batch_processing_jthread_worker> kcenon::logger::async::batch_processor::processing_worker_

private

Definition at line 257 of file batch_processor.h.

Referenced by start(), and stop().

queue_

std::unique_ptr<lockfree_spsc_queue<batch_entry, queue_size> > kcenon::logger::async::batch_processor::queue_

private

Definition at line 254 of file batch_processor.h.

Referenced by add_entry(), collect_entries(), get_queue_size(), and stop().

queue_size

size_t kcenon::logger::async::batch_processor::queue_size = 8192

staticconstexprprivate

Definition at line 253 of file batch_processor.h.

Referenced by adjust_batch_size(), handle_back_pressure(), and stop().

recent_processing_time_ms_

std::atomic<double> kcenon::logger::async::batch_processor::recent_processing_time_ms_ {0.0}

private

Definition at line 273 of file batch_processor.h.

{0.0};

Referenced by adjust_batch_size(), and update_stats().

recent_queue_size_

std::atomic<size_t> kcenon::logger::async::batch_processor::recent_queue_size_ {0}

private

Definition at line 274 of file batch_processor.h.

{0};

Referenced by update_stats().

running_

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

private

Definition at line 258 of file batch_processor.h.

{false};

Referenced by add_entry(), is_healthy(), start(), and stop().

should_stop_

std::atomic<bool> kcenon::logger::async::batch_processor::should_stop_ {false}

private

Definition at line 259 of file batch_processor.h.

{false};

Referenced by collect_entries(), start(), and stop().

stats_

processing_stats kcenon::logger::async::batch_processor::stats_

mutableprivate

Definition at line 270 of file batch_processor.h.

Referenced by add_entry(), adjust_batch_size(), get_stats(), handle_back_pressure(), process_batch(), process_loop_iteration(), reset_stats(), and update_stats().

writer_

log_writer_ptr kcenon::logger::async::batch_processor::writer_

private

Definition at line 250 of file batch_processor.h.

Referenced by batch_processor(), is_healthy(), process_batch(), and stop().

---

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

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