timerwheel 0.1.0

// Copyright © 2026-present The Timerwheel Authors. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::fmt;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};

use crate::timer::metric_sink::MetricSink;

/// Immutable snapshot of timer scheduler counters.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TimerMetrics {
    /// Number of schedules accepted by the timer.
    pub scheduled_timeouts: u64,
    /// Number of timeouts accepted by an executor for execution.
    pub expired_timeouts: u64,
    /// Number of timeouts cancelled before execution.
    pub cancelled_timeouts: u64,
    /// Number of schedules rejected by capacity or closed state.
    pub rejected_schedules: u64,
    /// Number of expired task dispatch attempts rejected by an executor.
    pub rejected_dispatches: u64,
    /// Number of scheduled tasks that panicked during execution.
    pub panicked_timeouts: u64,
    /// Number of accepted timeouts still owned by the timer.
    pub pending_timeouts: usize,
    /// Current number of schedule commands waiting for the scheduler.
    pub command_queue_depth: usize,
    /// Number of entries offered into timer-wheel buckets.
    pub bucket_offers: u64,
    /// Number of bucket expiration scans performed by the scheduler.
    pub bucket_expirations: u64,
    /// Maximum observed bucket delay in milliseconds.
    pub max_bucket_delay: u64,
}

pub(crate) struct TimerMetricsInner {
    scheduled_timeouts: AtomicU64,
    expired_timeouts: AtomicU64,
    cancelled_timeouts: AtomicU64,
    rejected_schedules: AtomicU64,
    rejected_dispatches: AtomicU64,
    panicked_timeouts: AtomicU64,
    pending_timeouts: AtomicUsize,
    command_queue_depth: AtomicUsize,
    bucket_offers: AtomicU64,
    bucket_expirations: AtomicU64,
    max_bucket_delay: AtomicU64,
    sink: Arc<dyn MetricSink>,
}

impl Default for TimerMetricsInner {
    fn default() -> Self {
        Self::new(Arc::new(crate::timer::metric_sink::NoopMetricSink))
    }
}

impl fmt::Debug for TimerMetricsInner {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("TimerMetricsInner")
            .field("snapshot", &self.snapshot())
            .finish_non_exhaustive()
    }
}

impl TimerMetricsInner {
    pub(crate) fn new(sink: Arc<dyn MetricSink>) -> Self {
        Self {
            scheduled_timeouts: AtomicU64::new(0),
            expired_timeouts: AtomicU64::new(0),
            cancelled_timeouts: AtomicU64::new(0),
            rejected_schedules: AtomicU64::new(0),
            rejected_dispatches: AtomicU64::new(0),
            panicked_timeouts: AtomicU64::new(0),
            pending_timeouts: AtomicUsize::new(0),
            command_queue_depth: AtomicUsize::new(0),
            bucket_offers: AtomicU64::new(0),
            bucket_expirations: AtomicU64::new(0),
            max_bucket_delay: AtomicU64::new(0),
            sink,
        }
    }

    pub(crate) fn scheduled(&self) {
        self.scheduled_timeouts.fetch_add(1, Ordering::Relaxed);
        self.pending_timeouts.fetch_add(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn expired(&self) {
        self.expired_timeouts.fetch_add(1, Ordering::Relaxed);
        self.pending_timeouts.fetch_sub(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn panicked(&self) {
        self.panicked_timeouts.fetch_add(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn cancelled(&self) {
        self.cancelled_timeouts.fetch_add(1, Ordering::Relaxed);
        self.pending_timeouts.fetch_sub(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn rejected_schedule(&self) {
        self.rejected_schedules.fetch_add(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn rejected_dispatch(&self) {
        self.rejected_dispatches.fetch_add(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn dropped_after_dispatch_reject(&self) {
        self.rejected_dispatches.fetch_add(1, Ordering::Relaxed);
        self.pending_timeouts.fetch_sub(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn cancelled_pending(&self) {
        self.cancelled_timeouts.fetch_add(1, Ordering::Relaxed);
        self.pending_timeouts.fetch_sub(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn command_queued(&self) {
        self.command_queue_depth.fetch_add(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn command_drained(&self) {
        self.command_queue_depth.fetch_sub(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn bucket_offered(&self) {
        self.bucket_offers.fetch_add(1, Ordering::Relaxed);
        self.observe();
    }

    pub(crate) fn bucket_expired(&self) {
        self.bucket_expirations.fetch_add(1, Ordering::Relaxed);
        self.observe();
    }

    fn observe(&self) {
        self.sink.observe_metrics(self.snapshot());
    }

    pub(crate) fn snapshot(&self) -> TimerMetrics {
        TimerMetrics {
            scheduled_timeouts: self.scheduled_timeouts.load(Ordering::Relaxed),
            expired_timeouts: self.expired_timeouts.load(Ordering::Relaxed),
            cancelled_timeouts: self.cancelled_timeouts.load(Ordering::Relaxed),
            rejected_schedules: self.rejected_schedules.load(Ordering::Relaxed),
            rejected_dispatches: self.rejected_dispatches.load(Ordering::Relaxed),
            panicked_timeouts: self.panicked_timeouts.load(Ordering::Relaxed),
            pending_timeouts: self.pending_timeouts.load(Ordering::Relaxed),
            command_queue_depth: self.command_queue_depth.load(Ordering::Relaxed),
            bucket_offers: self.bucket_offers.load(Ordering::Relaxed),
            bucket_expirations: self.bucket_expirations.load(Ordering::Relaxed),
            max_bucket_delay: self.max_bucket_delay.load(Ordering::Relaxed),
        }
    }
}