timer_deque_rs/

timer_signal.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
 */

use std::fmt;

use crate::
{
    common, 
    error::{TimerErrorType, TimerResult}, 
    map_timer_err, 
    timer::{OrderedTimerDeque, OrderedTimerDequeIntrf}, 
    timer_err, 
    timer_portable::timer::TimerExpMode, TimerReadRes
};

/// A trait which should be implemented on the struct which will be provided
/// to timer dequeue. The trait implements the uniqid and signalling. There 
/// are two realizations: `sig_timeout` for sync and `a_sig_timeout` for
/// async. Both functions must use any methods to send signal but it must
/// never block the thread!
pub trait TimerDequeueSignal
{
    /// A uniq id type. It is used to identify the instance.
    type TimerQueueID: Eq + PartialEq + fmt::Display + fmt::Debug;

    /// Error type returned by the signal functions.
    type TimeoutErr: fmt::Display + fmt::Debug;

    /// This function should return the uniq id of the instance.
    fn get_id(&self) -> Self::TimerQueueID;
    
    /// This function should notify i.e send signal or whatever. The
    /// important condition is not to block the thread. 
    fn sig_timeout(self) -> Result<(), Self::TimeoutErr>
    where Self: Sized
    {
        return Ok(());
    }

    /// This function should notify i.e send signal or whatever. The
    /// important condition is not to block the thread. 
    async fn a_sig_timeout(self) -> Result<(), Self::TimeoutErr>
    where Self: Sized
    {
        return Ok(());
    }
}

/// Defines the type of the dequeue. This type of the dequeue is sending notification when
/// the timeout event happens. The signal will be sent using `SIG`.
/// 
/// # Arguments
/// 
/// `SIG` - is an instance which provides the signaling and ID. It must implement:
///     [TimerDequeueSignal], [fmt::Debug], [fmt::Display], [Eq], [PartialEq].
#[derive(Debug)]
pub struct TimerDequeueSignalTicket<SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq>
{
    /// A notification and id instance.
    signal: SIG,
    
    /// An absolute timestamp in seconds (UTC time) sets the timer.
    timeout_absolute: TimerExpMode,
}

impl<SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq> TimerDequeueSignalTicket<SIG>
{
    pub 
    fn new(sig_hnd: SIG, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<Self>
    {
        let abs_time_nsec = abs_time_nsec & 999_999_999;

        let cur = common::get_current_timestamp();
        let req = TimerExpMode::OneShot { sec: abs_time_sec, nsec: abs_time_nsec };

        if TimerExpMode::from(cur) > req
        {
            timer_err!(TimerErrorType::Expired, "time already expired");
        }

        
        return Ok(
            Self
            {
                signal: 
                    sig_hnd,
                timeout_absolute: 
                    req
            }
        );
    }

    fn get_time_until(&self) -> TimerExpMode
    {
        return self.timeout_absolute;
    }

    fn send_sig_timeout(self) -> TimerResult<()>
    {
        return 
            self
                .signal
                .sig_timeout()
                .map_err(|e|
                    map_timer_err!(TimerErrorType::ExternalError, "cannot send signal, error: {}", e)
                );
    }

    async 
    fn async_send_sig_timeout(self) -> TimerResult<()>
    {
        return 
            self
                .signal
                .a_sig_timeout()
                .await
                .map_err(|e|
                    map_timer_err!(TimerErrorType::ExternalError, "cannot send signal, error: {}", e)
                );
    }
}

impl<SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq> fmt::Display for TimerDequeueSignalTicket<SIG>
{  
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> 
    { 
        write!(f, "{} until: {}", self.signal.get_id(), self.timeout_absolute)
    }
}

impl<SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq> Eq for TimerDequeueSignalTicket<SIG> {}

impl<SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq>  PartialEq for TimerDequeueSignalTicket<SIG>
{
    fn eq(&self, other: &Self) -> bool 
    {
        return self.signal == other.signal;
    }
}


impl<SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq> Ord for TimerDequeueSignalTicket<SIG>
{
    fn cmp(&self, other: &Self) -> std::cmp::Ordering 
    {
        return self.timeout_absolute.cmp(&other.timeout_absolute);
    }
}

impl<SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq> PartialOrd for TimerDequeueSignalTicket<SIG>
{
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> 
    {
        return Some(self.cmp(other));
    }
}

impl<SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq> OrderedTimerDequeIntrf for TimerDequeueSignalTicket<SIG>
{
    type Target = SIG;
    type Ticket = common::NoTicket;

    #[inline]
    fn wrap(target: Self::Target, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<(Self, Self::Ticket)> 
    {
        return Ok((Self::new(target, abs_time_sec, abs_time_nsec)?, common::NoTicket));
    }

    #[inline]
    fn get_timeout_absolute(&self) -> TimerExpMode 
    {
        return self.timeout_absolute;
    }
}

impl<SIG> OrderedTimerDeque<TimerDequeueSignalTicket<SIG>>
where SIG: TimerDequeueSignal + fmt::Debug + fmt::Display + Eq + PartialEq
{
    /// Adds the new absolute timeout to the timer dequeue.
    /// 
    /// # Arguemnts
    /// 
    /// `sig_hnd` - a `SIG` generic instance which would identify the instance and
    ///     provide identification.
    /// 
    /// `abs_time_sec` - a [i64] absolute time in seconds. Can not be 0.
    /// 
    /// `abs_time_nsec` - a [i64] asolute time in nanosecodns. Can be 0.
    /// 
    /// # Returns
    /// 
    /// A [Result] as alias [TimerResult] is returned with:
    /// 
    /// * [Result::Ok] with the [TimerDequeueTicket] ticket.
    /// 
    /// * [Result::Err] with error description.
    #[inline]
    pub   
    fn add_to_timer(&mut self, sig_hnd: SIG, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<common::NoTicket>
    {
        return self.add_to_timer_local(sig_hnd, abs_time_sec, abs_time_nsec);
    }

    /// Removes the instance from the queue by the `SIG::TimerQueueID` ID.
    /// 
    /// # Arguments
    /// 
    /// `arg_uniq_id` - a reference to instance's ID `SIG::TimerQueueID`.
    /// 
    /// # Returns
    /// 
    /// A [Result] as alias [TimerResult] is returned with:
    /// 
    /// * [Result::Ok] witout any innder data.
    /// 
    /// * [Result::Err] with error description.
    pub 
    fn remove_from_sched_queue(&mut self, arg_uniq_id: &SIG::TimerQueueID) -> TimerResult<Option<()>>
    {
        if self.deque_timeout_list.len() == 0
        {
            timer_err!(TimerErrorType::QueueEmpty, "queue list is empty!");
        }
        else
        {
            // search for the item in list
            if self.deque_timeout_list.len() == 1
            {
                // just pop the from the front
                let _ = self.deque_timeout_list.pop_front().unwrap();

                // stop timer
                self.stop_timer()?;

                return Ok(Some(()));
            }
            else
            {
                // in theory the `ticket` is a reference to ARC, so the weak should be upgraded 
                // succesfully for temoved instance.

                for (pos, q_item) 
                in self.deque_timeout_list.iter().enumerate()
                {
                    if &q_item.signal.get_id() == arg_uniq_id
                    {
                        // remove by the index
                        let _ = 
                            self.deque_timeout_list.remove(pos);

                        // call timer reset if index is 0 (front)
                        if pos == 0 
                        {
                            self.reschedule_timer()?;
                        }

                        return Ok(Some(()));
                    }
                }

                return Ok(None);
            }
        }
    }

    /// Handles the event which was `read` from the timer. The functions [Self::wait_for_event]
    /// or [Self::poll] can be used to obtain the event.
    /// 
    /// # Arguments
    /// 
    /// `res` - [TimerReadRes] an event from the timer to handle.
    /// 
    /// `timeout_items` - a vector of tickets which were not dropped and timeout.
    /// 
    /// A [Result] as alias [TimerResult] is returned with:
    /// 
    /// * [Result::Ok] witout any innder data.
    /// 
    /// * [Result::Err] with error description.
    pub 
    fn timeout_event_handler(&mut self, res: TimerReadRes<u64>) -> TimerResult<()>
    {
        // ignore wouldblock
        if let TimerReadRes::WouldBlock = res
        {
            return Ok(());
        }
        
        // check if the time reset is required
        let force_timer_resched = 
            TimerReadRes::Cancelled == res || TimerReadRes::Ok(0) == res;

        let mut error: TimerResult<()> = Ok(());
        let mut last_time_until: TimerExpMode = TimerExpMode::None;

        let cur_timestamp = common::get_current_timestamp();

        loop 
        {
            // get from front of the queue
            let Some(front_entity) = 
                self.deque_timeout_list.front() 
                else 
                { 
                    self.stop_timer()?;

                    return Ok(());
                };

            let time_until = front_entity.get_time_until();

            if time_until <= TimerExpMode::from(cur_timestamp)
            {
                let front_entity = self.deque_timeout_list.pop_front().unwrap();

                // send signal
                error = front_entity.send_sig_timeout();
            }
            else
            {
                // we need to rearm the timer if so
                if last_time_until != TimerExpMode::None || force_timer_resched == true
                {
                    // call timer reshedule
                    self.reschedule_timer()?;
                }

                break;
            }

            last_time_until = time_until;
        } // loop


        return error;
    }

    /// An async version of the event handler.
    /// 
    /// Handles the event which was `read` from the timer. The functions [Self::wait_for_event]
    /// or [Self::poll] can be used to obtain the event.
    /// 
    /// # Arguments
    /// 
    /// `res` - [TimerReadRes] an event from the timer to handle.
    /// 
    /// `timeout_items` - a vector of tickets which were not dropped and timeout.
    /// 
    /// A [Result] as alias [TimerResult] is returned with:
    /// 
    /// * [Result::Ok] witout any innder data.
    /// 
    /// * [Result::Err] with error description.
    pub async
    fn async_timeout_event_handler(&mut self, res: TimerReadRes<u64>) -> TimerResult<()>
    {
        // ignore wouldblock
        if let TimerReadRes::WouldBlock = res
        {
            return Ok(());
        }
        
        // check if the time reset is required
        let force_timer_resched = 
            TimerReadRes::Cancelled == res || TimerReadRes::Ok(0) == res;

        let mut error: TimerResult<()> = Ok(());
        let mut last_time_until: TimerExpMode = TimerExpMode::None;

        let cur_timestamp = common::get_current_timestamp();

        loop 
        {
            // get from front of the queue
            let Some(front_entity) = 
                self.deque_timeout_list.front() 
                else 
                { 
                    self.stop_timer()?;

                    return Ok(());
                };

            let time_until = front_entity.get_time_until();

            if time_until <= TimerExpMode::from(cur_timestamp)
            {
                let front_entity = self.deque_timeout_list.pop_front().unwrap();

                // send signal
                error = front_entity.async_send_sig_timeout().await;
            }
            else
            {
                // we need to rearm the timer if so
                if last_time_until != TimerExpMode::None || force_timer_resched == true
                {
                    // call timer reshedule
                    self.reschedule_timer()?;
                }

                break;
            }

            last_time_until = time_until;
        } // loop


        return error;
    }

}


#[cfg(test)]
mod tests
{
    use std::{collections::VecDeque, fmt, sync::mpsc::{self, SendError, Sender}, time::{Duration, Instant}};

    use crate::{common, timer::OrderedTimerDeque, timer_portable::timer::TimerReadRes, timer_signal::{TimerDequeueSignal, TimerDequeueSignalTicket}};

    #[test]
    fn test_timer_test()
    {
        #[derive(Debug)]
        struct TestSigStruct
        {
            uniq_id: u64,
            mpsc_sender: Sender<u64>,
        }

        impl Eq for TestSigStruct{}

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

        impl fmt::Display for TestSigStruct
        {
            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
            {
                write!(f, "sig handler timer id: {}", self.uniq_id)
            }
        }

        impl TimerDequeueSignal for TestSigStruct
        {
            type TimerQueueID = u64;
        
            type TimeoutErr = SendError<Self::TimerQueueID>;
        
            fn get_id(&self) -> Self::TimerQueueID 
            {
                return self.uniq_id;
            }
        
            fn sig_timeout(self) -> Result<(), Self::TimeoutErr> 
            {
                return self.mpsc_sender.send(self.uniq_id);
            }
        }

         let mut time_list = 
            OrderedTimerDeque
                ::<TimerDequeueSignalTicket<TestSigStruct>>
                ::new("test_label".into(), 4, false).unwrap();


        let s = Instant::now();

        let tss_set = common::get_current_timestamp().timestamp()+2;

        let (snd, rcv) = mpsc::channel::<u64>();

        let sig_tm = TestSigStruct{ uniq_id: 456, mpsc_sender: snd };

        println!("signal handler: {}", sig_tm);
        
        let _ = time_list.add_to_timer(sig_tm, tss_set, 0).unwrap();

        for _ in 0..3
        {
            let event = time_list.wait_for_event().unwrap();
            if let TimerReadRes::WouldBlock = event
            {
                std::thread::sleep(Duration::from_millis(1001));

                continue;
            }

            assert_eq!(event, TimerReadRes::Ok(1));

            let e = s.elapsed();
            let ts = common::get_current_timestamp();

            assert_eq!(ts.timestamp(), tss_set);
            println!("instant: {:?} ts: {}, timerset: {}", e, ts.timestamp(), tss_set);

            time_list.timeout_event_handler(event).unwrap();

            let timer_id = rcv.recv_timeout(Duration::from_secs(1)).unwrap();

            assert_eq!(timer_id, 456);

            return;
        }

        panic!("timeout befor timer!");
    }

    #[test]
    fn test_pop()
    {
        let mut ve = VecDeque::<[u8; 32]>::with_capacity(5);

        ve.push_back([0_u8; 32]);
        ve.push_back([1_u8; 32]);
        ve.push_back([2_u8; 32]);
        ve.push_back([3_u8; 32]);

        let mut small: Option<Duration> = None;
        let mut large: Option<Duration> = None;
        let mut some_cnt: u32 = 0;
        let mut mean: u128 = 0;
        
        for _ in 0..5000
        {
            let s = Instant::now();

            let v = ve.pop_front().unwrap();

            if v.len() == 32
            {
                some_cnt += 1;
            }

            ve.push_front(v);

            let e = s.elapsed();

            if small.is_none() == true
            {
                small = Some(e);
            }
            else if small.as_ref().unwrap() > &e
            {
                small = Some(e);
            }

            if large.is_none() == true
            {
                large = Some(e);
            }
            else if large.as_ref().unwrap() < &e
            {
                large = Some(e);
            }

            mean += e.as_nanos();
        }

        println!("s: {:?}, e: {:?}, mean: {} some_cnt: {}", small.unwrap(), large.unwrap(), mean/100, some_cnt);
    }

    #[test]
    fn test_pop2()
    {
        let mut ve = VecDeque::<[u8; 32]>::with_capacity(5);

        ve.push_back([0_u8; 32]);
        ve.push_back([1_u8; 32]);
        ve.push_back([2_u8; 32]);
        ve.push_back([3_u8; 32]);

        let mut small: Option<Duration> = None;
        let mut large: Option<Duration> = None;
        let mut some_cnt: u32 = 0;
        let mut mean: u128 = 0;
        
        for _ in 0..5000
        {
            let s = Instant::now();

            let v = ve.front().unwrap();

            if v.len() == 32
            {
                some_cnt += 1;
            }

            let v = ve.pop_front().unwrap();
            

            let e = s.elapsed();
            ve.push_front(v);

            if small.is_none() == true
            {
                small = Some(e);
            }
            else if small.as_ref().unwrap() > &e
            {
                small = Some(e);
            }

            if large.is_none() == true
            {
                large = Some(e);
            }
            else if large.as_ref().unwrap() < &e
            {
                large = Some(e);
            }

            mean += e.as_nanos();
        }

        println!("s: {:?}, e: {:?}, mean: {} some_cnt: {}", small.unwrap(), large.unwrap(), mean/100, some_cnt);
    }
}