timer_deque_rs/deque_timeout/

mod.rs

/*-
 * timer-deque-rs - a Rust crate which provides timer and timer queues based on target OS
 *  functionality.
 * 
 * Copyright (C) 2025 Aleksandr Morozov alex@nixd.org
 *  4neko.org alex@4neko.org
 * 
 * The timer-rs crate can be redistributed and/or modified
 * under the terms of either of the following licenses:
 *
 *   1. the Mozilla Public License Version 2.0 (the “MPL”) OR
 *                     
 *   2. EUROPEAN UNION PUBLIC LICENCE v. 1.2 EUPL © the European Union 2007, 2016
 */

/// A `consumer` type of the timer which consumes the intance and returns it when
/// timer triggers. The `consumed` instance normally whould be [Send] because it will
/// be moved into the timer.
pub mod timer_consumer;

/// A `ticket` issuer. Issues a ticket which should be assigned to the instance whcih was added
/// to the timer's queue. The `ticket` can be used to remove the item from queue before the 
/// timeout event. If ticket is dropped i.e connection closed, the ticket will be
/// in timer's queue until timeout where it will be ignored on timeout event.
pub mod timer_tickets;

/// A `signal` sender. Calls the specified callback which must never block the executing thread.
pub mod timer_signal;

pub use std::os::fd::{AsFd, AsRawFd};
use std::{borrow::Cow, collections::VecDeque, fmt, os::fd::{BorrowedFd, RawFd}};

use crate::
{
    error::{TimerErrorType, TimerResult}, 
    map_timer_err, 
    timer_portable::
    {
        poll::AsTimerFd, 
        timer::
        {
            AbsoluteTime, 
            FdTimerCom, 
            RelativeTime, 
            TimerExpMode, 
            TimerFd, 
            TimerFlags, 
            TimerReadRes, 
            TimerType
        }
    }
};

/// A trait which is implemented by the structs which defines the behaviour 
/// of the timer queue.
pub trait OrderedTimerDequeIntrf: Ord + PartialOrd + PartialEq + Eq + fmt::Debug + fmt::Display
{
    /// A timer item for the queue which is passed as argument. If noting is 
    /// provided the `NoTarget` can be used.
    type Target: PartialEq + Eq + fmt::Display + fmt::Debug;

    /// A timer queue identification in the queue which may be retuened.
    /// If nothing is retuned the `NoTicket` can be returned.
    type Ticket: PartialEq + Eq + fmt::Display + fmt::Debug;


    /// Should return the absolute time and the timer mode.
    fn get_timeout_absolute(&self) -> AbsoluteTime;
}

/// A trait which is implemented by the struct which defines the operation mode of the deque.
pub trait OrderedTimerDequeMode: fmt::Debug + fmt::Display + Ord + PartialOrd + Eq + PartialEq
{
    /// Returns the `absolute` timeout for the instance of the deque.
    fn get_absolut_timeout(&self) -> AbsoluteTime;

    /// Updates the timeout when the deque works in periodic mode.
    fn advance_timeout(&mut self)
    {
        return;
    }
}

/// This queue mode removes all entries from the queue that have timed out.
/// 
/// The further behaviour is defined by the type of the deque.
#[derive(Debug)]
pub struct OrderdTimerDequeOnce 
{
    /// A timeout for the item in the queue.
    absolute_timeout: AbsoluteTime,
}

impl fmt::Display for OrderdTimerDequeOnce
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        write!(f, "{}", self.absolute_timeout)
    }
}

impl Ord for OrderdTimerDequeOnce
{
    fn cmp(&self, other: &Self) -> std::cmp::Ordering 
    {
        return self.absolute_timeout.cmp(&other.absolute_timeout);
    }
}

impl PartialOrd for OrderdTimerDequeOnce
{
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> 
    {
        return Some(self.cmp(other));
    }
}

impl Eq for OrderdTimerDequeOnce {}

impl PartialEq for OrderdTimerDequeOnce
{
    fn eq(&self, other: &Self) -> bool 
    {
        return self.absolute_timeout == other.absolute_timeout;
    }
}


impl OrderedTimerDequeMode for OrderdTimerDequeOnce
{
    fn get_absolut_timeout(&self) -> AbsoluteTime
    {
        return self.absolute_timeout;
    }
}

impl OrderdTimerDequeOnce
{
    /// Creates new instacne.
    pub(crate)
    fn new(absolute_timeout: AbsoluteTime) -> Self
    {
        return 
            Self
            {
                absolute_timeout: 
                    absolute_timeout
            };
    }
}

/// This queue mode does not remove an element that has timed out (by `absolute_timeout`), 
/// but extends (by `relative_period`) the timeout and returns the element back to the queue.
/// 
/// The further behaviour is defined by the type of the deque.
#[derive(Debug)]
pub struct OrderdTimerDequePeriodic 
{
    /// Extends the timer until the next timeout. This is `relative`
    /// not absolute.
    relative_period: RelativeTime,

    /// A timeout value.
    absolute_timeout: AbsoluteTime,
}

impl fmt::Display for OrderdTimerDequePeriodic
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        write!(f, "{}, rel: {}", self.absolute_timeout, self.relative_period)
    }
}

impl Ord for OrderdTimerDequePeriodic
{
    fn cmp(&self, other: &Self) -> std::cmp::Ordering 
    {
        return self.absolute_timeout.cmp(&other.absolute_timeout);
    }
}

impl PartialOrd for OrderdTimerDequePeriodic
{
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> 
    {
        return Some(self.cmp(other));
    }
}

impl Eq for OrderdTimerDequePeriodic {}

impl PartialEq for OrderdTimerDequePeriodic
{
    fn eq(&self, other: &Self) -> bool 
    {
        return self.absolute_timeout == other.absolute_timeout;
    }
}


impl OrderedTimerDequeMode for OrderdTimerDequePeriodic
{
    fn get_absolut_timeout(&self) -> AbsoluteTime
    {
        return self.absolute_timeout;
    }

    fn advance_timeout(&mut self)
    {
        self.absolute_timeout += self.relative_period;
    }
}

impl OrderdTimerDequePeriodic
{
    /// Creates new instance.
    pub(crate)
    fn new(rel_time: RelativeTime) -> Self
    {
        let mut inst = 
            Self
            {
                relative_period:
                    rel_time,
                absolute_timeout: 
                    AbsoluteTime::now(),
            };

        inst.advance_timeout();

        return inst;
    }
}

/// A [VecDeque] based queue which is sorted (in ascending order) by the timeout 
/// which is `absolute` time.
/// 
/// The queue automatically manages the `timer` i.e setting, unsetting.
/// 
/// Also for each type of the deque, a event procesing function is providided.
/// 
/// There are two types of queue:
/// 
/// * [OrderdTimerDequeOnce] - after timeout the element is removed from the queue.
/// 
/// * [OrderdTimerDequePeriodic] - after timeout the element timeout is extended
///     until the item is not removed from the queue manually.
/// 
/// # Generics
/// 
/// * `DQI` - a deque type. There are three types are available:
///     * - [crate::TimerDequeueTicketIssuer] issues a ticket for the instance for which the timer was set.
///     ```ignore
///         let mut time_list = 
///             OrderedTimerDeque
///                 ::<TimerDequeueTicketIssuer<OrderdTimerDequeOnce>>
///                 ::new("test_label".into(), 4, false).unwrap();
///     ```  
///     or
///     ```ignore
///         let mut time_list = 
///             OrderedTimerDeque
///                 ::<TimerDequeueTicketIssuer<OrderdTimerDequePeriodic>>
///                 ::new("test_label".into(), 4, false).unwrap();
///     ```     
///     * - [crate::TimerDequeueConsumer] consumes the instance for which the timer is set.
///     ```ignore
///         let mut time_list = 
///             OrderedTimerDeque
///                 ::<TimerDequeueConsumer<TestItem, OrderdTimerDequeOnce>>
///                 ::new("test_label".into(), 4, false).unwrap();
///     ```  
///     or
///     ```ignore
///         let mut time_list = 
///             OrderedTimerDeque
///                 ::<TimerDequeueConsumer<TestItem, OrderdTimerDequePeriodic>>
///                 ::new("test_label".into(), 4, false).unwrap();
///     ``` 
///     * - [crate::TimerDequeueSignalTicket] sends a signal to destination.
///     ```ignore
///         let mut time_list = 
///             OrderedTimerDeque
///                 ::<TimerDequeueSignalTicket<TestSigStruct, OrderdTimerDequeOnce>>
///                 ::new("test_label".into(), 4, false).unwrap();
///     ```
///     or
///     ```ignore
///         let mut time_list = 
///             OrderedTimerDeque
///                 ::<TimerDequeueSignalTicket<TestSigStruct, OrderdTimerDequePeriodic>>
///                 ::new("test_label".into(), 4, false).unwrap();
///     ```
#[derive(Debug)]
pub struct OrderedTimerDeque<DQI: OrderedTimerDequeIntrf>
{
    /// A [VecDeque] list which is sorted by time in ascending order -
    /// lower first, largest last
    pub(crate) deque_timeout_list: VecDeque<DQI>,

    /// An instance of the FFI (Kernel Supported Timer)
    pub(crate) timer: TimerFd,
}

impl<DQI> AsTimerFd for OrderedTimerDeque<DQI> where DQI: OrderedTimerDequeIntrf
{
    #[inline]
    fn get_bind(&self) -> Option<std::sync::Arc<crate::timer_portable::DefaultEventWatch>> 
    {
        return self.timer.get_bind();
    }

    #[inline]
    fn bind_poll(&self, timer_weak_ref: std::sync::Weak<crate::timer_portable::DefaultEventWatch>) 
    {
        return self.timer.bind_poll(timer_weak_ref);
    }

    fn unbind_poll(&self) 
    {
        return self.timer.unbind_poll();
    }
}   

impl<DQI> Drop for OrderedTimerDeque<DQI> where DQI: OrderedTimerDequeIntrf
{
    fn drop(&mut self) 
    {
        let _ = self.clean_up_timer();
    }
}

impl<DQI> AsFd for OrderedTimerDeque<DQI> where DQI: OrderedTimerDequeIntrf 
{
    fn as_fd(&self) -> BorrowedFd<'_> 
    {
        return self.timer.as_fd();
    }
}

impl<DQI> AsRawFd for OrderedTimerDeque<DQI> where DQI: OrderedTimerDequeIntrf
{
    fn as_raw_fd(&self) -> RawFd 
    {
        return self.timer.as_raw_fd();
    }
}

impl<DQI> fmt::Display for OrderedTimerDeque<DQI> where DQI: OrderedTimerDequeIntrf 
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        write!(f, "timer: '{}', fd: '{}', queue_len: '{}'", 
            self.timer, self.timer.as_fd().as_raw_fd(), self.deque_timeout_list.len())
    }
}

impl<DQI> OrderedTimerDeque<DQI> where DQI: OrderedTimerDequeIntrf 
{
    /// Creates new deque instance.
    /// 
    /// # Argument
    /// 
    /// `timer_label` - a label which helps to identify the timer.
    /// 
    /// `deq_len` - a minimal, pre-allocated deque length.
    /// 
    /// `cloexec` - when set to `true` sets the `CLOEXEC` flag to FD.
    /// 
    /// # Returns
    /// 
    /// A [Result] as alias [TimerResult] is returned with
    /// 
    /// * [Result::Ok] with the instance.
    /// 
    /// * [Result::Err] with the error description.
    pub 
    fn new(timer_label: Cow<'static, str>, deq_len: usize, cloexec: bool) -> TimerResult<OrderedTimerDeque<DQI>>
    {
        let deq_len = 
            if deq_len == 0
            {
                10
            }
            else
            {
                deq_len
            };

        let tf = 
            if cloexec == true
            {
                TimerFlags::TFD_CLOEXEC
            }
            else
            {
                TimerFlags::empty()
            };

        // init timer
        let timer = 
            TimerFd::new(timer_label, TimerType::CLOCK_REALTIME, TimerFlags::TFD_NONBLOCK | tf)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "{}", e)
                )?;

        return Ok( 
            Self
            { 
                deque_timeout_list: VecDeque::with_capacity(deq_len), 
                timer: timer
            } 
        );
    }

    /// Reads the FD to retrive the event type.
    /// 
    /// This function behaves differently when the timer is set with [TimerFlags::TFD_NONBLOCK].
    /// 
    /// Normally, all timer_* creates nonblocking timer, so this function behaves like when
    /// the [TimerFlags::TFD_NONBLOCK] is set.
    /// 
    /// When [TimerFlags::TFD_NONBLOCK] is not set, this function will block reading the FD.
    /// In case of 'EINTR', the read attempt will be repeated. It will be reapeated in both cases.
    /// 
    /// When [TimerFlags::TFD_NONBLOCK] is set, the function will return with some result.
    /// 
    /// # Return
    /// 
    /// * In case of `EAGAIN`, the [TimerReadRes::WouldBlock] will be returned. 
    /// 
    /// * In case of `ECANCELLD`, the [TimerReadRes::WouldBlock] will be returned.
    /// 
    /// * In case of any other error the [Result::Err] is returned.
    /// 
    /// When a timer fires an event the [Result::Ok] is returned with the amount of
    /// timer overrun. Normally it is 1.
    pub 
    fn wait_for_event(&self) -> TimerResult<TimerReadRes<u64>>
    {
        let res = 
            self
                .timer
                .read()
                .map_err(|e|
                    map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "{}", e)
                );

        return res;
    }

    /// Asynchronious polling. The timer's FD is set to nonblocking,
    /// so each time it will return `pending` and load CPU. If you are using
    /// `tokio` or `smol` suing corresponding helpers like tokio's `AsyncFd`.
    pub async
    fn poll(&self) -> TimerResult<TimerReadRes<u64>>
    {
        return 
            (&self.timer)
                .await
                .map_err(|e|
                    map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "{}", e)
                );
    }

    /// Returns the queue length.
    pub 
    fn timer_queue_len(&self) -> usize
    {
        return self.deque_timeout_list.len();
    }
    

    /// Setting the `timer` instance to the new values or unsets the timer if
    /// `queue` becomes empty.
    pub  
    fn reschedule_timer(&mut self) -> TimerResult<()>
    {
        if let Some(front_entity) = self.deque_timeout_list.front()
        {
            let timer_exp = 
                TimerExpMode::<AbsoluteTime>::new_oneshot(front_entity.get_timeout_absolute());

            return
                self
                    .timer
                    .set_time(timer_exp)
                    .map_err(|e|
                        map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "{}", e)
                    );
        }
        else
        {
            // queue is empty, force timer to stop

            return 
                self
                    .timer
                    .unset_time()
                    .map_err(|e|
                        map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "{}", e)
                    );
        }
    }

    /// Unarms timer and clears the queue.
    pub 
    fn clean_up_timer(&mut self) -> TimerResult<()>
    {
        self
            .timer
            .unset_time()
            .map_err(|e| map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "{}", e))?;

        self.deque_timeout_list.clear();

        return Ok(());
    }

    /// Unarms timer only.
    pub 
    fn stop_timer(&mut self) -> TimerResult<()>
    {
        return 
            self
                .timer
                .unset_time()
                .map_err(|e| map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "{}", e));
    }

    /// An internal function to recude code duplication. Each deque item type has its own `add` function
    /// realization.
    pub(crate)   
    fn add_to_timer_local(&mut self, inst: DQI) -> TimerResult<()>
    {
        if self.deque_timeout_list.len() == 0
        {
            let timer_stamp = 
                TimerExpMode::<AbsoluteTime>::new_oneshot(inst.get_timeout_absolute());

            // setup timer
            self
                .timer
                .set_time(timer_stamp)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "{}", e)
                )?;

            self.deque_timeout_list.push_front(inst);      
        }
        else
        {
            // list can not be empty from this point
            let front_timeout = 
                self.deque_timeout_list.front().unwrap().get_timeout_absolute();

            // intances timeout
            let inst_timeout = inst.get_timeout_absolute();

            if front_timeout >= inst_timeout
            {
                // push to front
                self.deque_timeout_list.push_front(inst);

                self.reschedule_timer()?;
            }
            else 
            {
                let back_banuntil = 
                    self
                        .deque_timeout_list
                        .back()
                        .unwrap()
                        .get_timeout_absolute();

                if back_banuntil <= inst_timeout
                {
                    // push to the back
                    self.deque_timeout_list.push_back(inst);
                }
                else
                {
                    let pos = 
                        self
                            .deque_timeout_list
                            .binary_search_by( |se| 
                                se.get_timeout_absolute().cmp(&inst.get_timeout_absolute())
                            )
                            .map_or_else(|e| e, |r| r);

                    self.deque_timeout_list.insert(pos, inst);
                }
            }
        }

        return Ok(());
    }
}