enhanced_metrics.h

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.
 
#pragma once
 
#include <kcenon/thread/metrics/latency_histogram.h>
#include <kcenon/thread/metrics/metrics_base.h>
#include <kcenon/thread/metrics/sliding_window_counter.h>
 
#include <atomic>
#include <chrono>
#include <cstdint>
#include <mutex>
#include <optional>
#include <string>
#include <vector>
 
namespace kcenon::thread::metrics {
 
struct EnhancedSnapshot {
    // =========================================================================
    // Basic Counters (from ThreadPoolMetrics)
    // =========================================================================
 
    std::uint64_t tasks_submitted{0};
 
    std::uint64_t tasks_executed{0};
 
    std::uint64_t tasks_failed{0};
 
    // =========================================================================
    // Latency Percentiles (microseconds for readability)
    // =========================================================================
 
    double enqueue_latency_p50_us{0.0};
 
    double enqueue_latency_p90_us{0.0};
 
    double enqueue_latency_p99_us{0.0};
 
    double execution_latency_p50_us{0.0};
 
    double execution_latency_p90_us{0.0};
 
    double execution_latency_p99_us{0.0};
 
    double wait_time_p50_us{0.0};
 
    double wait_time_p90_us{0.0};
 
    double wait_time_p99_us{0.0};
 
    // =========================================================================
    // Throughput
    // =========================================================================
 
    double throughput_1s{0.0};
 
    double throughput_1m{0.0};
 
    // =========================================================================
    // Queue Health
    // =========================================================================
 
    std::size_t current_queue_depth{0};
 
    std::size_t peak_queue_depth{0};
 
    double avg_queue_depth{0.0};
 
    // =========================================================================
    // Worker Utilization
    // =========================================================================
 
    double worker_utilization{0.0};
 
    std::vector<double> per_worker_utilization;
 
    std::size_t active_workers{0};
 
    // =========================================================================
    // Timing Information
    // =========================================================================
 
    std::uint64_t total_busy_time_ns{0};
 
    std::uint64_t total_idle_time_ns{0};
 
    std::chrono::steady_clock::time_point snapshot_time;
};
 
struct WorkerMetrics {
    std::size_t worker_id{0};
 
    std::uint64_t tasks_executed{0};
 
    std::uint64_t busy_time_ns{0};
 
    std::uint64_t idle_time_ns{0};
 
    bool is_busy{false};
};
 
class EnhancedThreadPoolMetrics : public MetricsBase {
public:
    explicit EnhancedThreadPoolMetrics(std::size_t worker_count = 0);
 
    ~EnhancedThreadPoolMetrics() override = default;
 
    // Non-copyable, non-movable for thread safety
    EnhancedThreadPoolMetrics(const EnhancedThreadPoolMetrics&) = delete;
    EnhancedThreadPoolMetrics& operator=(const EnhancedThreadPoolMetrics&) = delete;
    EnhancedThreadPoolMetrics(EnhancedThreadPoolMetrics&&) = delete;
    EnhancedThreadPoolMetrics& operator=(EnhancedThreadPoolMetrics&&) = delete;
 
    // =========================================================================
    // Recording Methods (called by thread_pool)
    // =========================================================================
 
    void record_submission();
 
    void record_enqueue(std::chrono::nanoseconds latency);
 
    void record_execution(std::chrono::nanoseconds latency, bool success);
 
    void record_wait_time(std::chrono::nanoseconds wait);
 
    void record_queue_depth(std::size_t depth);
 
    void record_worker_state(
        std::size_t worker_id,
        bool busy,
        std::uint64_t duration_ns = 0);
 
    void set_active_workers(std::size_t count);
 
    // =========================================================================
    // Query Methods
    // =========================================================================
 
    [[nodiscard]] EnhancedSnapshot snapshot() const;
 
    [[nodiscard]] const LatencyHistogram& enqueue_latency() const;
 
    [[nodiscard]] const LatencyHistogram& execution_latency() const;
 
    [[nodiscard]] const LatencyHistogram& wait_time() const;
 
    [[nodiscard]] std::vector<WorkerMetrics> worker_metrics() const;
 
    [[nodiscard]] const SlidingWindowCounter& throughput_1s() const;
 
    [[nodiscard]] const SlidingWindowCounter& throughput_1m() const;
 
    // =========================================================================
    // Management Methods
    // =========================================================================
 
    void reset() override;
 
    void update_worker_count(std::size_t count);
 
    // =========================================================================
    // Export Methods
    // =========================================================================
 
    [[nodiscard]] std::string to_json() const;
 
    [[nodiscard]] std::string to_prometheus(
        const std::string& prefix = "thread_pool") const;
 
private:
    // Latency histograms
    LatencyHistogram enqueue_latency_;
    LatencyHistogram execution_latency_;
    LatencyHistogram wait_time_;
 
    // Throughput counters
    SlidingWindowCounter throughput_1s_;
    SlidingWindowCounter throughput_1m_;
 
    // Note: Basic counters (tasks_submitted_, tasks_executed_, tasks_failed_,
    // total_busy_time_ns_, total_idle_time_ns_) are inherited from MetricsBase
 
    // Queue depth tracking
    std::atomic<std::size_t> current_queue_depth_{0};
    std::atomic<std::size_t> peak_queue_depth_{0};
    std::atomic<std::uint64_t> queue_depth_sum_{0};
    std::atomic<std::uint64_t> queue_depth_samples_{0};
 
    // Worker tracking
    std::atomic<std::size_t> active_workers_{0};
 
    // Per-worker metrics
    mutable std::mutex workers_mutex_;
    std::vector<WorkerMetrics> per_worker_metrics_;
 
    // Helper to convert nanoseconds to microseconds
    [[nodiscard]] static double ns_to_us(double ns) { return ns / 1000.0; }
};
 
} // namespace kcenon::thread::metrics