kcenon::logger::log_collector_jthread_worker Class Reference

Worker thread for log processing with jthread compatibility. More...

Collaboration diagram for kcenon::logger::log_collector_jthread_worker:

Public Member Functions
	log_collector_jthread_worker (std::shared_ptr< log_collector_shared_state > state)
	~log_collector_jthread_worker ()
void	start ()
void	stop ()
void	notify_work ()
bool	is_running () const noexcept

Static Private Member Functions
static void	worker_loop (std::shared_ptr< log_collector_shared_state > state, async::simple_stop_source &stop)
static void	write_to_all (const std::shared_ptr< log_collector_shared_state > &state, const log_entry &entry)

Private Attributes
std::shared_ptr< log_collector_shared_state >	state_
async::compat_jthread	thread_
std::atomic< bool >	running_

Detailed Description

Worker thread for log processing with jthread compatibility.

Uses std::jthread where available, falls back to std::thread with manual stop mechanism otherwise.

Definition at line 69 of file log_collector.cpp.

Constructor & Destructor Documentation

log_collector_jthread_worker()

kcenon::logger::log_collector_jthread_worker::log_collector_jthread_worker ( std::shared_ptr< log_collector_shared_state > state )

inlineexplicit

Definition at line 71 of file log_collector.cpp.

        : state_(std::move(state))
        , running_(false)
    {}

~log_collector_jthread_worker()

kcenon::logger::log_collector_jthread_worker::~log_collector_jthread_worker ( )

inline

Definition at line 76 of file log_collector.cpp.

                                    {
        stop();
    }

References stop().

Here is the call graph for this function:

Member Function Documentation

is_running()

bool kcenon::logger::log_collector_jthread_worker::is_running ( ) const

inlinenodiscardnoexcept

Definition at line 125 of file log_collector.cpp.

                                                   {
        return running_.load(std::memory_order_acquire);
    }

References running_.

notify_work()

void kcenon::logger::log_collector_jthread_worker::notify_work ( )

inline

Definition at line 119 of file log_collector.cpp.

                       {
        if (state_) {
            state_->queue_cv.notify_one();
        }
    }

References state_.

start()

void kcenon::logger::log_collector_jthread_worker::start ( )

inline

Definition at line 80 of file log_collector.cpp.

                 {
        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
    }

References running_, state_, stop(), thread_, and worker_loop().

Here is the call graph for this function:

stop()

void kcenon::logger::log_collector_jthread_worker::stop ( )

inline

Definition at line 102 of file log_collector.cpp.

                {
        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();
    }

References kcenon::logger::async::compat_jthread::join(), kcenon::logger::async::compat_jthread::request_stop(), running_, state_, and thread_.

Referenced by start(), worker_loop(), and ~log_collector_jthread_worker().

Here is the call graph for this function:

Here is the caller graph for this function:

worker_loop()

static void kcenon::logger::log_collector_jthread_worker::worker_loop ( std::shared_ptr< log_collector_shared_state > state, async::simple_stop_source & stop )

inlinestaticprivate

Definition at line 173 of file log_collector.cpp.

                                                          {
        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);
            }
        }
    }

References stop(), and write_to_all().

Referenced by start().

Here is the call graph for this function:

Here is the caller graph for this function:

write_to_all()

static void kcenon::logger::log_collector_jthread_worker::write_to_all ( const std::shared_ptr< log_collector_shared_state > & state, const log_entry & entry )

inlinestaticprivate

Definition at line 216 of file log_collector.cpp.

                                                    {
        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
            }
        }
    }

Referenced by worker_loop().

Here is the caller graph for this function:

Member Data Documentation

running_

std::atomic<bool> kcenon::logger::log_collector_jthread_worker::running_

private

Definition at line 247 of file log_collector.cpp.

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

state_

std::shared_ptr<log_collector_shared_state> kcenon::logger::log_collector_jthread_worker::state_

private

Definition at line 245 of file log_collector.cpp.

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

thread_

async::compat_jthread kcenon::logger::log_collector_jthread_worker::thread_

private

Definition at line 246 of file log_collector.cpp.

Referenced by start(), and stop().

---

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

• src/core/log_collector.cpp