log_collector.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 <kcenon/logger/core/log_collector.h>
#include <kcenon/logger/writers/base_writer.h>
#include <kcenon/logger/interfaces/log_entry.h>
#include <kcenon/common/interfaces/logger_interface.h>
 
#include "../impl/async/jthread_compat.h"
 
#include <atomic>
#include <condition_variable>
#include <cstdio>
#include <functional>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>
 
namespace kcenon::logger {
 
// Type alias for log_level
using log_level = common::interfaces::log_level;
 
struct log_collector_shared_state {
    std::queue<log_entry> queue;
    mutable std::mutex queue_mutex;
#if LOGGER_HAS_JTHREAD
    std::condition_variable_any queue_cv;  // Works with stop_token
#else
    std::condition_variable queue_cv;       // Standard condition variable
#endif
    std::vector<std::weak_ptr<log_writer_interface>> writers;
    std::mutex writers_mutex;
    const std::size_t batch_size;
    const std::size_t buffer_size;
 
    explicit log_collector_shared_state(std::size_t buffer_sz, std::size_t batch_sz)
        : batch_size(batch_sz)
        , buffer_size(buffer_sz) {}
};
 
class log_collector_jthread_worker {
public:
    explicit log_collector_jthread_worker(std::shared_ptr<log_collector_shared_state> state)
        : state_(std::move(state))
        , running_(false)
    {}
 
    ~log_collector_jthread_worker() {
        stop();
    }
 
    void start() {
        if (running_.exchange(true, std::memory_order_acq_rel)) {
            return;  // Already started
        }
 
#if LOGGER_HAS_JTHREAD
        // Capture state by shared_ptr - survives worker destruction
        auto state = state_;
        thread_ = async::compat_jthread([state](std::stop_token stop_token) {
            worker_loop(state, 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
        auto state = state_;
        thread_ = async::compat_jthread([state](async::simple_stop_source& stop) {
            worker_loop(state, stop);
        });
#endif
    }
 
    void stop() {
        if (!running_.exchange(false, std::memory_order_acq_rel)) {
            return;  // Already stopped
        }
 
        // Request stop
        thread_.request_stop();
 
        // Wake up the worker in case it's waiting
        if (state_) {
            state_->queue_cv.notify_all();
        }
 
        // Wait for thread to finish
        thread_.join();
    }
 
    void notify_work() {
        if (state_) {
            state_->queue_cv.notify_one();
        }
    }
 
    [[nodiscard]] bool is_running() const noexcept {
        return running_.load(std::memory_order_acquire);
    }
 
private:
#if LOGGER_HAS_JTHREAD
    static void worker_loop(std::shared_ptr<log_collector_shared_state> state,
                           std::stop_token stop_token) {
        if (!state) {
            return;
        }
 
        while (!stop_token.stop_requested()) {
            std::vector<log_entry> batch;
 
            {
                std::unique_lock<std::mutex> lock(state->queue_mutex);
 
                // Wait for work or stop signal using condition_variable_any
                bool has_work = state->queue_cv.wait(lock, stop_token, [&state]() {
                    return !state->queue.empty();
                });
 
                // Check if stop was requested
                if (stop_token.stop_requested()) {
                    break;
                }
 
                // Check if we actually have work
                if (!has_work || state->queue.empty()) {
                    continue;
                }
 
                // Extract batch from queue
                batch.reserve(std::min(state->batch_size, state->queue.size()));
                while (!state->queue.empty() && batch.size() < state->batch_size) {
                    batch.push_back(std::move(state->queue.front()));
                    state->queue.pop();
                }
            }
 
            // Process batch outside the lock
            for (const auto& entry : batch) {
                write_to_all(state, entry);
            }
        }
    }
#else
    static void worker_loop(std::shared_ptr<log_collector_shared_state> state,
                           async::simple_stop_source& stop) {
        if (!state) {
            return;
        }
 
        while (!stop.stop_requested()) {
            std::vector<log_entry> batch;
 
            {
                std::unique_lock<std::mutex> lock(state->queue_mutex);
 
                // Wait for work or stop signal
                state->queue_cv.wait(lock, [&state, &stop]() {
                    return stop.stop_requested() || !state->queue.empty();
                });
 
                // Check if stop was requested
                if (stop.stop_requested()) {
                    break;
                }
 
                // Check if we actually have work
                if (state->queue.empty()) {
                    continue;
                }
 
                // Extract batch from queue
                batch.reserve(std::min(state->batch_size, state->queue.size()));
                while (!state->queue.empty() && batch.size() < state->batch_size) {
                    batch.push_back(std::move(state->queue.front()));
                    state->queue.pop();
                }
            }
 
            // Process batch outside the lock
            for (const auto& entry : batch) {
                write_to_all(state, entry);
            }
        }
    }
#endif
 
    static void write_to_all(const std::shared_ptr<log_collector_shared_state>& state,
                            const log_entry& entry) {
        if (!state) {
            return;
        }
 
        // Snapshot writers under lock
        std::vector<std::shared_ptr<log_writer_interface>> writers_snapshot;
        {
            std::lock_guard<std::mutex> lock(state->writers_mutex);
            writers_snapshot.reserve(state->writers.size());
            for (auto& weak_writer : state->writers) {
                if (auto writer = weak_writer.lock()) {
                    writers_snapshot.push_back(writer);
                }
            }
        }
 
        // Write to all writers without holding mutex
        for (auto& writer : writers_snapshot) {
            try {
                writer->write(entry);
            } catch (...) {
                // Swallow exceptions to prevent thread termination
            }
        }
    }
 
private:
    std::shared_ptr<log_collector_shared_state> state_;
    async::compat_jthread thread_;
    std::atomic<bool> running_;
};
 
class log_collector::impl {
public:
    explicit impl(std::size_t buffer_size, std::size_t batch_size)
        : state_(std::make_shared<log_collector_shared_state>(buffer_size, batch_size))
        , worker_(std::make_unique<log_collector_jthread_worker>(state_)) {
    }
 
    ~impl() {
        stop();
    }
 
    bool enqueue(log_level level,
                 const std::string& message,
                 const std::string& file,
                 int line,
                 const std::string& function,
                 const std::chrono::system_clock::time_point& timestamp) {
        {
            std::lock_guard<std::mutex> lock(state_->queue_mutex);
 
            // Check if queue is full
            if (state_->queue.size() >= state_->buffer_size) {
                // Track dropped message
                std::uint64_t dropped = dropped_messages_.fetch_add(1, std::memory_order_relaxed) + 1;
 
                // Log warning periodically (every 100 dropped messages) to avoid spam
                if (dropped % 100 == 1) {
                    std::fprintf(stderr, "[WARNING] Log queue full: %llu messages dropped total\n",
                               static_cast<unsigned long long>(dropped));
                }
 
                return false;
            }
 
            // Create log_entry with optional source location
            log_entry entry(level, message, timestamp);
            if (!file.empty() || line != 0 || !function.empty()) {
                entry.location = source_location{file, line, function};
            }
            state_->queue.push(std::move(entry));
        }
 
        // Notify worker thread
        if (worker_) {
            worker_->notify_work();
        }
        return true;
    }
 
    void add_writer(std::shared_ptr<log_writer_interface> writer) {
        if (!writer) {
            return;
        }
        std::lock_guard<std::mutex> lock(state_->writers_mutex);
        state_->writers.push_back(writer);
    }
 
    void clear_writers() {
        std::lock_guard<std::mutex> lock(state_->writers_mutex);
        state_->writers.clear();
    }
 
    void start() {
        if (worker_) {
            worker_->start();
        }
    }
 
    void stop() {
        // Stop the worker thread first
        if (worker_) {
            worker_->stop();
        }
 
        // Drain remaining entries
        drain_queue();
 
        // Flush all writers
        flush_writers();
    }
 
    void flush() {
        // Wait for queue to be empty
        while (true) {
            {
                std::lock_guard<std::mutex> lock(state_->queue_mutex);
                if (state_->queue.empty()) {
                    break;
                }
            }
            // Brief yield to allow worker to process
            std::this_thread::sleep_for(std::chrono::microseconds(100));
        }
 
        // Flush all writers
        flush_writers();
    }
 
    [[nodiscard]] std::pair<std::size_t, std::size_t> get_queue_metrics() const {
        std::lock_guard<std::mutex> lock(state_->queue_mutex);
        return {state_->queue.size(), state_->buffer_size};
    }
 
private:
    void drain_queue() {
        std::queue<log_entry> remaining;
        {
            std::lock_guard<std::mutex> lock(state_->queue_mutex);
            std::swap(remaining, state_->queue);
        }
 
        // Process remaining entries
        while (!remaining.empty()) {
            auto entry = std::move(remaining.front());
            remaining.pop();
            write_to_all(entry);
        }
    }
 
    void flush_writers() {
        std::vector<std::shared_ptr<log_writer_interface>> writers_snapshot;
        {
            std::lock_guard<std::mutex> lock(state_->writers_mutex);
            writers_snapshot.reserve(state_->writers.size());
            for (auto& weak_writer : state_->writers) {
                if (auto writer = weak_writer.lock()) {
                    writers_snapshot.push_back(writer);
                }
            }
        }
 
        for (auto& writer : writers_snapshot) {
            try {
                writer->flush();
            } catch (...) {
                // Swallow exceptions during flush
            }
        }
    }
 
    void write_to_all(const log_entry& entry) {
        // Snapshot writers under lock
        std::vector<std::shared_ptr<log_writer_interface>> writers_snapshot;
        {
            std::lock_guard<std::mutex> lock(state_->writers_mutex);
            writers_snapshot.reserve(state_->writers.size());
            for (auto& weak_writer : state_->writers) {
                if (auto writer = weak_writer.lock()) {
                    writers_snapshot.push_back(writer);
                }
            }
        }
 
        // Write to all writers without holding mutex
        for (auto& writer : writers_snapshot) {
            try {
                writer->write(entry);
            } catch (...) {
                // Swallow exceptions to prevent issues
            }
        }
    }
 
private:
    std::shared_ptr<log_collector_shared_state> state_;
    std::unique_ptr<log_collector_jthread_worker> worker_;
    std::atomic<std::uint64_t> dropped_messages_{0};
};
 
// log_collector public interface implementation
 
log_collector::log_collector(std::size_t buffer_size, std::size_t batch_size)
    : pimpl_(std::make_unique<impl>(buffer_size, batch_size)) {
}
 
log_collector::~log_collector() = default;
 
bool log_collector::enqueue(log_level level,
                           const std::string& message,
                           const std::string& file,
                           int line,
                           const std::string& function,
                           const std::chrono::system_clock::time_point& timestamp) {
    return pimpl_->enqueue(level, message, file, line, function, timestamp);
}
 
void log_collector::add_writer(std::shared_ptr<log_writer_interface> writer) {
    pimpl_->add_writer(writer);
}
 
void log_collector::clear_writers() {
    pimpl_->clear_writers();
}
 
void log_collector::start() {
    pimpl_->start();
}
 
void log_collector::stop() {
    pimpl_->stop();
}
 
void log_collector::flush() {
    pimpl_->flush();
}
 
std::pair<std::size_t, std::size_t> log_collector::get_queue_metrics() const {
    return pimpl_->get_queue_metrics();
}
 
} // namespace kcenon::logger