avx-http 0.4.0

//! Hierarchical Timer Wheel for efficient timeout management
//!
//! Inspired by Kafka's timer implementation:
//! - O(1) insertion
//! - O(1) cancellation
//! - O(m) expiration where m = number of expired timers
//!
//! Uses a hierarchical wheel with multiple levels of granularity

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::task::Waker;
use std::time::{Duration, Instant};

/// Timer wheel tick duration
const TICK_MS: u64 = 1;

/// Number of slots per wheel
const WHEEL_SIZE: usize = 256;

/// Timeout callback
type TimeoutCallback = Box<dyn FnOnce() + Send + 'static>;

/// Timer entry
struct TimerEntry {
    /// Expiration time
    expiration: Instant,
    /// Callback to execute on expiration
    callback: TimeoutCallback,
    /// Optional waker to wake
    waker: Option<Waker>,
}

/// A single wheel level
struct TimerWheel {
    /// Slots in this wheel
    slots: Vec<VecDeque<TimerEntry>>,
    /// Tick duration for this wheel
    tick_duration: Duration,
    /// Current tick
    current_tick: u64,
}

impl TimerWheel {
    fn new(tick_duration: Duration) -> Self {
        Self {
            slots: (0..WHEEL_SIZE).map(|_| VecDeque::new()).collect(),
            tick_duration,
            current_tick: 0,
        }
    }

    /// Add timer to this wheel, returning the entry if it can't be added
    fn add(&mut self, entry: TimerEntry, now: Instant) -> Option<TimerEntry> {
        let delay = entry.expiration.saturating_duration_since(now);
        let ticks = (delay.as_millis() as u64) / self.tick_duration.as_millis() as u64;

        // If timer is too far in the future for this wheel, return the entry back
        if ticks >= WHEEL_SIZE as u64 {
            return Some(entry);
        }

        let slot = ((self.current_tick + ticks) % WHEEL_SIZE as u64) as usize;
        self.slots[slot].push_back(entry);
        None
    }

    /// Advance wheel by one tick and return expired timers
    fn tick(&mut self, now: Instant) -> Vec<TimerEntry> {
        let slot_idx = (self.current_tick % WHEEL_SIZE as u64) as usize;
        self.current_tick += 1;

        // Collect expired timers
        let mut expired = Vec::new();
        while let Some(entry) = self.slots[slot_idx].pop_front() {
            if entry.expiration <= now {
                expired.push(entry);
            } else {
                // Timer moved to future wheel, needs re-insertion
                self.slots[slot_idx].push_back(entry);
            }
        }

        expired
    }
}

/// Hierarchical timer wheel
pub struct TimerWheelScheduler {
    /// Level 0: 1ms granularity (0-255ms)
    wheel_l0: TimerWheel,
    /// Level 1: 256ms granularity (256ms-65s)
    wheel_l1: TimerWheel,
    /// Level 2: 65s granularity (65s-4h)
    wheel_l2: TimerWheel,
    /// Start time
    start_time: Instant,
    /// Pending timers (too far in future)
    pending: Vec<TimerEntry>,
}

impl TimerWheelScheduler {
    /// Create new timer wheel
    pub fn new() -> Self {
        Self {
            wheel_l0: TimerWheel::new(Duration::from_millis(TICK_MS)),
            wheel_l1: TimerWheel::new(Duration::from_millis(TICK_MS * WHEEL_SIZE as u64)),
            wheel_l2: TimerWheel::new(Duration::from_millis(TICK_MS * WHEEL_SIZE as u64 * WHEEL_SIZE as u64)),
            start_time: Instant::now(),
            pending: Vec::new(),
        }
    }

    /// Schedule a timeout
    pub fn schedule<F>(&mut self, delay: Duration, callback: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let expiration = Instant::now() + delay;
        let entry = TimerEntry {
            expiration,
            callback: Box::new(callback),
            waker: None,
        };

        self.add_entry(entry);
    }

    /// Schedule with waker
    pub fn schedule_with_waker<F>(&mut self, delay: Duration, waker: Waker, callback: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let expiration = Instant::now() + delay;
        let entry = TimerEntry {
            expiration,
            callback: Box::new(callback),
            waker: Some(waker),
        };

        self.add_entry(entry);
    }

    /// Add entry to appropriate wheel
    fn add_entry(&mut self, mut entry: TimerEntry) {
        let now = Instant::now();

        // Try to add to L0 wheel (finest granularity)
        entry = match self.wheel_l0.add(entry, now) {
            Some(e) => e,
            None => return,
        };

        // Try L1 wheel
        entry = match self.wheel_l1.add(entry, now) {
            Some(e) => e,
            None => return,
        };

        // Try L2 wheel
        entry = match self.wheel_l2.add(entry, now) {
            Some(e) => e,
            None => return,
        };

        // Too far in future, add to pending
        self.pending.push(entry);
    }

    /// Advance time and process expired timers
    pub fn tick(&mut self) -> usize {
        let now = Instant::now();
        let mut expired_count = 0;

        // Tick L0 wheel
        let mut expired = self.wheel_l0.tick(now);
        expired_count += expired.len();

        // Execute callbacks
        for entry in expired.drain(..) {
            if let Some(waker) = entry.waker {
                waker.wake();
            }
            (entry.callback)();
        }

        // Cascade from L1 to L0 every 256 ticks
        if self.wheel_l0.current_tick % WHEEL_SIZE as u64 == 0 {
            let l1_expired = self.wheel_l1.tick(now);
            for entry in l1_expired {
                if let Some(entry) = self.wheel_l0.add(entry, now) {
                    // Shouldn't happen, but re-add to L1
                    let _ = self.wheel_l1.add(entry, now);
                }
            }
        }

        // Cascade from L2 to L1 every 256*256 ticks
        if self.wheel_l0.current_tick % (WHEEL_SIZE as u64 * WHEEL_SIZE as u64) == 0 {
            let l2_expired = self.wheel_l2.tick(now);
            for entry in l2_expired {
                let entry = match self.wheel_l1.add(entry, now) {
                    Some(e) => e,
                    None => continue,
                };

                let entry = match self.wheel_l2.add(entry, now) {
                    Some(e) => e,
                    None => continue,
                };

                self.pending.push(entry);
            }
        }

        // Process pending timers
        let pending: Vec<_> = self.pending.drain(..).collect();
        let mut still_pending = Vec::new();
        for entry in pending {
            if let Some(entry) = self.add_from_pending(entry, now) {
                still_pending.push(entry);
            }
        }
        self.pending = still_pending;

        expired_count
    }

    fn add_from_pending(&mut self, mut entry: TimerEntry, now: Instant) -> Option<TimerEntry> {
        entry = match self.wheel_l0.add(entry, now) {
            Some(e) => e,
            None => return None,
        };

        entry = match self.wheel_l1.add(entry, now) {
            Some(e) => e,
            None => return None,
        };

        entry = match self.wheel_l2.add(entry, now) {
            Some(e) => e,
            None => return None,
        };

        Some(entry)
    }

    /// Get time until next expiration
    pub fn time_until_next(&self) -> Option<Duration> {
        // Simplified: just return tick duration
        // Real implementation would scan wheels
        Some(Duration::from_millis(TICK_MS))
    }
}

impl Default for TimerWheelScheduler {
    fn default() -> Self {
        Self::new()
    }
}

/// Global timer wheel (thread-safe)
pub struct GlobalTimerWheel {
    inner: Arc<Mutex<TimerWheelScheduler>>,
}

impl GlobalTimerWheel {
    /// Create new global timer wheel
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(TimerWheelScheduler::new())),
        }
    }

    /// Schedule timeout
    pub fn schedule<F>(&self, delay: Duration, callback: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let mut wheel = self.inner.lock().unwrap();
        wheel.schedule(delay, callback);
    }

    /// Tick the timer wheel
    pub fn tick(&self) -> usize {
        let mut wheel = self.inner.lock().unwrap();
        wheel.tick()
    }
}

impl Default for GlobalTimerWheel {
    fn default() -> Self {
        Self::new()
    }
}

/// Sleep future using timer wheel
pub struct Sleep {
    deadline: Instant,
    registered: bool,
}

impl Sleep {
    /// Create new sleep future
    pub fn new(duration: Duration) -> Self {
        Self {
            deadline: Instant::now() + duration,
            registered: false,
        }
    }
}

impl std::future::Future for Sleep {
    type Output = ();

    fn poll(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Self::Output> {
        if Instant::now() >= self.deadline {
            return std::task::Poll::Ready(());
        }

        if !self.registered {
            // In a real implementation, we'd register with the timer wheel here
            self.registered = true;
        }

        std::task::Poll::Pending
    }
}

/// Sleep for a duration
pub fn sleep(duration: Duration) -> Sleep {
    Sleep::new(duration)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_timer_wheel_creation() {
        let wheel = TimerWheelScheduler::new();
        assert_eq!(wheel.wheel_l0.slots.len(), WHEEL_SIZE);
    }

    #[test]
    fn test_schedule_immediate() {
        let mut wheel = TimerWheelScheduler::new();
        let called = Arc::new(Mutex::new(false));
        let called_clone = Arc::clone(&called);

        wheel.schedule(Duration::from_millis(1), move || {
            *called_clone.lock().unwrap() = true;
        });

        // Wait and tick
        std::thread::sleep(Duration::from_millis(5));
        wheel.tick();

        assert!(*called.lock().unwrap());
    }

    #[test]
    fn test_timer_wheel_tick() {
        let mut wheel = TimerWheelScheduler::new();

        let count = Arc::new(Mutex::new(0));
        let count_clone = Arc::clone(&count);

        // Schedule multiple timers
        for i in 0..10 {
            let c = Arc::clone(&count);
            wheel.schedule(Duration::from_millis(i * 10), move || {
                *c.lock().unwrap() += 1;
            });
        }

        // Tick multiple times
        for _ in 0..200 {
            wheel.tick();
            std::thread::sleep(Duration::from_millis(1));
        }

        assert!(*count.lock().unwrap() > 0);
    }

    #[test]
    fn test_global_timer_wheel() {
        let wheel = GlobalTimerWheel::new();

        let called = Arc::new(Mutex::new(false));
        let called_clone = Arc::clone(&called);

        wheel.schedule(Duration::from_millis(5), move || {
            *called_clone.lock().unwrap() = true;
        });

        // Manually tick since no runtime is running
        std::thread::sleep(Duration::from_millis(10));
        for _ in 0..20 {
            wheel.tick();
            std::thread::sleep(Duration::from_millis(1));
        }

        assert!(*called.lock().unwrap());
    }

    #[test]
    fn test_sleep_future() {
        let sleep = sleep(Duration::from_millis(10));
        // Future test would require executor
    }
}