thread_worker.cpp

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#include <kcenon/thread/core/thread_worker.h>
#include <kcenon/thread/diagnostics/thread_pool_diagnostics.h>
 
#include <kcenon/thread/utils/formatter.h>
 
#include <thread>
 
// Platform-specific CPU pause intrinsics
#if defined(_MSC_VER) && (defined(_M_IX86) || defined(_M_X64))
    #include <emmintrin.h>  // For _mm_pause()
#endif
 
using namespace utility_module;
 
namespace kcenon::thread
{
    // Initialize static member
    std::atomic<std::size_t> thread_worker::next_worker_id_{0};
    thread_worker::thread_worker(const bool& use_time_tag, const thread_context& context)
        : 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::~thread_worker(void) {}
 
    auto thread_worker::set_job_queue(std::shared_ptr<job_queue> job_queue) -> void
    {
        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();
        }
    }
 
auto thread_worker::set_context(const thread_context& context) -> void
{
    context_ = context;
}
 
void thread_worker::set_metrics(std::shared_ptr<metrics::ThreadPoolMetrics> metrics)
{
    metrics_ = std::move(metrics);
}
 
void thread_worker::set_diagnostics(diagnostics::thread_pool_diagnostics* diag)
{
    diagnostics_ = diag;
}
 
void thread_worker::set_diagnostics_sample_rate(std::uint32_t rate)
{
    diagnostics_sample_rate_ = (rate > 0) ? rate : 1;
}
 
void thread_worker::set_policy(const worker_policy& policy)
{
    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*>>();
    }
}
 
const worker_policy& thread_worker::get_policy() const
{
    return policy_;
}
 
lockfree::work_stealing_deque<job*>* thread_worker::get_local_deque() noexcept
{
    return local_deque_.get();
}
 
void thread_worker::set_steal_function(std::function<job*(std::size_t)> steal_fn)
{
    steal_function_ = std::move(steal_fn);
}
 
std::unique_ptr<job> thread_worker::try_get_job()
{
    // 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;
}
 
std::unique_ptr<job> thread_worker::try_steal_work()
{
    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;
}
 
    auto thread_worker::get_context(void) const -> const thread_context&
    {
        return context_;
    }
 
    auto thread_worker::should_continue_work() const -> bool
    {
        // 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();
    }
 
    auto thread_worker::do_work() -> common::VoidResult
    {
        // 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();
    }
 
    std::size_t thread_worker::get_worker_id() const
    {
        return worker_id_;
    }
 
    bool thread_worker::is_idle() const noexcept
    {
        return is_idle_.load(std::memory_order_relaxed);
    }
 
    auto thread_worker::on_stop_requested() -> void
    {
        // 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_));
        }
    }
 
    std::uint64_t thread_worker::get_jobs_completed() const noexcept
    {
        return jobs_completed_.load(std::memory_order_relaxed);
    }
 
    std::uint64_t thread_worker::get_jobs_failed() const noexcept
    {
        return jobs_failed_.load(std::memory_order_relaxed);
    }
 
    std::chrono::nanoseconds thread_worker::get_total_busy_time() const noexcept
    {
        return std::chrono::nanoseconds{total_busy_time_ns_.load(std::memory_order_relaxed)};
    }
 
    std::chrono::nanoseconds thread_worker::get_total_idle_time() const noexcept
    {
        return std::chrono::nanoseconds{total_idle_time_ns_.load(std::memory_order_relaxed)};
    }
 
    std::chrono::steady_clock::time_point thread_worker::get_state_since() const noexcept
    {
        auto rep = state_since_rep_.load(std::memory_order_acquire);
        return std::chrono::steady_clock::time_point{
            std::chrono::steady_clock::duration{rep}
        };
    }
 
    std::optional<diagnostics::job_info> thread_worker::get_current_job_info() const noexcept
    {
        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;
    }
} // namespace kcenon::thread