timer-deque-rs 0.8.0

/*-
 * 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. The MIT License (MIT)
 *                     
 *   3. EUROPEAN UNION PUBLIC LICENCE v. 1.2 EUPL © the European Union 2007, 2016
 */

/// A timer polling based on IOPC.

use std::
{
    collections::{BTreeMap}, 
    fmt, 
    mem, 
    os::
    {
        raw::c_void, 
        windows::io::{AsRawHandle, FromRawHandle, OwnedHandle, RawHandle}
    }, 
    ptr::null_mut, 
    sync::{Arc, RwLock, atomic::{AtomicBool, Ordering}}
};

use windows::
{
    Win32::
    {
        Foundation::{GENERIC_ALL, HANDLE, INVALID_HANDLE_VALUE}, 
        System::{IO::{CreateIoCompletionPort, GetQueuedCompletionStatusEx, OVERLAPPED_ENTRY}, Threading::INFINITE}
    }, 
    core::HRESULT
};

use crate::
{
    TimerFd, 
    TimerReadRes, 
    error::{TimerErrorType, TimerResult}, 
    map_timer_err, 
    timer_err, 
    timer_portable::{FdTimerMarker, PollEventType, TimerId, TimerPollOps, poll::PollInterrupt, windows::EventFd}
};

/// FFI for missing items in `windows` crate.
pub mod nt_crap_not_covered_by_crate
{
    use std::os::{raw::c_void, windows::io::RawHandle};

    use windows::Win32::{Foundation::NTSTATUS};

    #[allow(non_camel_case_types)]
    #[allow(non_snake_case)]
    #[repr(C)]
    #[derive(Debug)]
    pub struct UNICODE_STRING 
    {
        Length: u16,
        MaximumLength: u16,
        Buffer: *mut u16,
    }

    #[allow(non_camel_case_types)]
    #[allow(non_snake_case)]
    #[repr(C)]
    #[derive(Debug)]
    pub struct OBJECT_ATTRIBUTES 
    {
        Length: u32,
        RootDirectory: RawHandle,
        ObjectName: *mut UNICODE_STRING,
        Attributes: u32,
        SecurityDescriptor: *mut c_void,
        SecurityQualityOfService: *mut c_void,
    }

    #[link(name = "ntdll")]
    unsafe extern "system" 
    {
        pub unsafe fn NtCreateWaitCompletionPacket(
            IoCompletionHandle: RawHandle, // out
            DesiredAccess: u32,
            ObjectAttributes: *mut OBJECT_ATTRIBUTES,
        ) -> NTSTATUS;

        pub unsafe fn NtAssociateWaitCompletionPacket(
            WaitCompletionPacketHandle: RawHandle,
            IoCompletionHandle: RawHandle,
            TargetObjectHandle: RawHandle,
            KeyContext: *mut c_void,
            ApcContext: *mut c_void,
            IoStatus: i32,
            IoStatusInformation: usize,
            AlreadySignaled: *mut u8,
        ) -> NTSTATUS;
    }

    /*#[link(name = "kernel32.dll", kind = "raw-dylib", modifiers = "+verbatim")]
    unsafe extern "system" 
    {
        #[link_name="GetQueuedCompletionStatusEx"]
        pub unsafe fn GetQueuedCompletionStatusEx(
            CompletionPort: RawHandle,
            lpCompletionPortEntries: *mut OVERLAPPED_ENTRY,
            ulCount: u32,
            ulNumEntriesRemoved: *mut u32,
            dwMilliseconds: u32,
            fAlertable: i32,
        ) -> i32;
    }*/
}


/// A wrapper around the [CreateIoCompletionPort] [HANDLE].
/// 
/// The solution was adopted from the 
/// [Win32: Processing Event Objects through I/O Completion Port](https://github.com/tringi/win32-iocp-events)
///
/// This method is **NOT** limited to [windows::Win32::System::SystemServices::MAXIMUM_WAIT_OBJECTS] HANDLER
/// and not based on `WaitMultipleObjects`. Another "genious" technology is used which is called IOCP.
#[derive(Debug)]
pub struct IocpCrap
{
    /// A handler to [CreateIoCompletionPort].
    handle: OwnedHandle,
}

impl fmt::Display for IocpCrap
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        write!(f, "IOCP:{}", self.handle.as_raw_handle() as usize)
    }
}

impl AsRawHandle for IocpCrap
{
    fn as_raw_handle(&self) -> RawHandle 
    {
        return self.handle.as_raw_handle();
    }
}

impl IocpCrap
{
    /// Initializes a new instance of `IoCompletionPort`.
    fn new() -> TimerResult<Self>
    {
        let iopc = 
            unsafe 
            {
                CreateIoCompletionPort(INVALID_HANDLE_VALUE, None, 0, 0)
                    .map_err(|e|
                        map_timer_err!(TimerErrorType::EPoll(e), "CreateIoCompletionPort failed")

                    )?
            };
           
        return Ok( 
            Self 
            { 
                handle: unsafe { OwnedHandle::from_raw_handle(iopc.0) },
            } 
        );
    }

    /// Waits for the events. The caller should provide a list of [RawHandle]s to 
    /// the method. 
    /// 
    /// # Arguments 
    /// 
    /// * `timer_handlers` - a slice of [RawHandle] with handlers to `event timers` and
    /// `events`.
    /// 
    /// * `timeout` - a time spent on polling the completions on the port. 
    /// 
    ///     - [Option::Some] sets the timeout in miliseconds. If value eq 0, immidiate return
    ///         without blocking the thread.
    /// 
    ///     - [Option::None] sets the timeout to [INFINITE] (blocking) until error or result.
    /// 
    /// # Returns
    /// 
    /// * [Result::Ok] - if operation was successfull. The inner type is a vector of time out
    ///     elements. If [Vec] is empty, then timeout happened before any source fired an event.
    /// 
    /// * [Result::Err] - with error description.
    fn wait(&self, timer_handlers: &[RawHandle], timeout: Option<u32>) -> TimerResult<Vec<TimerId>>
    {
        // completion packet
        //let mut hpacks_map: HashMap<CompPackWrap, RawHandle> = HashMap::with_capacity(timer_handlers.len());

        // a list of assosiations
        let mut hpacks: Vec<OwnedHandle> = Vec::with_capacity(timer_handlers.len());

        for (index, th) in timer_handlers.iter().enumerate()
        {
            let mut hps: *mut c_void = null_mut();

            unsafe
            {
                nt_crap_not_covered_by_crate::
                    NtCreateWaitCompletionPacket(
                        &mut hps as *mut _ as *mut _, 
                        GENERIC_ALL.0, 
                        null_mut())
                .ok()
                .map_err(|e|
                    map_timer_err!(TimerErrorType::EPoll(e), "NtCreateWaitCompletionPacket failed 1")

                )?;
            }; 

            let mut flag: u8 = 0;

            let hpack = unsafe { OwnedHandle::from_raw_handle(hps) };

            unsafe
            {
                nt_crap_not_covered_by_crate::
                    NtAssociateWaitCompletionPacket(hpack.as_raw_handle(), self.handle.as_raw_handle(), *th, null_mut(), null_mut(), 0,
                        index, &mut flag as *mut u8)
                    .ok()
                    .map_err(|e|
                        map_timer_err!(TimerErrorType::EPoll(e), "NtAssociateWaitCompletionPacket failed")

                    )?;
            }

            hpacks.push(hpack);
        }

        
        // setting poll timeout
        let poll_timeout = timeout.map_or(INFINITE, |f| f);


        let mut overlp_res: Vec<OVERLAPPED_ENTRY> = vec![unsafe { mem::zeroed()}; timer_handlers.len()]; 
        let mut complitions: u32 = 0;

        // poll

        let res = 
            unsafe
            {
                //nt_crap_not_covered_by_crate::
                    GetQueuedCompletionStatusEx(
                        HANDLE(self.handle.as_raw_handle()), 
                        &mut overlp_res,
                        &mut complitions as *mut _, 
                        poll_timeout, 
                        false.into()
                    )
                    /*.map_err(|e|
                        map_timer_err!(TimerErrorType::EPoll(e), "GetQueuedCompletionStatusEx failed")
                    )?;*/
            };

        // timer result
        let mut pet: Vec<TimerId> = Vec::with_capacity(complitions as usize);

        // check if timeout happened
        const ERR_TIMEOUT: HRESULT = HRESULT::from_win32(0x80070102_u32);

        if let Err(ref e) = res && e.code() == ERR_TIMEOUT
        {
            // wouldblock
            return Ok(pet);
        }

        // return mapped error if any
        res
            .map_err(|e|
                map_timer_err!(TimerErrorType::EPoll(e), "GetQueuedCompletionStatusEx failed")

            )?;

        

        for (ov, _index) in overlp_res.into_iter().zip(0..complitions as usize)
        {
            pet.push(TimerId::from(timer_handlers[ov.dwNumberOfBytesTransferred as usize]));
        }

        return Ok(pet);
    }
}

/// A [IOCP] based event monitor.
#[derive(Debug)]
pub struct TimerEventWatch
{
    /// A event notification facility.
    epoll: IocpCrap, 

    // A waker which interrupts the `poll`.
    wakeup_event: Arc<EventFd>,

    /// A poll guard which prevents the multiple call to poll.
    polling_flag: AtomicBool,

    /// All regestered timers.
    timers: RwLock<BTreeMap<TimerId, TimerFd>>,
}

impl fmt::Display for TimerEventWatch
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        write!(f, "fd:{}, cnt:{}", 
            (self.epoll.as_raw_handle() as usize).to_string(), 
            self.timers.try_read().map_or("locked".to_string(), |f| f.len().to_string())
        )
    }
}

impl Eq for TimerEventWatch {}

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

impl TimerPollOps for TimerEventWatch
{
    fn new() -> TimerResult<Self>
    {
        let epoll = IocpCrap::new()?;

        let wakeup_event = 
            EventFd::new("wakeup_event".into())?;

        return Ok( 
            Self
            {
                epoll: epoll, 
                wakeup_event: Arc::new(wakeup_event),
                polling_flag: AtomicBool::new(false),
                timers: RwLock::new(BTreeMap::new())
            } 
        );
    }

    

    fn add(&self, timer: TimerFd) -> TimerResult<()>
    {
        // lock shared lock for exclusive access
        let mut timers_lock = 
            self
                .timers
                .write()
                .map_or_else(|e| e.into_inner(), |v| v);

        // check if timer was added previosly
        let false = 
            timers_lock.contains_key(&(timer.as_raw_handle() as usize))
        else
        {
            timer_err!(TimerErrorType::Duplicate, "can not add timer {} to epoll, reason duplicate", 
                timer)
        };

        //let _ = self.wakeup_event.write(1);

        // store on the list
        timers_lock.insert(TimerId::from(timer.as_raw_handle()), timer);

        return Ok(());
    }
    
    fn delete<FD: FdTimerMarker>(&self, timer: &FD) -> TimerResult<()>
    { 
        let mut timers_lock =
            self
                .timers
                .write()
                .map_or_else(|e| e.into_inner(), |v| v);

        // check if timer was added previosly
        let true =
            timers_lock.contains_key(&timer.as_timer_id())
        else
        {
            timer_err!(TimerErrorType::Duplicate, "can not add timer {} to epoll, reason does not exist",
                timer.as_timer_id())
        };

        // remove from the list
        let _ = timers_lock.remove(&timer.as_timer_id());

        return Ok(());
    }

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

        let poll_ops = ||
            {
                let rd = 
                    self
                        .timers
                        .read()
                        .unwrap_or_else(|e| e.into_inner());

                let mut raw_handlers = 
                    rd.iter().map(|(tid, _)| tid.0 as RawHandle).collect::<Vec<RawHandle>>();

                raw_handlers.push(self.wakeup_event.as_raw_handle());

                drop(rd);

                let res =
                    self
                        .epoll
                        .wait(&raw_handlers, timeout.map(|v| v as u32));

                return res;
            };

        let evs_res =  poll_ops();

        // indicate poll completed.
        self.polling_flag.store(false, Ordering::SeqCst);


        let poll_res: Vec<PollEventType> = 
            evs_res?
                .into_iter()
                .map(|res|
                    PollEventType::TimerRes(res, TimerReadRes::ok())

                )
                .collect();

        if poll_res.is_empty() == false
        {
            return Ok(Some(poll_res));
        }
        else
        {
            return Ok(None);
        }
    }
    
    fn get_count(&self) -> usize 
    {
        //return self.epoll_fds.load(Ordering::SeqCst);
        return 
            self
                .timers
                .read()
                .map_or_else(|e| e.into_inner(), |v| v)
                .len();
    }

    fn get_poll_interruptor(&self) -> PollInterrupt
    {
        return PollInterrupt::new(Arc::downgrade(&self.wakeup_event));
    }
    
    /// Interrupts poll genrally.
    fn interrupt_poll(&self) -> bool 
    {
       return self.wakeup_event.write(0).is_ok();
    }
}