timer_deque_rs/deque_timeout/

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

use std::fmt;

use crate::
{
    common, 
    deque_timeout::{OrderdTimerDequeOnce, OrderdTimerDequePeriodic, OrderedTimerDequeMode}, 
    error::{TimerErrorType, TimerResult}, 
    map_timer_err, 
    timer_err, 
    timer_portable::timer::{AbsoluteTime, RelativeTime}, 
    TimerReadRes
};

use super::{OrderedTimerDeque, OrderedTimerDequeIntrf};

/// 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: fmt::Debug + fmt::Display + Eq + PartialEq
{
    /// 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(&mut 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(&mut 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].
/// 
/// `MODE` - a [OrderedTimerDequeMode] which defines the deque behaviour. There are
///     two types of the operation:
/// 
/// * [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.
/// 
/// # Example
/// 
/// ```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 TimerDequeueSignalTicket<SIG: TimerDequeueSignal, MODE: OrderedTimerDequeMode>
{
    /// A notification and id instance.
    signal: SIG,
    
    /// A timeout depending on the mode of the operation
    timeout_mode: MODE
}


/// implemetation for the [OrderdTimerDequeOnce]
impl<SIG> TimerDequeueSignalTicket<SIG, OrderdTimerDequeOnce> 
where SIG: TimerDequeueSignal
{
    pub 
    fn new(sig_hnd: SIG, abs_time: AbsoluteTime) -> TimerResult<Self>
    {
        let cur = AbsoluteTime::now();

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

        
        return Ok(
            Self
            {
                signal: 
                    sig_hnd,
                timeout_mode: 
                    OrderdTimerDequeOnce::new(abs_time)
            }
        );
    }
}

/// implemetation for the [OrderdTimerDequePeriodic]
impl<SIG> TimerDequeueSignalTicket<SIG, OrderdTimerDequePeriodic> 
where SIG: TimerDequeueSignal
{
    pub 
    fn new(sig_hnd: SIG, rel_time: RelativeTime) -> TimerResult<Self>
    {        
        return Ok(
            Self
            {
                signal: 
                    sig_hnd,
                timeout_mode:
                    OrderdTimerDequePeriodic::new(rel_time)
            }
        );
    }
}

impl<SIG: TimerDequeueSignal, MODE: OrderedTimerDequeMode> TimerDequeueSignalTicket<SIG, MODE>
{
    fn send_sig_timeout(&mut 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(&mut 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, MODE: OrderedTimerDequeMode> fmt::Display for TimerDequeueSignalTicket<SIG, MODE>
{  
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> 
    { 
        write!(f, "{} until: {}", self.signal.get_id(), self.timeout_mode)
    }
}

impl<SIG: TimerDequeueSignal, MODE: OrderedTimerDequeMode> Eq for TimerDequeueSignalTicket<SIG, MODE> {}

impl<SIG: TimerDequeueSignal, MODE: OrderedTimerDequeMode>  PartialEq for TimerDequeueSignalTicket<SIG, MODE>
{
    fn eq(&self, other: &Self) -> bool 
    {
        return self.signal == other.signal;
    }
}


impl<SIG: TimerDequeueSignal, MODE: OrderedTimerDequeMode> Ord for TimerDequeueSignalTicket<SIG, MODE>
{
    fn cmp(&self, other: &Self) -> std::cmp::Ordering 
    {
        return self.timeout_mode.cmp(&other.timeout_mode);
    }
}

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

impl<SIG: TimerDequeueSignal, MODE: OrderedTimerDequeMode> OrderedTimerDequeIntrf 
for TimerDequeueSignalTicket<SIG, MODE>
{
    type Target = SIG;
    type Ticket = common::NoTicket;

    #[inline]
    fn get_timeout_absolute(&self) -> AbsoluteTime
    {
        return self.timeout_mode.get_absolut_timeout();
    }
}

impl<SIG> OrderedTimerDeque<TimerDequeueSignalTicket<SIG, OrderdTimerDequeOnce>> 
where SIG: TimerDequeueSignal
{
    /// Adds the new `absolute` (`relative`) timeout to the timer dequeue.
    /// 
    /// # Arguemnts
    /// 
    /// `sig_hnd` - a `SIG` generic instance which would identify the instance and
    ///     provide identification.
    /// 
    /// `abs_time` - a [AbsoluteTime] absolute time in future when the timeout
    ///     should happen.
    /// 
    /// # 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: AbsoluteTime) -> TimerResult<common::NoTicket>
    {
        let inst = 
            TimerDequeueSignalTicket::<SIG, OrderdTimerDequeOnce>::new(sig_hnd, abs_time)?;

        return self.add_to_timer_local(inst).map(|_| common::NoTicket);
    }

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

        let cur_timestamp = AbsoluteTime::now();

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

            let time_until = front_entity.get_timeout_absolute();

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

                // send signal
                front_entity.send_sig_timeout()?;
            }
            else
            {
                break;
            }
        } // loop

        // call timer reschedule
        self.reschedule_timer()?;

        return Ok(());
    }

    /// 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(());
        }
        
        let cur_timestamp = AbsoluteTime::now();

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

            let time_until = front_entity.get_timeout_absolute();

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

                // send signal
                front_entity.async_send_sig_timeout().await?;
            }
            else
            {
                break;
            }
        } // loop

        // call timer reschedule
        self.reschedule_timer()?;

        return Ok(());
    }
}


impl<SIG> OrderedTimerDeque<TimerDequeueSignalTicket<SIG, OrderdTimerDequePeriodic>> 
where SIG: TimerDequeueSignal
{
    /// Adds the new `periodic` timeout to the timer dequeue.
    /// 
    /// # Arguemnts
    /// 
    /// `sig_hnd` - a `SIG` generic instance which would identify the instance and
    ///     provide identification.
    /// 
    /// `rel_time` - a [RelativeTime] which will be used to extend the further 
    ///     timeouts. 
    /// 
    /// # 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, rel_time: RelativeTime) -> TimerResult<common::NoTicket>
    {
        let inst = 
            TimerDequeueSignalTicket::<SIG, OrderdTimerDequePeriodic>::new(sig_hnd, rel_time)?;

        return self.add_to_timer_local(inst).map(|_| common::NoTicket);
    }

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

        let cur_timestamp = AbsoluteTime::now();

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

            let time_until = front_entity.get_timeout_absolute();

            if time_until <= cur_timestamp
            {
                let mut deq = self.deque_timeout_list.pop_front().unwrap();

                // send signal
                deq.send_sig_timeout()?;

                // advance timer
                deq.timeout_mode.advance_timeout();

                // return back to deque with new timeout
                self.add_to_timer_local(deq)?;
            }
            else
            {
                break;
            }
        } // loop

        // call timer reschedule
        self.reschedule_timer()?;

        return Ok(());
    }

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

        let cur_timestamp = AbsoluteTime::now();

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

            let time_until = front_entity.get_timeout_absolute();

            if time_until <= cur_timestamp
            {
                let mut deq = self.deque_timeout_list.pop_front().unwrap();

                // send signal
                deq.async_send_sig_timeout().await?;

                // advance timeout
                deq.timeout_mode.advance_timeout();

                // return back to deque with new timeout
                self.add_to_timer_local(deq)?;
            }
            else
            {
                break;
            }
        } // loop

        // call timer reschedule
        self.reschedule_timer()?;

        return Ok(());
    }
}

impl<SIG, MODE> OrderedTimerDeque<TimerDequeueSignalTicket<SIG, MODE>> 
where SIG: TimerDequeueSignal, MODE: OrderedTimerDequeMode
{
    

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


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

    use crate::{common, deque_timeout::{OrderdTimerDequeOnce, OrderdTimerDequePeriodic}, timer_portable::timer::{AbsoluteTime, RelativeTime, TimerReadRes}, OrderedTimerDeque, TimerDequeueSignal, TimerDequeueSignalTicket};

#[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(&mut self) -> Result<(), Self::TimeoutErr> 
        {
            return self.mpsc_sender.send(self.uniq_id);
        }
    }

    #[test]
    fn test_timer_test()
    {
        

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


        let s = Instant::now();
        let ts = common::get_current_timestamp();

        let tss_set = 
            AbsoluteTime
                ::new_time(ts.timestamp()+2, ts.timestamp_subsec_nanos() as i64)
                    .unwrap();

        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.clone()).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 = AbsoluteTime::now();

            assert_eq!(ts.seconds_cmp(&tss_set) == Ordering::Equal, true);
            println!("ev: {}, instant: {:?} ts: {}, timerset: {}", event, e, ts, 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_timer_period_test()
    {
        

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

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

        let sig_tm = TestSigStruct{ uniq_id: 456, mpsc_sender: snd.clone() };
        let tss_set = RelativeTime::from(Duration::from_secs(2));

        let sig_tm2 = TestSigStruct{ uniq_id: 457, mpsc_sender: snd };
        let tss_set2 = RelativeTime::from(Duration::from_secs(2));

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

        for _ in 0..3
        {
            let abs_time_before = AbsoluteTime::now();
            let s = Instant::now();
            std::thread::sleep(Duration::from_millis(2001));

            let event = time_list.wait_for_event().unwrap();
            let TimerReadRes::Ok(cnt) = event
                else 
                {
                    panic!("incorrect event res: {}", event);
                };

            assert_eq!(cnt, 1);

            let e = s.elapsed();
            let ts = AbsoluteTime::now() - abs_time_before;

            println!("ev: {}, instant: {:?} ts: {}, timerset: {}", event, e, ts, tss_set);

            assert_eq!(ts.seconds_cmp(&AbsoluteTime::new_time(2, 0).unwrap()) == Ordering::Equal, true);

            time_list.timeout_event_handler(event).unwrap();

            while let Ok(timer_id) = rcv.recv_timeout(Duration::from_millis(1))
            {
                println!("received timer item ID: {}", timer_id);
                assert_eq!(timer_id == 456 || timer_id == 457, true);
            }

            
        }

        time_list.remove_from_sched_queue(&456).unwrap();

        return;
    }
    /*
    #[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);
    }*/
}