A specialized worker thread that processes jobs from a job_queue. More...

#include <thread_worker.h>

Inheritance diagram for kcenon::thread::thread_worker:

Collaboration diagram for kcenon::thread::thread_worker:

Public Member Functions
	thread_worker (const bool &use_time_tag=true, const thread_context &context=thread_context())
	Constructs a new `thread_worker`.

virtual	~thread_worker (void)
	Virtual destructor. Ensures the worker thread is stopped before destruction.

auto	set_job_queue (std::shared_ptr< job_queue > job_queue) -> void
	Sets the `job_queue` that this worker should process.

auto	set_context (const thread_context &context) -> void
	Sets the thread context for this worker.

void	set_metrics (std::shared_ptr< metrics::ThreadPoolMetrics > metrics)
	Provide shared metrics storage for this worker.

void	set_diagnostics (diagnostics::thread_pool_diagnostics *diag)
	Set the diagnostics instance for event tracing.

void	set_diagnostics_sample_rate (std::uint32_t rate)
	Set the diagnostics sampling rate.

void	set_policy (const worker_policy &policy)
	Set the worker policy for this worker.

const worker_policy &	get_policy () const
	Get the current worker policy.

lockfree::work_stealing_deque< job * > *	get_local_deque () noexcept
	Get the local work-stealing deque for this worker.

void	set_steal_function (std::function< job *(std::size_t)> steal_fn)
	Set the steal function for finding other workers' deques.

std::size_t	get_worker_id () const
	Get the worker ID.

auto	get_context (void) const -> const thread_context &
	Gets the thread context for this worker.

bool	is_idle () const noexcept
	Checks if the worker is currently idle (not processing a job).

std::uint64_t	get_jobs_completed () const noexcept
	Gets the total number of jobs successfully completed by this worker.

std::uint64_t	get_jobs_failed () const noexcept
	Gets the total number of jobs that failed during execution.

std::chrono::nanoseconds	get_total_busy_time () const noexcept
	Gets the total time spent executing jobs (busy time).

std::chrono::nanoseconds	get_total_idle_time () const noexcept
	Gets the total time spent waiting for jobs (idle time).

std::chrono::steady_clock::time_point	get_state_since () const noexcept
	Gets the time when the worker entered its current state.

std::optional< diagnostics::job_info >	get_current_job_info () const noexcept
	Gets information about the currently executing job.

Public Member Functions inherited from kcenon::thread::thread_base
	thread_base (const thread_base &)=delete

thread_base &	operator= (const thread_base &)=delete

	thread_base (thread_base &&)=delete

thread_base &	operator= (thread_base &&)=delete

	thread_base (const std::string &thread_title="thread_base")
	Constructs a new `thread_base` object.

virtual	~thread_base (void)
	Virtual destructor. Ensures proper cleanup of derived classes.

auto	set_wake_interval (const std::optional< std::chrono::milliseconds > &wake_interval) -> void
	Sets the interval at which the worker thread should wake up (if any).

auto	get_wake_interval () const -> std::optional< std::chrono::milliseconds >
	Gets the current wake interval setting.

auto	start (void) -> common::VoidResult
	Starts the worker thread.

auto	stop (void) -> common::VoidResult
	Requests the worker thread to stop and waits for it to finish.

auto	get_thread_title () const -> std::string
	Returns the worker thread's title.

auto	is_running () const -> bool
	Checks whether the worker thread is currently running.

auto	get_thread_id () const -> std::thread::id
	Gets the native thread ID of the worker thread.

virtual auto	to_string (void) const -> std::string
	Returns a string representation of this `thread_base` object.

Protected Member Functions
auto	should_continue_work () const -> bool override
	Determines if there are jobs available in the queue to continue working on.

auto	do_work () -> common::VoidResult override
	Processes one or more jobs from the queue.

auto	on_stop_requested () -> void override
	Called when the worker is requested to stop.

Protected Member Functions inherited from kcenon::thread::thread_base
virtual auto	before_start (void) -> common::VoidResult
	Called just before the worker thread starts running.

virtual auto	after_stop (void) -> common::VoidResult
	Called immediately after the worker thread has stopped.

Private Member Functions
std::unique_ptr< job >	try_get_job ()
	Try to get a job from local deque first, then global queue.

std::unique_ptr< job >	try_steal_work ()
	Try to steal work from other workers.

Private Attributes
std::size_t	worker_id_ {0}
	Unique ID for this worker instance.

bool	use_time_tag_
	Indicates whether to use time tags or timestamps for job processing.

std::shared_ptr< job_queue >	job_queue_
	A shared pointer to the job queue from which this worker obtains jobs.

thread_context	context_
	The thread context providing access to logging and monitoring services.

std::shared_ptr< metrics::ThreadPoolMetrics >	metrics_
	Shared metrics aggregator provided by the owning thread pool.

diagnostics::thread_pool_diagnostics *	diagnostics_ {nullptr}
	Pointer to the diagnostics instance for event tracing.

cancellation_token	worker_cancellation_token_
	Cancellation token for this worker.

std::uint32_t	diagnostics_sample_rate_ {1}
	Diagnostics sampling rate (record every Nth job).

std::uint64_t	diagnostics_counter_ {0}
	Counter for diagnostics sampling.

std::atomic< job * >	current_job_ {nullptr}
	Pointer to the currently executing job.

std::atomic< bool >	is_idle_ {true}
	Indicates whether the worker is currently idle (not processing a job).

std::atomic< std::uint64_t >	jobs_completed_ {0}
	Total number of jobs successfully completed by this worker.

std::atomic< std::uint64_t >	jobs_failed_ {0}
	Total number of jobs that failed during execution.

std::atomic< std::uint64_t >	total_busy_time_ns_ {0}
	Total time spent executing jobs (busy time) in nanoseconds.

std::atomic< std::uint64_t >	total_idle_time_ns_ {0}
	Total time spent waiting for jobs (idle time) in nanoseconds.

std::atomic< std::chrono::steady_clock::time_point::rep >	state_since_rep_
	Time point when the worker entered its current state.

std::chrono::steady_clock::time_point	current_job_start_time_
	Time point when the current job started executing.

std::mutex	queue_mutex_
	Mutex protecting job queue replacement.

std::condition_variable	queue_cv_
	Condition variable for queue replacement synchronization.

bool	queue_being_replaced_ {false}
	Indicates whether queue replacement is in progress.

worker_policy	policy_
	Worker policy configuration.

std::unique_ptr< lockfree::work_stealing_deque< job * > >	local_deque_
	Local work-stealing deque for this worker.

std::function< job *(std::size_t)>	steal_function_
	Function to steal work from other workers.

std::size_t	steal_victim_index_ {0}
	Counter for round-robin steal victim selection.

Static Private Attributes
static std::atomic< std::size_t >	next_worker_id_ {0}
	Static counter for generating unique worker IDs.

Additional Inherited Members
Protected Attributes inherited from kcenon::thread::thread_base
std::optional< std::chrono::milliseconds >	wake_interval_
	Interval at which the thread is optionally awakened.

Detailed Description

A specialized worker thread that processes jobs from a job_queue.

The thread_worker class inherits from thread_base, leveraging its life-cycle control methods (start, stop, etc.) and provides an implementation for job processing using a shared job_queue. By overriding should_continue_work() and do_work(), it polls the queue for available jobs and executes them.

Typical Usage

auto my_queue = std::make_shared<job_queue>();
auto worker   = std::make_unique<thread_worker>();
worker->set_job_queue(my_queue);
worker->start(); // Worker thread begins processing jobs
 
// Enqueue jobs into my_queue...
 
// Eventually...
worker->stop();  // Waits for current job to finish, then stops

Examples: crash_protection/main.cpp.

Definition at line 67 of file thread_worker.h.

Constructor & Destructor Documentation

◆ thread_worker()

kcenon::thread::thread_worker::thread_worker	(	const bool &	use_time_tag = true,
		const thread_context &	context = thread_context() )

Constructs a new thread_worker.

Constructs a worker thread with optional timing capabilities.

Parameters

use_time_tag	If set to `true` (default), the worker may log or utilize timestamps/tags when processing jobs.
context	Optional thread context for logging and monitoring (defaults to empty context).

This flag can be used to measure job durations, implement logging with timestamps, or any other time-related features in your job processing. The context provides access to logging and monitoring services.

Implementation details:

Inherits from thread_base to get thread management functionality
Sets descriptive name "thread_worker" for debugging and logging
Initializes timing flag for optional performance measurement
Job queue is not set initially (must be set before starting work)
Stores thread context for logging and monitoring
Creates a cancellation token for job cancellation support

Performance Timing:

When enabled, measures execution time for each job
Uses high_resolution_clock for precise measurements
Minimal overhead when disabled (single boolean check)

Parameters

use_time_tag	If true, enables timing measurements for job execution
context	Thread context providing logging and monitoring services

Definition at line 70 of file thread_worker.cpp.

        : thread_base("thread_worker"),
          worker_id_(next_worker_id_.fetch_add(1)),
          use_time_tag_(use_time_tag),
          job_queue_(nullptr),
          context_(context),
          worker_cancellation_token_(cancellation_token::create()),
          current_job_(nullptr)
    {
    }

◆ ~thread_worker()

kcenon::thread::thread_worker::~thread_worker ( void )

virtual

Virtual destructor. Ensures the worker thread is stopped before destruction.

Destroys the worker thread.

Implementation details:

Base class destructor handles thread shutdown
No manual cleanup required due to RAII design
Shared pointer to job queue is automatically released

Definition at line 89 of file thread_worker.cpp.

89{}

Member Function Documentation

◆ do_work()

auto kcenon::thread::thread_worker::do_work ( void ) -> common::VoidResult

overrideprotectedvirtual

Processes one or more jobs from the queue.

Executes a single work cycle by processing one job from the queue.

Returns: common::VoidResult containing an error if the work fails, or success value otherwise.

This method fetches a job from the queue (if available), executes it, and may repeat depending on the implementation. If any job fails, an error is returned. Otherwise, return a success value.

Implementation details:

Uses non-blocking try_dequeue() to avoid condition variable deadlock
Polls the queue with minimal CPU overhead via short sleep intervals
Validates job pointer before execution
Optionally measures execution timing for performance analysis
Associates job with queue for potential re-submission
Logs execution results with appropriate detail level

Job Processing Workflow:

Validate job queue availability
Attempt non-blocking dequeue of next job
If queue is empty: sleep briefly and return (will be called again)
Validate dequeued job pointer
Optionally record start time for measurement
Associate job with queue for context
Execute job's do_work() method
Handle execution errors with detailed logging
Log successful completion with timing info if enabled

Hybrid Wait Strategy:

Phase 1: Bounded spin (16 iterations) with CPU pause hints for fast pickup
Phase 2: Blocking dequeue() on job_queue's condition_variable
Wakes immediately on enqueue (notify_one) or stop (notify_all)
should_continue_work() determines when to exit (on queue stop)

Performance Characteristics:

Sub-microsecond pickup when jobs arrive during spin phase
Near-zero idle CPU usage (CV blocking, not polling)
Immediate wake-up on enqueue (<100μs vs previous 10ms floor)
No busy-waiting overhead when idle

Error Handling:

Missing job queue: Returns resource allocation error
Empty queue: Returns success after brief sleep (normal polling)
Null job pointer: Returns job invalid error
Job execution failure: Returns execution failed error with details

Performance Measurements:

High-resolution timing when use_time_tag_ is enabled
Nanosecond precision for accurate profiling
Minimal overhead when timing is disabled

Logging Behavior:

Standard success message when timing is disabled
Timestamped success message when timing is enabled
Error details are propagated up the call stack

Returns: result_void indicating success or detailed error information

Reimplemented from kcenon::thread::thread_base.

Definition at line 383 of file thread_worker.cpp.

    {
        // Acquire lock to safely get queue pointer
        std::unique_lock<std::mutex> lock(queue_mutex_);
 
        // Validate that job queue is available for processing
        if (job_queue_ == nullptr)
        {
            lock.unlock();
            return common::error_info{static_cast<int>(error_code::resource_allocation_failed), "there is no job_queue", "thread_system"};
        }
 
        // Make a local copy of the queue pointer while holding the lock
        // The shared_ptr keeps the queue alive even if set_job_queue() replaces it
        // No need to wait for !queue_being_replaced_ - the local copy is safe to use
        std::shared_ptr<job_queue> local_queue = job_queue_;
 
        // Release lock before dequeuing to allow other operations
        lock.unlock();
 
        std::unique_ptr<job> current_job;
 
        // Work-stealing path: try local deque first, then global queue, then steal
        if (policy_.enable_work_stealing)
        {
            // Step 1: Try local deque (LIFO for cache locality)
            if (local_deque_)
            {
                auto local_result = local_deque_->pop();
                if (local_result.has_value())
                {
                    current_job = std::unique_ptr<job>(*local_result);
                }
            }
 
            // Step 2: Try global queue
            if (!current_job)
            {
                auto dequeue_result = local_queue->try_dequeue();
                if (dequeue_result.is_ok())
                {
                    current_job = std::move(dequeue_result.value());
                }
            }
 
            // Step 3: Try to steal from other workers
            if (!current_job)
            {
                current_job = try_steal_work();
            }
 
            // No work found - mark as idle and sleep
            if (!current_job)
            {
                is_idle_.store(true, std::memory_order_relaxed);
                std::this_thread::sleep_for(policy_.idle_sleep_duration);
                return common::ok();
            }
        }
        else
        {
            // Original non-work-stealing path
            // Hybrid wait strategy: short spin followed by blocking dequeue
            // This approach provides:
            // - Sub-ms pickup latency (via spin loop)
            // - Near-zero idle CPU usage (via blocking dequeue)
            // - No busy-waiting overhead
 
            // Phase 1: Short bounded spin (16 iterations)
            // Optimized for scenarios where jobs arrive quickly
            constexpr int spin_count = 16;
            for (int i = 0; i < spin_count; ++i)
            {
                auto dequeue_result = local_queue->try_dequeue();
                if (dequeue_result.is_ok())
                {
                    // Job found during spin - fast path
                    current_job = std::move(dequeue_result.value());
                    break;
                }
 
                // CPU pause hint to reduce contention and power consumption
                // Different intrinsics per compiler and architecture
                #if defined(_MSC_VER)
                    // MSVC: Use _mm_pause() for x86/x64, YieldProcessor() for ARM
                    #if defined(_M_IX86) || defined(_M_X64)
                        _mm_pause();
                    #elif defined(_M_ARM) || defined(_M_ARM64)
                        __yield();
                    #else
                        std::this_thread::yield();
                    #endif
                #elif defined(__GNUC__) || defined(__clang__)
                    // GCC/Clang: Use builtin functions
                    #if defined(__x86_64__) || defined(__i386__)
                        __builtin_ia32_pause();
                    #elif defined(__aarch64__) || defined(__arm__)
                        __asm__ __volatile__("yield");
                    #else
                        std::this_thread::yield();
                    #endif
                #else
                    std::this_thread::yield();
                #endif
            }
 
            // Phase 2: Block on job_queue's condition variable if spin didn't find a job
            // Uses the queue's blocking dequeue() which waits on its internal CV.
            // This wakes immediately when a new job is enqueued (via notify_one)
            // or when the queue is stopped (via notify_all in job_queue::stop).
            if (!current_job)
            {
                is_idle_.store(true, std::memory_order_relaxed);
 
                // Blocking dequeue: waits on job_queue's condition_variable
                // Wakes on: enqueue() notify_one, or stop() notify_all
                auto dequeue_result = local_queue->dequeue();
                if (dequeue_result.is_ok())
                {
                    current_job = std::move(dequeue_result.value());
                }
                else
                {
                    // Queue is stopped or empty after wake — return to let
                    // should_continue_work() decide whether to exit
                    return common::ok();
                }
            }
        }
 
        // Validate job pointer
        if (current_job == nullptr)
        {
            return common::error_info{static_cast<int>(error_code::job_invalid), "error executing job: nullptr", "thread_system"};
        }
 
        // Capture job info for event tracing (before any state changes)
        const auto job_id = current_job->get_job_id();
        const auto job_name_for_event = current_job->get_name();
        const auto enqueue_time = current_job->get_enqueue_time();
 
        // Determine whether this job should record diagnostics events.
        // The sampling counter reduces clock-read overhead by skipping
        // events for most jobs when sample_rate > 1.
        const bool should_trace = diagnostics_
            && diagnostics_->is_tracing_enabled()
            && (++diagnostics_counter_ % diagnostics_sample_rate_ == 0);
 
        // Record dequeued event if tracing is enabled and sampled
        if (should_trace)
        {
            diagnostics::job_execution_event dequeued_event;
            dequeued_event.job_id = job_id;
            dequeued_event.job_name = job_name_for_event;
            dequeued_event.type = diagnostics::event_type::dequeued;
            dequeued_event.timestamp = std::chrono::steady_clock::now();
            dequeued_event.system_timestamp = std::chrono::system_clock::now();
            dequeued_event.thread_id = std::this_thread::get_id();
            dequeued_event.worker_id = worker_id_;
            dequeued_event.wait_time = std::chrono::duration_cast<std::chrono::nanoseconds>(
                dequeued_event.timestamp - enqueue_time);
            diagnostics_->record_event(dequeued_event);
        }
 
        // Update idle time statistics before transitioning to busy state
        auto now = std::chrono::steady_clock::now();
        auto state_since = get_state_since();
        if (is_idle_.load(std::memory_order_relaxed))
        {
            auto idle_duration = std::chrono::duration_cast<std::chrono::nanoseconds>(now - state_since);
            total_idle_time_ns_.fetch_add(
                static_cast<std::uint64_t>(idle_duration.count()),
                std::memory_order_relaxed
            );
        }
 
        // Mark worker as busy and update state timestamp
        is_idle_.store(false, std::memory_order_relaxed);
        state_since_rep_.store(now.time_since_epoch().count(), std::memory_order_release);
        current_job_start_time_ = now;
 
        // Initialize timing measurement if performance monitoring is enabled
        std::optional<std::chrono::time_point<std::chrono::high_resolution_clock>>
            started_time_point = std::nullopt;
        if (use_time_tag_)
        {
            started_time_point = std::chrono::high_resolution_clock::now();
        }
 
        // Associate the job with its source queue for potential re-submission
        // Use local_queue to avoid race with set_job_queue()
        current_job->set_job_queue(local_queue);
 
        // Set cancellation token on the job for cooperative cancellation
        // This allows the job to check if it should cancel during execution
        current_job->set_cancellation_token(worker_cancellation_token_);
 
        // Track currently executing job atomically for on_stop_requested()
        // Use release ordering to ensure job state is visible to cancellation thread
        current_job_.store(current_job.get(), std::memory_order_release);
 
        // Record started event if tracing is enabled and sampled
        if (should_trace)
        {
            diagnostics::job_execution_event started_event;
            started_event.job_id = job_id;
            started_event.job_name = job_name_for_event;
            started_event.type = diagnostics::event_type::started;
            started_event.timestamp = std::chrono::steady_clock::now();
            started_event.system_timestamp = std::chrono::system_clock::now();
            started_event.thread_id = std::this_thread::get_id();
            started_event.worker_id = worker_id_;
            started_event.wait_time = std::chrono::duration_cast<std::chrono::nanoseconds>(
                started_event.timestamp - enqueue_time);
            diagnostics_->record_event(started_event);
        }
 
        // Execute the job's work method and capture the result
        auto work_result = current_job->do_work();
        std::uint64_t execution_duration_ns = 0;
        if (started_time_point.has_value())
        {
            auto end_time = std::chrono::high_resolution_clock::now();
            execution_duration_ns = static_cast<std::uint64_t>(
                std::chrono::duration_cast<std::chrono::nanoseconds>(
                    end_time - started_time_point.value()).count());
        }
 
        // Capture job name before clearing (for logging after job destruction)
        std::string job_name = current_job->get_name();
 
        // Clear current job tracking and destroy job under mutex protection
        // This prevents race condition with on_stop_requested() (Issue #225)
        // The mutex ensures job is not accessed while being destroyed
        {
            std::lock_guard<std::mutex> notify_lock(queue_mutex_);
            current_job_.store(nullptr, std::memory_order_release);
 
            // Explicitly destroy the job while holding the mutex
            // This is critical for thread safety - on_stop_requested() acquires
            // the same mutex before accessing current_job_, ensuring it cannot
            // access a job that is being destroyed
            current_job.reset();
 
            // Notify any waiting set_job_queue() that job has completed
            // This allows queue replacement to proceed safely
            // Lock is held to prevent lost wakeup between predicate check and wait
            queue_cv_.notify_all();
        }
 
        // Update busy time and state transition for diagnostics
        {
            auto end_now = std::chrono::steady_clock::now();
            auto busy_duration = std::chrono::duration_cast<std::chrono::nanoseconds>(
                end_now - current_job_start_time_
            );
            total_busy_time_ns_.fetch_add(
                static_cast<std::uint64_t>(busy_duration.count()),
                std::memory_order_relaxed
            );
            // Transition back to idle state
            is_idle_.store(true, std::memory_order_relaxed);
            state_since_rep_.store(end_now.time_since_epoch().count(), std::memory_order_release);
        }
 
        if (work_result.is_err())
        {
            // Increment failed job counter
            jobs_failed_.fetch_add(1, std::memory_order_relaxed);
 
            // Record failed event if tracing is enabled and sampled
            if (should_trace)
            {
                diagnostics::job_execution_event failed_event;
                failed_event.job_id = job_id;
                failed_event.job_name = job_name_for_event;
                failed_event.type = diagnostics::event_type::failed;
                failed_event.timestamp = std::chrono::steady_clock::now();
                failed_event.system_timestamp = std::chrono::system_clock::now();
                failed_event.thread_id = std::this_thread::get_id();
                failed_event.worker_id = worker_id_;
                failed_event.wait_time = std::chrono::duration_cast<std::chrono::nanoseconds>(
                    current_job_start_time_ - enqueue_time);
                failed_event.execution_time = std::chrono::nanoseconds(execution_duration_ns);
                failed_event.error_code = work_result.error().code;
                failed_event.error_message = work_result.error().message;
                diagnostics_->record_event(failed_event);
            }
 
            if (metrics_)
            {
                metrics_->record_execution(0, false);
            }
            return common::error_info{static_cast<int>(error_code::job_execution_failed),
                formatter::format("error executing job: {}", work_result.error().message), "thread_system"};
        }
 
        // Increment completed job counter
        jobs_completed_.fetch_add(1, std::memory_order_relaxed);
 
        // Record completed event if tracing is enabled and sampled
        if (should_trace)
        {
            diagnostics::job_execution_event completed_event;
            completed_event.job_id = job_id;
            completed_event.job_name = job_name_for_event;
            completed_event.type = diagnostics::event_type::completed;
            completed_event.timestamp = std::chrono::steady_clock::now();
            completed_event.system_timestamp = std::chrono::system_clock::now();
            completed_event.thread_id = std::this_thread::get_id();
            completed_event.worker_id = worker_id_;
            completed_event.wait_time = std::chrono::duration_cast<std::chrono::nanoseconds>(
                current_job_start_time_ - enqueue_time);
            completed_event.execution_time = std::chrono::nanoseconds(execution_duration_ns);
            diagnostics_->record_event(completed_event);
        }
 
        // Log successful job completion based on timing configuration
        // Note: Using captured job_name since job is already destroyed
        if (!started_time_point.has_value())
        {
            // Standard logging without timing information
            context_.log(common::interfaces::log_level::debug,
                        formatter::format("job executed successfully: {} on thread_worker",
                                        job_name));
        }
        else
        {
            // Enhanced logging with execution timing information
            context_.log(common::interfaces::log_level::debug,
                        formatter::format("job executed successfully: {} on thread_worker ({}ns)",
                                        job_name, execution_duration_ns));
 
            // Update worker metrics if monitoring is available
            if (context_.monitoring())
            {
                common::interfaces::worker_metrics metrics(worker_id_);
                metrics.jobs_processed.value = 1;
                metrics.total_processing_time_ns.value = static_cast<double>(execution_duration_ns);
                // Use proper worker ID instead of thread hash
                context_.update_worker_metrics(worker_id_, metrics);
            }
        }
 
        if (metrics_)
        {
            metrics_->record_execution(execution_duration_ns, true);
        }
 
        return common::ok();
    }

Here is the call graph for this function:

◆ get_context()

auto kcenon::thread::thread_worker::get_context ( void ) const -> const thread_context&

nodiscard

Gets the thread context for this worker.

Returns: The thread context providing access to logging and monitoring services.; The thread context providing access to logging and monitoring services

Definition at line 269 of file thread_worker.cpp.

    {
        return context_;
    }

◆ get_current_job_info()

std::optional< diagnostics::job_info > kcenon::thread::thread_worker::get_current_job_info ( ) const

nodiscardnoexcept

Gets information about the currently executing job.

Returns: Optional job_info if a job is currently executing, std::nullopt otherwise.

Thread Safety:

Safe to call from any thread
Provides snapshot of current state

Definition at line 842 of file thread_worker.cpp.

    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        auto* job_ptr = current_job_.load(std::memory_order_acquire);
        if (job_ptr == nullptr)
        {
            return std::nullopt;
        }
 
        diagnostics::job_info info;
        info.job_id = job_ptr->get_job_id();
        info.job_name = job_ptr->get_name();
        info.status = diagnostics::job_status::running;
        info.start_time = current_job_start_time_;
        info.enqueue_time = job_ptr->get_enqueue_time();
        info.executed_by = std::this_thread::get_id();
 
        auto now = std::chrono::steady_clock::now();
        info.execution_time = std::chrono::duration_cast<std::chrono::nanoseconds>(
            now - current_job_start_time_
        );
        // Calculate wait time from enqueue to start
        info.wait_time = std::chrono::duration_cast<std::chrono::nanoseconds>(
            current_job_start_time_ - job_ptr->get_enqueue_time()
        );
 
        return info;
    }

References current_job_, current_job_start_time_, kcenon::thread::info, kcenon::thread::diagnostics::job_info::job_id, queue_mutex_, and kcenon::thread::diagnostics::running.

◆ get_jobs_completed()

std::uint64_t kcenon::thread::thread_worker::get_jobs_completed ( ) const

nodiscardnoexcept

Gets the total number of jobs successfully completed by this worker.

Returns: Count of successfully completed jobs.

Thread Safety:

Safe to call from any thread
Uses atomic load with relaxed memory ordering

Definition at line 814 of file thread_worker.cpp.

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

References jobs_completed_.

◆ get_jobs_failed()

std::uint64_t kcenon::thread::thread_worker::get_jobs_failed ( ) const

nodiscardnoexcept

Gets the total number of jobs that failed during execution.

Returns: Count of failed jobs.

Thread Safety:

Safe to call from any thread
Uses atomic load with relaxed memory ordering

Definition at line 819 of file thread_worker.cpp.

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

References jobs_failed_.

◆ get_local_deque()

lockfree::work_stealing_deque< job * > * kcenon::thread::thread_worker::get_local_deque ( )

nodiscardnoexcept

Get the local work-stealing deque for this worker.

Returns: Pointer to the local deque (nullptr if work-stealing disabled).

This deque is used for work-stealing: other workers can steal jobs from this worker's local deque when they are idle.

Definition at line 199 of file thread_worker.cpp.

{
    return local_deque_.get();
}

References local_deque_.

◆ get_policy()

const worker_policy & kcenon::thread::thread_worker::get_policy ( ) const

nodiscard

Get the current worker policy.

Returns: The worker policy configuration.

Definition at line 194 of file thread_worker.cpp.

{
    return policy_;
}

References policy_.

◆ get_state_since()

std::chrono::steady_clock::time_point kcenon::thread::thread_worker::get_state_since ( ) const

nodiscardnoexcept

Gets the time when the worker entered its current state.

Returns: Time point when current state was entered.

Thread Safety:

Safe to call from any thread
Uses atomic load with acquire memory ordering

Definition at line 834 of file thread_worker.cpp.

    {
        auto rep = state_since_rep_.load(std::memory_order_acquire);
        return std::chrono::steady_clock::time_point{
            std::chrono::steady_clock::duration{rep}
        };
    }

References state_since_rep_.

◆ get_total_busy_time()

std::chrono::nanoseconds kcenon::thread::thread_worker::get_total_busy_time ( ) const

nodiscardnoexcept

Gets the total time spent executing jobs (busy time).

Returns: Duration of busy time in nanoseconds.

Thread Safety:

Safe to call from any thread
Uses atomic load with relaxed memory ordering

Definition at line 824 of file thread_worker.cpp.

    {
        return std::chrono::nanoseconds{total_busy_time_ns_.load(std::memory_order_relaxed)};
    }

References total_busy_time_ns_.

◆ get_total_idle_time()

std::chrono::nanoseconds kcenon::thread::thread_worker::get_total_idle_time ( ) const

nodiscardnoexcept

Gets the total time spent waiting for jobs (idle time).

Returns: Duration of idle time in nanoseconds.

Thread Safety:

Safe to call from any thread
Uses atomic load with relaxed memory ordering

Definition at line 829 of file thread_worker.cpp.

    {
        return std::chrono::nanoseconds{total_idle_time_ns_.load(std::memory_order_relaxed)};
    }

References total_idle_time_ns_.

◆ get_worker_id()

std::size_t kcenon::thread::thread_worker::get_worker_id ( ) const

nodiscard

Get the worker ID.

Returns: The unique ID for this worker instance.

Definition at line 735 of file thread_worker.cpp.

    {
        return worker_id_;
    }

References worker_id_.

◆ is_idle()

bool kcenon::thread::thread_worker::is_idle ( ) const

nodiscardnoexcept

Checks if the worker is currently idle (not processing a job).

Returns: true if the worker is idle (waiting for jobs), false if actively processing.

Thread Safety:

Safe to call from any thread
Uses atomic operations for lock-free access
Provides snapshot of current state (may change immediately after return)

Use Cases:

Thread pool statistics and monitoring
Load balancing decisions
Performance analysis

Implementation details:

Returns current state of is_idle_ flag
Relaxed memory ordering sufficient (advisory-only value)
No synchronization needed (snapshot of current state)

Returns: true if worker is idle, false if actively processing a job

Definition at line 750 of file thread_worker.cpp.

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

References is_idle_.

◆ on_stop_requested()

auto kcenon::thread::thread_worker::on_stop_requested ( void ) -> void

overrideprotectedvirtual

Called when the worker is requested to stop.

Propagates cancellation signal to the currently executing job.

Overrides the base class hook to propagate cancellation to the currently executing job (if any). This allows jobs to cooperatively cancel when the worker thread is stopped.

Thread Safety:

Called from thread requesting stop (not worker thread)
Safe concurrent access with do_work() via atomic operations

Implementation details:

Called from thread_base::stop() when worker shutdown is requested
First cancels the worker's cancellation token (affects all future jobs)
Then directly cancels the current job's token if a job is running
Uses mutex synchronization to prevent use-after-free race condition

Cancellation Propagation:

Cancel worker_cancellation_token_ (prevents new jobs from starting)
Acquire queue_mutex_ to safely access current job
If a job is running, get its cancellation token and cancel it
Job will detect cancellation on its next is_cancelled() check

Thread Safety (Issue #225 fix):

Called from stop() thread (not worker thread)
Uses queue_mutex_ to synchronize with do_work() job destruction
Prevents data race between job destructor and virtual method call
This fixes EXC_BAD_ACCESS on ARM64 caused by use-after-free

Race Condition Fixed:

Before: on_stop_requested() could access job while do_work() was destroying it
After: queue_mutex_ ensures job is not destroyed while being accessed
do_work() now destroys job while holding the mutex

Note: This implements cooperative cancellation - the job must check its cancellation token periodically to actually stop execution.

Reimplemented from kcenon::thread::thread_base.

Definition at line 784 of file thread_worker.cpp.

    {
        // Cancel the worker's token first
        // This ensures any future jobs will see cancellation immediately
        // Jobs receive this token via set_cancellation_token() in do_work()
        worker_cancellation_token_.cancel();
 
        // Acquire mutex to safely access current job
        // This prevents race condition with do_work() job destruction (Issue #225)
        std::lock_guard<std::mutex> lock(queue_mutex_);
 
        // Load the currently executing job pointer atomically
        auto* job_ptr = current_job_.load(std::memory_order_acquire);
 
        // If a job is currently executing, cancel it directly
        if (job_ptr != nullptr)
        {
            // Get the job's cancellation token and cancel it
            // This provides redundancy in case the job cached its token before
            // we cancelled worker_cancellation_token_
            auto job_token = job_ptr->get_cancellation_token();
            job_token.cancel();
 
            // Log cancellation attempt for debugging
            context_.log(common::interfaces::log_level::debug,
                        formatter::format("Cancellation requested for job: {} on worker {}",
                                        job_ptr->get_name(), worker_id_));
        }
    }

References kcenon::thread::utils::formatter::format().

Here is the call graph for this function:

◆ set_context()

auto kcenon::thread::thread_worker::set_context ( const thread_context & context ) -> void

Sets the thread context for this worker.

Parameters

context The thread context providing access to logging and monitoring services.

Implementation details:

Stores the context for use in logging and monitoring
Should be called before starting the worker thread
Context provides access to optional services

Parameters

context Thread context with logging and monitoring services

Definition at line 163 of file thread_worker.cpp.

{
    context_ = context;
}

◆ set_diagnostics()

void kcenon::thread::thread_worker::set_diagnostics ( diagnostics::thread_pool_diagnostics * diag )

Set the diagnostics instance for event tracing.

Parameters

diag	Pointer to the diagnostics instance.

When set, the worker will record execution events to the diagnostics instance if tracing is enabled. If nullptr, no events are recorded.

Definition at line 173 of file thread_worker.cpp.

{
    diagnostics_ = diag;
}

References diagnostics_.

◆ set_diagnostics_sample_rate()

void kcenon::thread::thread_worker::set_diagnostics_sample_rate ( std::uint32_t rate )

Set the diagnostics sampling rate.

Parameters

rate	Record diagnostics events every Nth job (1 = every job).

When rate > 1, only every Nth job records diagnostic events, reducing clock-read overhead while still providing representative data. The is_tracing_enabled() check remains the top-level gate.

Definition at line 178 of file thread_worker.cpp.

{
    diagnostics_sample_rate_ = (rate > 0) ? rate : 1;
}

References diagnostics_sample_rate_.

◆ set_job_queue()

auto kcenon::thread::thread_worker::set_job_queue ( std::shared_ptr< job_queue > job_queue ) -> void

Sets the job_queue that this worker should process.

Associates this worker with a job queue for processing.

Parameters

job_queue A shared pointer to the queue containing jobs.

Once the queue is set and start() is called, the worker will repeatedly poll the queue for new jobs and process them.

Implementation details:

Stores shared pointer to enable job dequeuing
Thread-safe queue replacement with proper synchronization
Waits for current job completion before replacing queue
Multiple workers can share the same job queue

Queue Replacement Synchronization:

Acquires mutex to prevent concurrent do_work() access
Sets replacement flag to prevent new job processing
Waits for current job to complete (current_job_ == nullptr)
Replaces queue pointer atomically
Notifies worker thread to resume

Thread Safety:

Safe to call from any thread
Coordinates with do_work() via mutex and condition variable
Prevents use-after-free during queue replacement

Parameters

job_queue Shared pointer to the job queue for this worker

Definition at line 114 of file thread_worker.cpp.

    {
        std::shared_ptr<kcenon::thread::job_queue> old_queue;
 
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
 
            // Signal that queue replacement is in progress
            queue_being_replaced_ = true;
 
            // Wait for current job to complete
            // Predicate ensures we don't proceed while a job is executing
            queue_cv_.wait(lock, [this] {
                return current_job_.load(std::memory_order_acquire) == nullptr;
            });
 
            // Save old queue so we can wake any blocked dequeue() calls
            old_queue = job_queue_;
 
            // Replace the queue pointer
            job_queue_ = std::move(job_queue);
 
            // Clear replacement flag
            queue_being_replaced_ = false;
 
            // Notify worker thread that replacement is complete
            queue_cv_.notify_all();
        }
 
        // Stop the old queue outside the lock to wake any worker thread
        // blocked on old_queue->dequeue(). Without this, the worker's
        // do_work() would block indefinitely on the old queue's CV after
        // the queue pointer has been replaced.
        if (old_queue)
        {
            old_queue->stop();
        }
    }

◆ set_metrics()

void kcenon::thread::thread_worker::set_metrics ( std::shared_ptr< metrics::ThreadPoolMetrics > metrics )

Provide shared metrics storage for this worker.

Definition at line 168 of file thread_worker.cpp.

{
    metrics_ = std::move(metrics);
}

References metrics_.

◆ set_policy()

void kcenon::thread::thread_worker::set_policy ( const worker_policy & policy )

Set the worker policy for this worker.

Parameters

policy The worker policy configuration.

Definition at line 183 of file thread_worker.cpp.

{
    policy_ = policy;
 
    // Initialize local deque if work-stealing is enabled
    if (policy_.enable_work_stealing && !local_deque_)
    {
        local_deque_ = std::make_unique<lockfree::work_stealing_deque<job*>>();
    }
}

References kcenon::thread::worker_policy::enable_work_stealing, local_deque_, and policy_.

◆ set_steal_function()

void kcenon::thread::thread_worker::set_steal_function ( std::function< job *(std::size_t)> steal_fn )

Set the steal function for finding other workers' deques.

Parameters

steal_fn Function that returns a job to steal, or nullptr.

The steal function is called when this worker's local deque and the global queue are both empty. It should try to steal work from other workers.

Definition at line 204 of file thread_worker.cpp.

{
    steal_function_ = std::move(steal_fn);
}

References steal_function_.

◆ should_continue_work()

auto kcenon::thread::thread_worker::should_continue_work ( void ) const -> bool

nodiscardoverrideprotectedvirtual

Determines if there are jobs available in the queue to continue working on.

Determines if the worker should continue processing jobs.

Returns: true if there is work in the queue, false otherwise.

Called in the thread's main loop (defined by thread_base) to decide if do_work() should be invoked. Returns true if the job queue is not empty; otherwise, false.

Implementation details:

Used by thread_base to control the work loop
Returns false if no job queue is set (prevents infinite loop)
Returns true while queue is not stopped (even if empty)
Actual job waiting is handled by non-blocking polling in do_work()
Thread-safe operation (job_queue methods are thread-safe)

Work Loop Control:

Worker continues until queue is explicitly stopped
Empty queue does NOT cause worker exit - do_work() will poll for jobs
This prevents premature worker termination before jobs arrive
Worker exits gracefully only when queue is stopped

Design Rationale - Solving the Two-Level Condition Variable Problem:

Thread_base waits on its own condition variable (worker_condition_)
Job_queue notifies its own condition variable (different object!)
If should_continue_work() returns false on empty queue:
1. thread_base waits on worker_condition_
2. job enqueue notifies job_queue's condition variable
3. Worker never wakes up - deadlock situation
By returning true until stopped, worker enters do_work() immediately
do_work() uses non-blocking try_dequeue() with polling
This completely avoids the two-level CV problem
CPU overhead is minimal due to sleep between polling attempts

Shutdown Safety:

thread_pool::stop() can call operations in any order without race conditions
Queue stop sets is_stopped() = true (atomic operation)
Worker sees stopped flag and exits cleanly
No dependency on operation ordering

Thread Safety:

Synchronizes access to job_queue_ with queue_mutex_
Prevents race conditions with set_job_queue() and do_work()
Uses lock_guard for RAII-based exception safety
Mutex marked mutable to allow locking in const method

Returns: true if worker should continue processing, false to exit

Reimplemented from kcenon::thread::thread_base.

Definition at line 317 of file thread_worker.cpp.

    {
        // Synchronize access to job_queue_ with set_job_queue() and do_work()
        std::lock_guard<std::mutex> lock(queue_mutex_);
 
        if (job_queue_ == nullptr)
        {
            return false;
        }
 
        // Continue while queue is not stopped - do_work() handles polling for jobs
        return !job_queue_->is_stopped();
    }

References job_queue_, and queue_mutex_.

◆ try_get_job()

std::unique_ptr< job > kcenon::thread::thread_worker::try_get_job ( )

nodiscardprivate

Try to get a job from local deque first, then global queue.

Returns: A unique_ptr to the job, or nullptr if no work available.

Definition at line 209 of file thread_worker.cpp.

{
    // If work-stealing is enabled, try local deque first (LIFO for cache locality)
    if (policy_.enable_work_stealing && local_deque_)
    {
        auto local_result = local_deque_->pop();
        if (local_result.has_value())
        {
            return std::unique_ptr<job>(*local_result);
        }
    }
 
    // Fallback to global queue
    std::lock_guard<std::mutex> lock(queue_mutex_);
    if (job_queue_ == nullptr)
    {
        return nullptr;
    }
 
    auto dequeue_result = job_queue_->try_dequeue();
    if (dequeue_result.is_ok())
    {
        return std::move(dequeue_result.value());
    }
 
    return nullptr;
}

References kcenon::thread::worker_policy::enable_work_stealing, job_queue_, local_deque_, policy_, and queue_mutex_.

◆ try_steal_work()

std::unique_ptr< job > kcenon::thread::thread_worker::try_steal_work ( )

nodiscardprivate

Try to steal work from other workers.

Returns: A unique_ptr to the stolen job, or nullptr if stealing failed.

Definition at line 237 of file thread_worker.cpp.

{
    if (!policy_.enable_work_stealing || !steal_function_)
    {
        return nullptr;
    }
 
    // Apply exponential backoff between steal attempts
    for (std::size_t attempt = 0; attempt < policy_.max_steal_attempts; ++attempt)
    {
        job* stolen = steal_function_(worker_id_);
        if (stolen != nullptr)
        {
            return std::unique_ptr<job>(stolen);
        }
 
        // Exponential backoff
        if (attempt > 0)
        {
            auto backoff = policy_.steal_backoff * (1 << (attempt - 1));
            std::this_thread::sleep_for(backoff);
        }
    }
 
    return nullptr;
}

References kcenon::thread::worker_policy::enable_work_stealing, kcenon::thread::worker_policy::max_steal_attempts, policy_, kcenon::thread::worker_policy::steal_backoff, steal_function_, and worker_id_.

Member Data Documentation

◆ context_

thread_context kcenon::thread::thread_worker::context_

private

The thread context providing access to logging and monitoring services.

This context enables the worker to log messages and report metrics through the configured services.

Definition at line 314 of file thread_worker.h.

◆ current_job_

std::atomic<job*> kcenon::thread::thread_worker::current_job_ {nullptr}

private

Pointer to the currently executing job.

This is set atomically at the start of job execution and cleared when the job completes. Used by on_stop_requested() to cancel the running job.

Memory Ordering:

Release when storing (ensures job state visible to cancellation thread)
Acquire when loading (ensures we see the correct job state)

Note: Raw pointer used because we don't own the job (it's owned by unique_ptr in do_work()). Safe because job lifetime is guaranteed during execution.

Definition at line 366 of file thread_worker.h.

366{nullptr};

Referenced by get_current_job_info().

◆ current_job_start_time_

std::chrono::steady_clock::time_point kcenon::thread::thread_worker::current_job_start_time_

private

Time point when the current job started executing.

Used to track job execution time for diagnostics. Only valid when a job is currently executing.

Definition at line 426 of file thread_worker.h.

Referenced by get_current_job_info().

◆ diagnostics_

diagnostics::thread_pool_diagnostics* kcenon::thread::thread_worker::diagnostics_ {nullptr}

private

Pointer to the diagnostics instance for event tracing.

When set, the worker records execution events if tracing is enabled. Raw pointer because the diagnostics outlives the worker.

Definition at line 327 of file thread_worker.h.

327{nullptr};

Referenced by set_diagnostics().

◆ diagnostics_counter_

std::uint64_t kcenon::thread::thread_worker::diagnostics_counter_ {0}

private

Counter for diagnostics sampling.

Incremented on each job execution. Diagnostics events are recorded when (diagnostics_counter_ % diagnostics_sample_rate_ == 0).

Definition at line 351 of file thread_worker.h.

351{0};

◆ diagnostics_sample_rate_

std::uint32_t kcenon::thread::thread_worker::diagnostics_sample_rate_ {1}

private

Diagnostics sampling rate (record every Nth job).

Default is 1 (every job) for backward compatibility.

Definition at line 343 of file thread_worker.h.

343{1};

Referenced by set_diagnostics_sample_rate().

◆ is_idle_

std::atomic<bool> kcenon::thread::thread_worker::is_idle_ {true}

private

Indicates whether the worker is currently idle (not processing a job).

This flag is set to true when the worker is waiting for jobs and false when actively processing a job. Updated atomically for thread-safe access.

Memory Ordering:

Relaxed ordering sufficient (no synchronization dependencies)
Value is advisory only (race conditions between check and state change are acceptable)

Note: Used by thread pool for statistics and monitoring purposes.

Definition at line 380 of file thread_worker.h.

380{true};

Referenced by is_idle().

◆ job_queue_

std::shared_ptr<job_queue> kcenon::thread::thread_worker::job_queue_

private

A shared pointer to the job queue from which this worker obtains jobs.

Multiple workers can share the same queue, enabling concurrent processing of queued jobs.

Definition at line 306 of file thread_worker.h.

Referenced by should_continue_work(), and try_get_job().

◆ jobs_completed_

std::atomic<std::uint64_t> kcenon::thread::thread_worker::jobs_completed_ {0}

private

Total number of jobs successfully completed by this worker.

Incremented atomically after each successful job execution.

Definition at line 387 of file thread_worker.h.

387{0};

Referenced by get_jobs_completed().

◆ jobs_failed_

std::atomic<std::uint64_t> kcenon::thread::thread_worker::jobs_failed_ {0}

private

Total number of jobs that failed during execution.

Incremented atomically when a job's do_work() returns an error.

Definition at line 394 of file thread_worker.h.

394{0};

Referenced by get_jobs_failed().

◆ local_deque_

std::unique_ptr<lockfree::work_stealing_deque<job*> > kcenon::thread::thread_worker::local_deque_

private

Local work-stealing deque for this worker.

When work-stealing is enabled, jobs submitted to this worker are stored in this deque. The owner (this worker) can push/pop from the bottom (LIFO), while other workers can steal from the top (FIFO).

Definition at line 469 of file thread_worker.h.

Referenced by get_local_deque(), set_policy(), and try_get_job().

◆ metrics_

std::shared_ptr<metrics::ThreadPoolMetrics> kcenon::thread::thread_worker::metrics_

private

Shared metrics aggregator provided by the owning thread pool.

Definition at line 319 of file thread_worker.h.

Referenced by set_metrics().

◆ next_worker_id_

std::atomic< std::size_t > kcenon::thread::thread_worker::next_worker_id_ {0}

staticprivate

Static counter for generating unique worker IDs.

Definition at line 284 of file thread_worker.h.

◆ policy_

worker_policy kcenon::thread::thread_worker::policy_

private

Worker policy configuration.

Controls worker behavior including work-stealing settings.

Definition at line 459 of file thread_worker.h.

Referenced by get_policy(), set_policy(), try_get_job(), and try_steal_work().

◆ queue_being_replaced_

bool kcenon::thread::thread_worker::queue_being_replaced_ {false}

private

Indicates whether queue replacement is in progress.

When true, the worker thread should wait before accessing the queue. Protected by queue_mutex_.

Definition at line 452 of file thread_worker.h.

452{false};

◆ queue_cv_

std::condition_variable kcenon::thread::thread_worker::queue_cv_

private

Condition variable for queue replacement synchronization.

Used to wait for current job completion before replacing the queue.

Definition at line 444 of file thread_worker.h.

◆ queue_mutex_

std::mutex kcenon::thread::thread_worker::queue_mutex_

mutableprivate

Mutex protecting job queue replacement.

This mutex synchronizes access to job_queue_ during replacement operations to prevent race conditions between do_work(), set_job_queue(), and should_continue_work().

Note: Marked mutable to allow locking in const methods like should_continue_work(). The const qualifier applies to the logical state, not the mutex itself.

Definition at line 437 of file thread_worker.h.

Referenced by get_current_job_info(), should_continue_work(), and try_get_job().

◆ state_since_rep_

std::atomic<std::chrono::steady_clock::time_point::rep> kcenon::thread::thread_worker::state_since_rep_

private

Initial value:

{
            std::chrono::steady_clock::now().time_since_epoch().count()
        }

Time point when the worker entered its current state.

Updated when transitioning between idle and busy states. Used to calculate current state duration.

Definition at line 416 of file thread_worker.h.

                                                                      {
            std::chrono::steady_clock::now().time_since_epoch().count()
        };

Referenced by get_state_since().

◆ steal_function_

std::function<job*(std::size_t)> kcenon::thread::thread_worker::steal_function_

private

Function to steal work from other workers.

This function is provided by the thread pool and returns a stolen job from another worker's deque, or nullptr if no work available.

Definition at line 477 of file thread_worker.h.

Referenced by set_steal_function(), and try_steal_work().

◆ steal_victim_index_

std::size_t kcenon::thread::thread_worker::steal_victim_index_ {0}

private

Counter for round-robin steal victim selection.

Definition at line 482 of file thread_worker.h.

482{0};

◆ total_busy_time_ns_

std::atomic<std::uint64_t> kcenon::thread::thread_worker::total_busy_time_ns_ {0}

private

Total time spent executing jobs (busy time) in nanoseconds.

Accumulated after each job execution with the job's execution duration.

Definition at line 401 of file thread_worker.h.

401{0};

Referenced by get_total_busy_time().

◆ total_idle_time_ns_

std::atomic<std::uint64_t> kcenon::thread::thread_worker::total_idle_time_ns_ {0}

private

Total time spent waiting for jobs (idle time) in nanoseconds.

Accumulated when transitioning from idle to busy state.

Definition at line 408 of file thread_worker.h.

408{0};

Referenced by get_total_idle_time().

◆ use_time_tag_

bool kcenon::thread::thread_worker::use_time_tag_

private

Indicates whether to use time tags or timestamps for job processing.

When true, the worker might record timestamps (e.g., job start/end times) or log them for debugging/monitoring. The exact usage depends on the job and override details in derived classes.

Definition at line 298 of file thread_worker.h.

◆ worker_cancellation_token_

cancellation_token kcenon::thread::thread_worker::worker_cancellation_token_

private

Cancellation token for this worker.

This token is propagated to jobs during execution, allowing them to cooperatively cancel when the worker is stopped. The token is cancelled in on_stop_requested().

Definition at line 336 of file thread_worker.h.

◆ worker_id_

std::size_t kcenon::thread::thread_worker::worker_id_ {0}

private

Unique ID for this worker instance.

Definition at line 289 of file thread_worker.h.

289{0};

Referenced by get_worker_id(), and try_steal_work().

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

include/kcenon/thread/core/thread_worker.h
src/impl/thread_pool/thread_worker.cpp

Public Member Functions

Protected Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Additional Inherited Members

Detailed Description

Typical Usage

Constructor & Destructor Documentation

◆ thread_worker()

◆ ~thread_worker()

Member Function Documentation

◆ do_work()

◆ get_context()

◆ get_current_job_info()

◆ get_jobs_completed()

◆ get_jobs_failed()

◆ get_local_deque()

◆ get_policy()

◆ get_state_since()

◆ get_total_busy_time()

◆ get_total_idle_time()

◆ get_worker_id()

◆ is_idle()

◆ on_stop_requested()

◆ set_context()

◆ set_diagnostics()

◆ set_diagnostics_sample_rate()

◆ set_job_queue()

◆ set_metrics()

◆ set_policy()

◆ set_steal_function()

◆ should_continue_work()

◆ try_get_job()

◆ try_steal_work()

Member Data Documentation

◆ context_

◆ current_job_

◆ current_job_start_time_

◆ diagnostics_

◆ diagnostics_counter_

◆ diagnostics_sample_rate_

◆ is_idle_

◆ job_queue_

◆ jobs_completed_

◆ jobs_failed_

◆ local_deque_

◆ metrics_

◆ next_worker_id_

◆ policy_

◆ queue_being_replaced_

◆ queue_cv_

◆ queue_mutex_

◆ state_since_rep_

◆ steal_function_

◆ steal_victim_index_

◆ total_busy_time_ns_

◆ total_idle_time_ns_

◆ use_time_tag_

◆ worker_cancellation_token_

◆ worker_id_