timer_deque_rs/timer_portable/linux/

timer_poll.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, 
    os::fd::{AsFd, AsRawFd, RawFd}, 
    sync::{atomic::{AtomicBool, AtomicUsize, Ordering}, Arc}
};

use nix::
{
    poll::PollTimeout, 
    sys::
    {
        epoll::{Epoll, EpollCreateFlags, EpollEvent, EpollFlags}, 
        eventfd::{EfdFlags, EventFd}
    }
};

use crate::
{
    error::{TimerErrorType, TimerResult}, 
    map_timer_err, 
    timer_err, 
    timer_portable::poll::{AsTimerFd, PollEventType, PollInterrupt, PollResult, TimerPollOps, TimerPollOpsUnregister}
};

/// A [Epoll] based event monitor.
/// 
/// * `0` - [Epoll] instance.
/// 
/// * `1` - [AtomicUsize] amount of timers added
#[derive(Debug)]
pub struct TimerEventWatch
{
    epoll: Epoll, 

    wakeup_event: Arc<EventFd>,

    add_event: EventFd,

    cancel_event: EventFd,

    polling_flag: AtomicBool,

    /// Amount of added FDs without internal
    epoll_fds: AtomicUsize
}

impl TimerPollOpsUnregister for TimerEventWatch
{
    #[inline]
    fn unregister<T: AsTimerFd>(&self, timer: &T) -> TimerResult<()> 
    {
        return <Self as TimerPollOps>::delete(&self, timer);
    }
}

impl fmt::Display for TimerEventWatch
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        write!(f, "fd:{}, cnt:{}", self.epoll.0.as_fd().as_raw_fd(), self.epoll_fds.load(Ordering::Acquire))
    }
}

impl Eq for TimerEventWatch{}

impl PartialEq for TimerEventWatch
{
    fn eq(&self, other: &Self) -> bool 
    {
        return self.epoll.0.as_raw_fd() == other.epoll.0.as_raw_fd();
    }
}

impl TimerPollOps for TimerEventWatch
{
    fn new() -> TimerResult<Self>
    {
        let epoll = 
            Epoll::new(EpollCreateFlags::empty())
                .map_err(|e|
                    map_timer_err!(TimerErrorType::EPoll(e), "epoll new() error")
                )?;

        let add_event = 
            EventFd::from_flags(EfdFlags::EFD_CLOEXEC | EfdFlags::EFD_NONBLOCK)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::EPoll(e), "epoll new() error")
                )?;

        let cancel_event = 
            EventFd::from_flags(EfdFlags::EFD_CLOEXEC | EfdFlags::EFD_NONBLOCK)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::EPoll(e), "epoll new() error")
                )?;

        let wakeup_event = 
            EventFd::from_flags(EfdFlags::EFD_CLOEXEC | EfdFlags::EFD_NONBLOCK)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::EPoll(e), "epoll new() error")
                )?;

        epoll
            .add(
                &add_event, 
                EpollEvent::new(EpollFlags::EPOLLIN,add_event.as_raw_fd() as u64)
            )
            .map_err(|e|
                map_timer_err!(TimerErrorType::EPoll(e), "epoll new() error")
            )?;

        epoll
            .add(
                &cancel_event, 
                EpollEvent::new(EpollFlags::EPOLLIN,cancel_event.as_raw_fd() as u64)
            )
            .map_err(|e|
                map_timer_err!(TimerErrorType::EPoll(e), "epoll new() error")
            )?;

         epoll
            .add(
                &wakeup_event, 
                EpollEvent::new(EpollFlags::EPOLLIN,wakeup_event.as_raw_fd() as u64)
            )
            .map_err(|e|
                map_timer_err!(TimerErrorType::EPoll(e), "epoll new() error")
            )?;

        return Ok( 
            Self
            {
                epoll: epoll, 
                wakeup_event: Arc::new(wakeup_event),
                add_event: add_event, 
                cancel_event: cancel_event,
                polling_flag: AtomicBool::new(false),
                epoll_fds: AtomicUsize::new(0)
            } 
        );
    }

    fn add<T: AsTimerFd>(&self, timer: &T) -> TimerResult<()>
    {
        // notify the poll (if running that it should exit)
        if self.polling_flag.load(Ordering::Acquire) == true
        {
            self
                .add_event
                .write(timer.as_fd().as_raw_fd() as u64)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::EPoll(e), "error while cancel_event for timer {} to epoll", timer)
                )?;
        }

        self
            .epoll
            .add(
                timer, 
                EpollEvent::new(EpollFlags::EPOLLIN, timer.as_fd().as_raw_fd() as u64)
            )
            .map_err(|e|
                map_timer_err!(TimerErrorType::EPoll(e), "can not add timer {} to epoll", timer)
            )?;

        self.epoll_fds.fetch_add(1, Ordering::AcqRel);

        return Ok(());
    }
    
    fn delete<T: AsFd + fmt::Display>(&self, timer: &T) -> TimerResult<()>
    { 
        // notify the poll (if running that it should exit)
        if self.polling_flag.load(Ordering::Acquire) == true
        {
            self
                .cancel_event
                .write(timer.as_fd().as_raw_fd() as u64)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::EPoll(e), "error while cancel_event for timer {} to epoll", timer)
                )?;
        }

        self
            .epoll
            .delete(timer)
            .map_err(|e|
                map_timer_err!(TimerErrorType::EPoll(e), "can not delete timer {} to epoll", timer)
            )?;

        self.epoll_fds.fetch_sub(1, Ordering::AcqRel);

        return Ok(());
    }

    fn poll(&self, timeout: Option<i32>) -> TimerResult<PollResult>
    {
        if self.polling_flag.swap(true, Ordering::AcqRel) == true
        {
            timer_err!(TimerErrorType::EPollAlreadyPolling, 
                "epoll fd: '{}' other thread already polling", self.epoll.0.as_raw_fd());
        }

        let mut events = vec![EpollEvent::empty(); self.epoll_fds.load(Ordering::Acquire) + 1];
        
        let poll_timeout =
            timeout
                .map_or(Ok(PollTimeout::NONE), |f| PollTimeout::try_from(f))
                .map_err(|e|
                    map_timer_err!(TimerErrorType::Conversion, "timeout value: '{:?}' is incorrect: '{}'", timeout, e)   
                )?;

        // reset previous events which happened before poll
        //let _ = self.cancel_event.read();
        
        let evs_res = 
            self
                .epoll
                .wait(events.as_mut_slice(), poll_timeout)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::EPoll(e), "poll error")   
                );

        //let mut poll_res = PollResult::None;

        self.polling_flag.store(false, Ordering::Release);
        
        let evs = evs_res?;
        if evs == 0
        {
            return Ok(PollResult::new_none());
        }
        else
        {
            let mut poll_res = PollResult::new(evs);

            for event in events[..evs].iter()
            {
                if event.data() as RawFd == self.cancel_event.as_raw_fd()
                {
                    let ev = 
                        self
                            .cancel_event
                            .read()
                            .map_or_else(
                                |err| PollEventType::SubError(map_timer_err!(TimerErrorType::EPoll(err), "cancel_event read error")), 
                                |timer_fd| PollEventType::TimerRemoved(timer_fd as RawFd)
                            );

                    poll_res.push(ev);
                }
                else if event.data() as RawFd == self.wakeup_event.as_raw_fd()
                {
                    let _ = self.wakeup_event.read();
                    // no event
                }
                else if event.data() as RawFd == self.add_event.as_raw_fd()
                {
                    let _ = self.add_event.read();
                    // no event
                }
                else
                {
                    poll_res.push(PollEventType::Some(event.data() as RawFd));
                }
            }
           
           return Ok(poll_res);
        }
    }
    
    fn get_count(&self) -> usize 
    {
        return self.epoll_fds.load(Ordering::Acquire);
    }

    fn get_poll_interruptor(&self) -> PollInterrupt
    {
        return PollInterrupt::new(Arc::downgrade(&self.wakeup_event));
    }
    
    fn interrupt_poll(&self) -> bool 
    {
        return self.wakeup_event.write(1).is_ok();
    }
}

impl TimerEventWatch
{


}

#[cfg(test)]
mod tests
{
    use std::{cmp::Ordering, os::fd::{AsFd, AsRawFd}, time::Instant};

    use crate::{deque_timeout::OrderdTimerDequeOnce, timer_portable::timer::AbsoluteTime, OrderedTimerDeque, TimerDequeueConsumer};

    use super::*;

    #[test]
    fn test_epoll0()
    {
        let mut time_list = 
            OrderedTimerDeque
                ::<TimerDequeueConsumer<i32, OrderdTimerDequeOnce>>
                ::new("test_label".into(), 4, false).unwrap();

        let epoll = TimerEventWatch::new().unwrap();

        epoll.add(&time_list).unwrap();

        let s = Instant::now();

        let tss_set = AbsoluteTime::now().add_sec(2);
        let ent1: i32  = 1;

        time_list.add_to_timer(ent1, tss_set).unwrap();

        let res = epoll.poll(Some(2001)).unwrap();

        let e = s.elapsed();
        let abs_now = AbsoluteTime::now();

        println!("elapsed: {:?}, now: {}, set: {}", e, abs_now, tss_set);

        println!("{:?}", res);

        assert_eq!(res.is_none(), false);
        assert_eq!(res[0], PollEventType::Some(time_list.as_fd().as_raw_fd()));
        assert_eq!(abs_now.seconds_cmp(&tss_set) == Ordering::Equal, true);
    }
}