timer_deque_rs/

timer.rs

/*-
 * timer-rs - a Rust crate which provides timer and timer queues based on target OS
 *  functionality.
 * 
 * Copyright (C) 2025 Aleksandr Morozov
 * 
 * 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
 */

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::timer::
    {
        FdTimerCom, TimerExpMode, TimerFd, TimerFlags, TimerReadRes, TimerSetTimeFlags, TimerType
    }
};

/// A trait which is implemented by the structs which defines the behaviour 
/// of the timer queue.
pub trait OrderedTimerDequeIntrf
{
    /// 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;

    /// Wraps the input and initializes the timer time.
    /// 
    /// # Returns
    /// 
    /// Should return the `Self` instance and optionaly a `Self::Ticket` type.
    fn wrap(target: Self::Target, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<(Self, Self::Ticket)> 
        where Self: Sized;

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

/// 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.
/// 
/// # Generics
/// 
/// * `DQI` - a deque type. There are three types are available:
///     * - [crate::TimerDequeueTicketIssuer] which issues a ticket for the instance for which
///     the timer was set.
///     * - [crate::TimerDequeueConsumer] which consumes the instance for which the timer is set.
///     * - [crate::TimerDequeueSignalTicket] which sends a signal to destination.
#[derive(Debug)]
pub struct OrderedTimerDeque<DQI: OrderedTimerDequeIntrf + Ord + PartialOrd + PartialEq + Eq + fmt::Debug + fmt::Display>
{
    /// 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> Drop for OrderedTimerDeque<DQI>
where DQI: OrderedTimerDequeIntrf + Ord + PartialOrd + PartialEq + Eq + fmt::Debug + fmt::Display
{
    fn drop(&mut self) 
    {
        let _ = self.clean_up_timer();
    }
}

impl<DQI> AsFd for OrderedTimerDeque<DQI>
where DQI: OrderedTimerDequeIntrf + Ord + PartialOrd + PartialEq + Eq + fmt::Debug + fmt::Display
{
    fn as_fd(&self) -> BorrowedFd<'_> 
    {
        return self.timer.as_fd();
    }
}

impl<DQI> AsRawFd for OrderedTimerDeque<DQI>
where DQI: OrderedTimerDequeIntrf + Ord + PartialOrd + PartialEq + Eq + fmt::Debug + fmt::Display
{
    fn as_raw_fd(&self) -> RawFd 
    {
        return self.timer.as_raw_fd();
    }
}

impl<DQI> fmt::Display for OrderedTimerDeque<DQI>
where DQI: OrderedTimerDequeIntrf + Ord + PartialOrd + PartialEq + Eq + fmt::Debug + fmt::Display
{
    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 + Ord + PartialOrd + PartialEq + Eq + fmt::Debug + fmt::Display
{
    /// 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)
                )?;

        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)
                );

        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)
                );
    }

    /// 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()
        {
            return
                self
                    .timer
                    .set_time(
                    TimerSetTimeFlags::TFD_TIMER_ABSTIME | TimerSetTimeFlags::TFD_TIMER_CANCEL_ON_SET, 
                    front_entity.get_timeout_absolute()
                    )
                    .map_err(|e|
                        map_timer_err!(TimerErrorType::TimerError, "{}", e)
                    );
        }
        else
        {
            // queue is empty, force timer to stop

            return 
                self
                    .timer
                    .unset_time()
                    .map_err(|e|
                        map_timer_err!(TimerErrorType::TimerError, "{}", 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))?;

        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));
    }

    /// 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, entity: DQI::Target, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<DQI::Ticket>
    {
        let (wrapped, ext) = 
            DQI::wrap(entity, abs_time_sec, abs_time_nsec)?;

        let flags = TimerSetTimeFlags::TFD_TIMER_ABSTIME | TimerSetTimeFlags::TFD_TIMER_CANCEL_ON_SET;

        if self.deque_timeout_list.len() == 0
        {
            let timer_stamp = wrapped.get_timeout_absolute();

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

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

            let timer_stamp = wrapped.get_timeout_absolute();

            if front_banuntil >= timer_stamp
            {
                self
                    .timer
                    .set_time(flags, timer_stamp)
                    .map_err(|e|
                        map_timer_err!(TimerErrorType::TimerError, "{}", e)
                    )?;

                // push to front
                self.deque_timeout_list.push_front(wrapped);
            }
            else 
            {
                let back_banuntil = 
                    self
                        .deque_timeout_list
                        .back()
                        .unwrap()
                        .get_timeout_absolute();

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

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

        return Ok(ext);
    }
}