thread_worker.h

Go to the documentation of this file.

 
#pragma once
 
// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#include <kcenon/thread/core/thread_base.h>
#include <kcenon/thread/utils/formatter.h>
#include <kcenon/thread/core/job_queue.h>
#include <kcenon/thread/core/cancellation_token.h>
#include <kcenon/thread/utils/convert_string.h>
#include <kcenon/thread/forward.h>
#include <kcenon/thread/interfaces/thread_context.h>
#include "worker_policy.h"
#include <kcenon/thread/metrics/thread_pool_metrics.h>
#include <kcenon/thread/lockfree/work_stealing_deque.h>
#include <kcenon/thread/diagnostics/job_info.h>
#include <kcenon/thread/diagnostics/execution_event.h>
 
#include <memory>
#include <optional>
#include <vector>
#include <atomic>
#include <mutex>
#include <condition_variable>
#include <functional>
 
// Forward declarations
namespace kcenon::thread::diagnostics
{
    class thread_pool_diagnostics;
}
 
namespace kcenon::thread
{
    class thread_worker : public thread_base
    {
    public:
        thread_worker(const bool& use_time_tag = true,
                     const thread_context& context = thread_context());
 
        virtual ~thread_worker(void);
 
        auto set_job_queue(std::shared_ptr<job_queue> job_queue) -> void;
 
        auto set_context(const thread_context& context) -> void;
 
        void set_metrics(std::shared_ptr<metrics::ThreadPoolMetrics> metrics);
 
        void set_diagnostics(diagnostics::thread_pool_diagnostics* diag);
 
        void set_diagnostics_sample_rate(std::uint32_t rate);
 
        void set_policy(const worker_policy& policy);
 
        [[nodiscard]] const worker_policy& get_policy() const;
 
        [[nodiscard]] lockfree::work_stealing_deque<job*>* get_local_deque() noexcept;
 
        void set_steal_function(std::function<job*(std::size_t)> steal_fn);
 
        [[nodiscard]] std::size_t get_worker_id() const;
 
        [[nodiscard]] auto get_context(void) const -> const thread_context&;
 
        [[nodiscard]] bool is_idle() const noexcept;
 
        [[nodiscard]] std::uint64_t get_jobs_completed() const noexcept;
 
        [[nodiscard]] std::uint64_t get_jobs_failed() const noexcept;
 
        [[nodiscard]] std::chrono::nanoseconds get_total_busy_time() const noexcept;
 
        [[nodiscard]] std::chrono::nanoseconds get_total_idle_time() const noexcept;
 
        [[nodiscard]] std::chrono::steady_clock::time_point get_state_since() const noexcept;
 
        [[nodiscard]] std::optional<diagnostics::job_info> get_current_job_info() const noexcept;
 
    protected:
        [[nodiscard]] auto should_continue_work() const -> bool override;
 
        auto do_work() -> common::VoidResult override;
 
        auto on_stop_requested() -> void override;
 
    private:
        static std::atomic<std::size_t> next_worker_id_;
 
        std::size_t worker_id_{0};
 
        bool use_time_tag_;
 
        std::shared_ptr<job_queue> job_queue_;
 
        thread_context context_;
 
        std::shared_ptr<metrics::ThreadPoolMetrics> metrics_;
 
        diagnostics::thread_pool_diagnostics* diagnostics_{nullptr};
 
        cancellation_token worker_cancellation_token_;
 
        std::uint32_t diagnostics_sample_rate_{1};
 
        std::uint64_t diagnostics_counter_{0};
 
        std::atomic<job*> current_job_{nullptr};
 
        std::atomic<bool> is_idle_{true};
 
        std::atomic<std::uint64_t> jobs_completed_{0};
 
        std::atomic<std::uint64_t> jobs_failed_{0};
 
        std::atomic<std::uint64_t> total_busy_time_ns_{0};
 
        std::atomic<std::uint64_t> total_idle_time_ns_{0};
 
        std::atomic<std::chrono::steady_clock::time_point::rep> state_since_rep_{
            std::chrono::steady_clock::now().time_since_epoch().count()
        };
 
        std::chrono::steady_clock::time_point current_job_start_time_;
 
        mutable std::mutex queue_mutex_;
 
        std::condition_variable queue_cv_;
 
        bool queue_being_replaced_{false};
 
        worker_policy policy_;
 
        std::unique_ptr<lockfree::work_stealing_deque<job*>> local_deque_;
 
        std::function<job*(std::size_t)> steal_function_;
 
        std::size_t steal_victim_index_{0};
 
        [[nodiscard]] std::unique_ptr<job> try_get_job();
 
        [[nodiscard]] std::unique_ptr<job> try_steal_work();
    };
} // namespace kcenon::thread
 
// ----------------------------------------------------------------------------
// Formatter specializations for thread_worker
// ----------------------------------------------------------------------------
 
template <>
struct std::formatter<kcenon::thread::thread_worker> : std::formatter<std::string_view>
{
    template <typename FormatContext>
    auto format(const kcenon::thread::thread_worker& item, FormatContext& ctx) const
    {
        return std::formatter<std::string_view>::format(item.to_string(), ctx);
    }
};
 
template <>
struct std::formatter<kcenon::thread::thread_worker, wchar_t>
    : std::formatter<std::wstring_view, wchar_t>
{
    template <typename FormatContext>
    auto format(const kcenon::thread::thread_worker& item, FormatContext& ctx) const
    {
        auto str = item.to_string();
        auto wstr = utility_module::convert_string::to_wstring(str);
        return std::formatter<std::wstring_view, wchar_t>::format(wstr, ctx);
    }
};