timer_deque_rs/periodic_task/

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

use std::
{
    cell::OnceCell, 
    cmp, 
    collections::HashMap, 
    fmt, 
    num::NonZeroUsize,
    sync::
    {
        Arc, Mutex, TryLockError, Weak, atomic::{AtomicBool, Ordering}, mpsc::{self, Receiver, Sender}
    }, 
    thread::JoinHandle, 
    time::Duration
};

use crossbeam_deque::{Injector, Steal};
use rand::random_range;

use crate::
{
    AbsoluteTime, 
    FdTimerCom, 
    RelativeTime, 
    TimerPoll, 
    TimerReadRes, 
    error::{TimerError, TimerErrorType, TimerResult}, 
    map_timer_err, timer_err, 
    timer_portable::
    {
        AsTimerId, PollEventType, PolledTimerFd, TimerExpMode, TimerFlags, TimerId, TimerType, poll::PollInterrupt, timer::TimerFd
    }
};

/// A result of the task execution which is returned by the user task.
/// See description for the each variant.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PeriodicTaskResult
{
    /// The task exec was ok. No errors.
    Ok,

    /// Cancels (but not removing) the task due to error or for other reason.
    /// By returning this result the timer is unset to stop generating events. 
    /// But, if since the `AbortTask` was received and a timer has timed out again 
    /// the event will be called again. It is not necessary to return the `AbortTask` again, 
    /// just return [PeriodicTaskResult::Ok] exiting imidiatly.
    CancelTask,

    /// A request to reschedule the task to a new time or interval. The `0` argument 
    /// should contain new time. The timer mode can be changed.
    TaskReSchedule(PeriodicTaskTime)
}

/// A trait which should be implemented by the instance which will be added as a `task`.
/// 
/// * the `instance` must implement [Send] because it may be moved to other thread.
/// 
/// * it is not necessary that the `instance` to impl [Sync] because the `instance` will never
/// be executed from two different threads and it is already protected by mutex.
/// 
/// A `task` should never block the thread i.e call some functions which may block for a large
/// period of time! 
pub trait PeriodicTask: Send + 'static
{
    /// A task entry point. The task should return a [PeriodicTaskResult].
    fn exec(&mut self) -> PeriodicTaskResult;
}

/// An alias for the boxed value.
pub type PeriodicTaskHndl = Box<dyn PeriodicTask>;


/// A global FIFO commands.
/// 
/// For all variants, the option [Option] [mpsc::Sender] is used for the error feedback and if
/// it is set to [Option::None] the error will not be reported.
#[derive(Debug)]
pub(crate) enum GlobalTasks
{
    /// Adds task to the task system. If error happens it will be reported over the `1` and item `0`
    /// will not be added to the instance.
    AddTask( String, PeriodicTaskTime, Arc<PeriodicTaskGuardInner>, Option<mpsc::Sender<TimerResult<()>>> ),

    /// Removes the task from the system.
    RemoveTask( Arc<PeriodicTaskGuardInner>, Option<mpsc::Sender<TimerResult<()>>> ),

    /// Changes the timer value.
    ReschedTask( Weak<PeriodicTaskGuardInner>, PeriodicTaskTime, Option<mpsc::Sender<TimerResult<()>>> ),

    /// Suspends the task but does not remove the task. The task which is planned to suspend, if it is already
    /// in the exec queue, will be executed.
    SuspendTask( Weak<PeriodicTaskGuardInner>, Option<mpsc::Sender<TimerResult<()>>> ),

    /// Resumes the task by enabling timer.
    ResumeTask( Weak<PeriodicTaskGuardInner>, Option<mpsc::Sender<TimerResult<()>>> ),
}

/// A local task queue commands.
#[derive(Debug)]
pub enum ThreadTask
{
    /// Execute the task.
    TaskExec( Arc<NewTaskTicket> )
}

/// A ticket which is assigned to specific task in order to send the same task to the same
/// thread which has been already assigned. If only one thread is allocated, then this is not
/// really necessary (will be optimized in future).
#[derive(Debug)]
pub struct NewTaskTicket
{
    /// A thread handler of the specific thread.
    task_thread: Arc<ThreadHandler>,

    /// A task. A weak reference is used in order to determine if task was
    /// removed and the exec is not necessary.
    ptgi: Weak<PeriodicTaskGuardInner>,
}

impl NewTaskTicket
{
    fn new(task_thread: Arc<ThreadHandler>, ptgi: Weak<PeriodicTaskGuardInner>) -> Self
    {
        return 
            Self
            {
                task_thread: 
                    task_thread,
                ptgi:
                    ptgi
            };
    }

    /// Sends the task on exec to the local queue of the thread.
    fn send_task(this: Arc<NewTaskTicket>, task_rep_count: u64, thread_hndl_cnt: usize)
    {
        let strong_cnt = Arc::strong_count(&this);
        let thread = this.task_thread.clone();

        for _ in 0..task_rep_count
        {
            thread.send_task(this.clone(), strong_cnt < 2 && thread_hndl_cnt > 1);
        }
    }
}

/// The internal structure which is stored in pair with the timer's FD. Contains necessary references
/// and values.
#[derive(Debug)]
pub(crate) struct PeriodicTaskTicket
{
    /// A task name.
    task_name: String,

    sync_timer: PolledTimerFd<TimerFd>,

    /// A time which was set to timer. Needed in case if timer was cancelled ny OS or suspended.
    ptt: PeriodicTaskTime,

    /// A [Weak] reference to the current [NewTaskTicket] which identifies to which thread the
    /// task exec was assigned. The [Arc] of [NewTaskTicket] is clonned for each task exec round.
    /// If is not `upgradable` the task is no logner assigned to thread.
    weak_ticket: Weak<NewTaskTicket>,

    /// A [Weak] reference to the [PeriodicTaskGuardInner]. If this reference is not `upgradable` 
    /// then the task is no logner valid.
    ptg: Weak<PeriodicTaskGuardInner>,
}

impl Ord for PeriodicTaskTicket 
{
    fn cmp(&self, other: &Self) -> cmp::Ordering 
    {
        return 
            self
                .sync_timer
                .get_inner()
                .as_timer_id()
                .cmp(&other.sync_timer.get_inner().as_timer_id());
    }
}

impl PartialOrd for PeriodicTaskTicket 
{
    fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> 
    {
        return Some(self.cmp(other));
    }
}

impl PartialEq for PeriodicTaskTicket 
{
    fn eq(&self, other: &Self) -> bool 
    {
        return 
            self.task_name == other.task_name && 
            self.sync_timer.get_inner().as_timer_id() == other.sync_timer.get_inner().as_timer_id();
    }
}

impl Eq for PeriodicTaskTicket {}

impl PeriodicTaskTicket
{
    fn new(task_name: String, ptt: PeriodicTaskTime, ptg: Arc<PeriodicTaskGuardInner>, poll: &TimerPoll) -> TimerResult<Self>
    {
        // create timer and add to poll
        let sync_timer = 
            TimerFd::new(task_name.clone().into(), TimerType::CLOCK_REALTIME, 
                TimerFlags::TFD_CLOEXEC | TimerFlags::TFD_NONBLOCK)
                .map_err(|e|
                    map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "cannot setup timer for task: '{}'", task_name)
                )
                .and_then(|timer| 
                    poll.add(timer)
                )?;

        let ptt = 
            Self
            {
                task_name: task_name,
                sync_timer: sync_timer,
                ptt: ptt,
                weak_ticket: Weak::new(),
                ptg: Arc::downgrade(&ptg),
            };

        ptt.set_timer()?;

        return Ok(ptt);
    }

    #[inline]
    fn set_timer(&self) -> TimerResult<()>
    {
        return self.get_timer_time().set_timer(self.sync_timer.get_inner());
    }

    #[inline]
    fn unset_timer(&self) -> TimerResult<()>
    {
        return 
            self
                .sync_timer
                .get_inner()
                .get_timer()
                .unset_time()
                .map_err(|e|
                    map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "unsetting timer '{}' returned error: {}", self.sync_timer, e)
                );
    }

    fn update_task_time(&mut self, ptt_new: PeriodicTaskTime) -> TimerResult<()>
    {
        self.ptt = ptt_new;

        return self.set_timer();
    } 

    fn get_task_guard(&self) -> TimerResult<Arc<PeriodicTaskGuardInner>>
    {
        return 
            self
                .ptg
                .upgrade()
                .ok_or_else(||
                    map_timer_err!(TimerErrorType::ReferenceGone, "task: '{}' reference to timer has gone", 
                        self.task_name)
                );
    }

    #[inline]
    fn get_timer_time(&self) -> &PeriodicTaskTime
    {
        return &self.ptt;
    }
}

/// An instance which is returned by the [SyncPeriodicTasks::add] whuch guards the 
/// task and allows to control its state. If task is no longer needed the instance
/// can be dropped and it will be removed from the system.
#[derive(Debug)]
pub struct PeriodicTaskGuard
{
    /// A task name
    task_name: String,

    /// A [Arc] to [PeriodicTaskGuardInner] which contains timer and other things. 
    /// Normally this is the base instance i.e reference and if it is dropped
    /// it indicates to the task `executor` that the instance is no logner valid.
    /// The autoremoval is perfomed via sending the `command` over the global
    /// qeueu.
    /// 
    /// The [Option] is needed to `take` the instance.
    guard: Option<Arc<PeriodicTaskGuardInner>>,

    /// A clonned instance to the executor spawner which contains the reference 
    /// to channel.
    spt: Arc<SyncPeriodicTasksInner>
}

impl Drop for PeriodicTaskGuard
{
    fn drop(&mut self) 
    {
        let guard = self.guard.take().unwrap();

        let _ = self.spt.send_global_cmd(GlobalTasks::RemoveTask(guard, None));

        return;
    }
}

impl PeriodicTaskGuard
{
    /// Requests the task rescheduling - changine the timer time or mode. 
    /// The `task` is represented by the calling instance. 
    /// 
    /// This function blocks
    /// the current thread for the maximum (in worst case) 5 seconds which is a 
    /// timeout for feedback reception.
    /// 
    /// # Arguments
    /// 
    /// * `ptt` - a new time to be set to timer.
    /// 
    /// # Returns 
    /// 
    /// A [Result] as alias [TimerResult] is returned.
    
    /// In case if error is retuned, the operation should be considered as not completed
    /// correctly and the executor instance is poisoned i.e a bug happened.
    /// 
    /// The common errors may be retuned:
    /// 
    /// * [TimerErrorType::MpscTimeout] - a feedback (with the result) reception timeout.
    ///     Probably this is because the task executor is too busy.
    /// 
    /// * [TimerErrorType::TimerError] - with the timer error.
    /// 
    /// * [TimerErrorType::NotFound] - if instance was not found in the internal records. 
    ///     Should not happen. If appears, then probably this is a bug.
    pub 
    fn reschedule_task(&self, ptt: PeriodicTaskTime) -> TimerResult<()>
    {
        let weak_ptgi = Arc::downgrade(self.guard.as_ref().unwrap());

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

        self.spt.send_global_cmd(GlobalTasks::ReschedTask(weak_ptgi, ptt, Some(snd)))?;

        return 
            rcv
                .recv_timeout(Duration::from_secs(10))
                .map_err(|e|
                    map_timer_err!(TimerErrorType::MpscTimeout, "reschedule_task(), task name: '{}', MPSC rcv timeout error: '{}'", 
                        self.task_name, e)
                )?;
    }

    /// Requests to suspent the current `task`.
    /// 
    /// This function blocks
    /// the current thread for the maximum (in worst case) 5 seconds which is a 
    /// timeout for feedback reception.
    /// 
    /// # Returns 
    /// 
    /// A [Result] as alias [TimerResult] is returned.
    /// 
    /// In case if error is retuned, the operation should be considered as not completed
    /// correctly and the executor instance is poisoned i.e a bug happened.
    /// 
    /// The common errors may be retuned:
    /// 
    /// * [TimerErrorType::MpscTimeout] - a feedback (with the result) reception timeout.
    ///     Probably this is because the task executor is too busy.
    /// 
    /// * [TimerErrorType::TimerError] - with the timer error.
    pub 
    fn suspend_task(&self) -> TimerResult<()>
    {
        let weak_ptgi = Arc::downgrade(self.guard.as_ref().unwrap());

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

        self.spt.send_global_cmd(GlobalTasks::SuspendTask(weak_ptgi, Some(snd)))?;

        return 
            rcv
                .recv_timeout(Duration::from_secs(10))
                .map_err(|e|
                    map_timer_err!(TimerErrorType::MpscTimeout, "suspend_task(), task name: '{}', MPSC rcv timeout error: '{}'", 
                        self.task_name, e)
                )?;
    }

    /// Requests to resume the task from the suspend state.
    /// 
    /// This function blocks
    /// the current thread for the maximum (in worst case) 5 seconds which is a 
    /// timeout for feedback reception.
    /// 
    /// # Returns 
    /// 
    /// A [Result] as alias [TimerResult] is returned.
    /// 
    /// In case if error is retuned, the operation should be considered as not completed
    /// correctly and the executor instance is poisoned i.e a bug happened.
    /// 
    /// The common errors may be retuned:
    /// 
    /// * [TimerErrorType::MpscTimeout] - a feedback (with the result) reception timeout.
    ///     Probably this is because the task executor is too busy.
    /// 
    /// * [TimerErrorType::TimerError] - with the timer error.
    pub 
    fn resume_task(&self) -> TimerResult<()>
    {
        let weak_ptgi = Arc::downgrade(self.guard.as_ref().unwrap());

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

        self.spt.send_global_cmd(GlobalTasks::ResumeTask(weak_ptgi, Some(snd)))?;

        return 
            rcv
                .recv_timeout(Duration::from_secs(10))
                .map_err(|e|
                    map_timer_err!(TimerErrorType::MpscTimeout, "resume_task(), task name: '{}', MPSC rcv timeout error: '{}'", 
                        self.task_name, e)
                )?;
    }
}

/// Programs the taks's timer to specific time and mode. This instance is
/// a wrapper around the [TimerExpMode] as this struct requers the generic to
/// be specified which defines the type of the time.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PeriodicTaskTime
{
    /// A provided time is absolute time in future.
    Absolute(TimerExpMode<AbsoluteTime>),

    /// A provided time is relative to current time.
    Relative(TimerExpMode<RelativeTime>),
}

impl From<TimerExpMode<AbsoluteTime>> for PeriodicTaskTime
{
    fn from(value: TimerExpMode<AbsoluteTime>) -> Self 
    {
        return Self::Absolute(value);
    }
}

impl From<TimerExpMode<RelativeTime>> for PeriodicTaskTime
{
    fn from(value: TimerExpMode<RelativeTime>) -> Self 
    {
        return Self::Relative(value);
    }
}

impl fmt::Display for PeriodicTaskTime
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        match self
        {
            Self::Absolute(t) => 
                write!(f, "{}", t),
            Self::Relative(t) => 
                write!(f, "{}", t),
        }
    }
}

impl PeriodicTaskTime
{
    /// The timer is set to time up once and for the specific `absolute` time in future 
    /// (to current realclock time).
    #[inline]
    pub 
    fn exact_time(abs_time: AbsoluteTime) -> Self
    {
        return Self::Absolute(TimerExpMode::<AbsoluteTime>::new_oneshot(abs_time));
    }

    /// The timer is set to time up in the interval for the `relative` time.
    #[inline]
    pub 
    fn interval(rel_time: RelativeTime) -> Self
    {
        return Self::Relative(TimerExpMode::<RelativeTime>::new_interval(rel_time));
    }

    /// The timer is set to time up in the interval for the `relative` time with the initial
    /// delay.
    /// 
    /// # Arguments
    /// 
    /// * `start_del_time` - a [RelativeTime] of the initial delay.
    /// 
    /// * `rel_int_time` - a [RelativeTime] of the interval.
    #[inline]
    pub 
    fn interval_with_start_delay(start_del_time: RelativeTime, rel_int_time: RelativeTime) -> Self
    {
        return Self::Relative(TimerExpMode::<RelativeTime>::new_interval_with_init_delay(start_del_time, rel_int_time));
    }

    /// Sets the `timer_fd` instance to the specific value stored in the current instance.
    fn set_timer(&self, timer_fd: &TimerFd) -> TimerResult<()>
    {
        match *self
        {
            Self::Absolute(timer_exp_mode) => 
                return 
                    timer_fd
                        .get_timer()
                        .set_time(timer_exp_mode)
                        .map_err(|e|
                            map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "cannot set time '{}' for timer: '{}'", timer_exp_mode, timer_fd )
                        ),
            Self::Relative(timer_exp_mode) => 
                return 
                    timer_fd
                        .get_timer()
                        .set_time(timer_exp_mode)
                        .map_err(|e|
                            map_timer_err!(TimerErrorType::TimerError(e.get_errno()), "cannot set time '{}' for timer: '{}'", timer_exp_mode, timer_fd )
                        ),
        }
    }
}


/// A instane which contains the task's timer and a task.
pub(crate) struct PeriodicTaskGuardInner
{
    task_name: String,

    /// A task itself. At the moment it is mutexed in order to provide
    /// the mutability, because the current instance is wrapped into [Arc].
    /// But, the task is never executed from different threads and mutex
    /// should be replaced with something that is [Send] and can provide
    /// mutable reference.
    task: Mutex<PeriodicTaskHndl>,
}

impl fmt::Debug for PeriodicTaskGuardInner
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        f.debug_struct("PeriodicTaskGuardInner").field("task_name", &self.task_name).finish()
    }
}

impl fmt::Display for PeriodicTaskGuardInner
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        write!(f, "task name: '{}'", self.task_name)
    }
}

impl PeriodicTaskGuardInner
{
    fn new(task_name: String, task_inst: PeriodicTaskHndl) -> TimerResult<Self>
    {
        return Ok(Self {  task_name: task_name, task: Mutex::new(task_inst) });
    }


}

/// Internal structure for every worker thread.
struct ThreadWorker
{
    /// A thread name, not task name.
    thread_name: String,

    /// An [Injector] queue for the global tasks received from other threads.
    global_task_injector: Arc<Injector<GlobalTasks>>,

    /// An [Injector] queue for local (specific to current thread) tasks.
    local_thread_inj: Arc<Injector<ThreadTask>>,

    /// A flag which is used to control the thread's loop. Is set to `false`
    /// when thread should exit.
    thread_run_flag: Arc<AtomicBool>,

    /// A shared instance of the [SyncPeriodicTasksInner] which holds all information
    /// about tasks and FD polling. The thread which aquired the mutex first will
    /// process the `global_task_injector` queue and poll the timers for the events which
    /// will be shared over with other threads.
    spti: Arc<Mutex<SharedPeriodicTasks>>,

    /// A last thread to which the task was assigned.
    thread_last_id: usize,

    /// A poll interrupt signal sender.
    poll_int: PollInterrupt,

    /// A error journal, if [Option::None] then no journal available
    tx_err: Option<Sender<TimerError>>,
}

impl ThreadWorker
{
    fn new(
        thread_name: String, 
        global_task_injector: Arc<Injector<GlobalTasks>>, 
        spti: Arc<Mutex<SharedPeriodicTasks>>, 
        poll_int: PollInterrupt, 
        tx_err: Option<Sender<TimerError>>
    ) -> TimerResult<ThreadHandler>
    {
        let local_thread_inj = Arc::new(Injector::<ThreadTask>::new());
        let thread_run_flag = Arc::new(AtomicBool::new(true));
        let thread_run_flag_weak = Arc::downgrade(&thread_run_flag);

        let worker = 
            ThreadWorker
            {
                thread_name: 
                    thread_name.clone(),
                global_task_injector: 
                    global_task_injector,
                local_thread_inj:
                    local_thread_inj.clone(),
                thread_run_flag: 
                    thread_run_flag,
                spti:
                    spti,
                thread_last_id:
                    0,
                poll_int:
                    poll_int,
                tx_err: 
                    tx_err
            };

        let thread_hndl =
            std::thread::Builder::new()
                .name(thread_name)
                .spawn(|| worker.worker())
                .map_err(|e|
                    map_timer_err!(TimerErrorType::SpawnError(e.kind()), "{}", e)
                )?;
            

        return Ok( ThreadHandler::new(thread_hndl, local_thread_inj, thread_run_flag_weak) );
    }


    fn worker(mut self) -> TimerResult<()>
    {
        // force to park to allow the main thread to initialize everything
        std::thread::park();

        while self.thread_run_flag.load(Ordering::Acquire) == true
        {
            // check local queue for the task or token
            while let Steal::Success(task) = self.local_thread_inj.steal()
            {
                match task
                {
                    ThreadTask::TaskExec(task_exec) =>
                    {
                        let Some(ptgi) = task_exec.ptgi.upgrade()
                            else
                            {   
                                // task was removed while was in queue.
                                continue;
                            };

                        // call task
                        match ptgi.task.lock().unwrap().exec()
                        {
                            PeriodicTaskResult::Ok => 
                                {},
                            PeriodicTaskResult::CancelTask => 
                            {
                                self
                                    .global_task_injector
                                    .push(GlobalTasks::SuspendTask(task_exec.ptgi.clone(), None));

                                let _ = self.poll_int.aquire().map(|v| v.interrupt_drop());
                            },
                            PeriodicTaskResult::TaskReSchedule(ptt) => 
                            {
                                self
                                    .global_task_injector
                                    .push(GlobalTasks::ReschedTask(task_exec.ptgi.clone(), ptt, None));

                                let _ = self.poll_int.aquire().map(|v| v.interrupt_drop());
                            }
                        }

                        drop(task_exec);
                    }
                }
            }

            let spti_lock_res =  self.spti.try_lock();

            if let Ok(mut task_token) = spti_lock_res
            {
                let thread_hndl_cnt = task_token.thread_pool.get().unwrap().len();

                // check global task queue
                while let Steal::Success(task) = self.global_task_injector.steal()
                {
                    match task
                    {
                        GlobalTasks::AddTask(task_name, ptt, ptg, opt_err_ret) => 
                        {
                            if task_token.contains_task(&task_name) == true
                            {
                                if let Some(err_ret) = opt_err_ret
                                {
                                    let err_msg = 
                                        map_timer_err!(TimerErrorType::Duplicate, 
                                            "thread: '{}', task: '{}' already exists", self.thread_name, task_name);

                                    let _ = err_ret.send(Err(err_msg));
                                }

                                continue;
                            }

                            let period_task_ticket = 
                                PeriodicTaskTicket::new(task_name.clone(), ptt, ptg, &task_token.timers_poll);

                            if let Err(e) = period_task_ticket 
                            {
                                if let Some(err_ret) = opt_err_ret
                                {
                                    let _ = err_ret.send(Err(e));
                                }

                                continue;
                            }

                            let period_task_ticket = period_task_ticket.unwrap();
                            
                            task_token.insert_task(period_task_ticket);
                            
                            if let Some(err_ret) = opt_err_ret
                            {
                                let _ = err_ret.send(Ok(()));
                            }
                        },
                        GlobalTasks::RemoveTask(ptg_arc, opt_err_ret) => 
                        {
                            let res = task_token.remove_task(&ptg_arc.task_name);

                            if let Err(e) = res
                            {
                                if let Some(err_ret) = opt_err_ret
                                {
                                    let err_msg = 
                                        map_timer_err!(e.get_error_type(), 
                                            "thread: '{}', {}", self.thread_name, e.get_error_msg());

                                    let _ = err_ret.send(Err(err_msg));
                                }

                                continue;
                            }

                            let ptt_ref = res.unwrap();

                            //let ptt = ptt_ref.into_inner();

                            // stop timer
                            if let Err(e) = ptt_ref.unset_timer()
                            {
                                if let Some(err_ret) = opt_err_ret.as_ref()
                                {
                                    let _ = err_ret.send(Err(e));
                                }

                                continue;
                            }

                            drop(ptt_ref);

                            /*if let Some(err_ret) = opt_err_ret
                            {
                                let _ = err_ret.send(Ok(()));
                            }

                            if let Some(ptt_ref) = task_token.tasks.remove(&ptg_arc.task_name)
                            {
                                
                            }
                            else
                            {
                                if let Some(err_ret) = opt_err_ret
                                {
                                    let err_msg = 
                                        map_timer_err!(TimerErrorType::NotFound, 
                                            "thread: '{}', task: '{}' was not found", self.thread_name, ptg_arc.task_name);

                                    let _ = err_ret.send(Err(err_msg));
                                }
                            }*/

                            drop(ptg_arc);
                        },
                        GlobalTasks::ReschedTask( ptg_weak, ptt, opt_err_ret ) =>
                        {
                            let Some(ptg_arc) = ptg_weak.upgrade()
                                else
                                {   
                                    // task was removed while was in queue.
                                    continue;
                                };

                            let res_task = task_token.get_task_by_name(&ptg_arc.task_name);

                            let res = 
                                match res_task
                                {
                                    Ok(task) => 
                                    {
                                        // stop timer
                                        let _ = task.unset_timer();

                                        // update timer value and set timer
                                        let res = task.update_task_time(ptt);

                                        if let Err(e) = res
                                        {
                                            if let Some(err_ret) = opt_err_ret.as_ref()
                                            {
                                                let _ = err_ret.send(Err(e));
                                            }

                                            continue;
                                        }

                                        Ok(())
                                    },
                                    Err(err) => 
                                    {
                                        Err(err)
                                    }
                                };

                            if let Some(err_ret) = opt_err_ret
                            {
                                let _ = err_ret.send(res);
                            }
                        },
                        GlobalTasks::SuspendTask(ptg_weak, opt_err_ret) =>
                        {
                            let Some(ptg_arc) = ptg_weak.upgrade()
                                else
                                {   
                                    // task was removed while was in queue.
                                    continue;
                                };

                            /*let res_task = 
                                task_token
                                    .tasks
                                    .get(&ptg_arc.task_name)
                                    .ok_or_else(||
                                        map_timer_err!(TimerErrorType::NotFound, 
                                            "thread: '{}', task: '{}' was not found", self.thread_name, ptg_arc.task_name)
                                    );*/

                            let res_task = task_token.get_task_by_name(&ptg_arc.task_name);

                            if let Err(e) = res_task
                            {
                                if let Some(err_ret) = opt_err_ret.as_ref()
                                {
                                    let _ = err_ret.send(Err(e));
                                }

                                continue;
                            }

                            let task = res_task.unwrap();

                            let res = task.unset_timer();

                            if let Some(err_ret) = opt_err_ret
                            {
                                let _ = err_ret.send(res);
                            }
                        },
                        GlobalTasks::ResumeTask(ptg_weak, opt_err_ret) =>
                        {
                            let Some(ptg_arc) = ptg_weak.upgrade()
                                else
                                {   
                                    // task was removed while was in queue.
                                    continue;
                                };

                            /*let res_task = 
                                task_token
                                    .tasks
                                    .get(&ptg_arc.task_name)
                                    .ok_or_else(||
                                        map_timer_err!(TimerErrorType::NotFound, 
                                            "thread: '{}', timer with FD: '{}' was not found", self.thread_name, ptg_arc.task_name)
                                    );*/

                            let res_task = task_token.get_task_by_name(&ptg_arc.task_name);

                            if let Err(e) = res_task
                            {
                                if let Some(err_ret) = opt_err_ret.as_ref()
                                {
                                    let _ = err_ret.send(Err(e));
                                }

                                continue;
                            }

                            let res = res_task.unwrap().set_timer();

                            if let Some(err_ret) = opt_err_ret.as_ref()
                            {
                                let _ = err_ret.send(res);
                            }
                        }
                    }
                }

                // poll
                let Some(res) = 
                    task_token
                        .timers_poll
                        .poll(Some(5000))?
                else
                    {
                        continue;
                    };

                for event in res
                {
                    match event
                    {
                        PollEventType::TimerRes(timer_fd, timer_res) => 
                        {
                            let task_by_id = 
                                task_token
                                    .get_task_by_timer_id(timer_fd)
                                    .map_err(|e|
                                        map_timer_err!(e.get_error_type(), "thread: '{}', {}", self.thread_name, e.get_error_msg())
                                    )
                                    .map(|task|
                                        (task.ptg.clone(), task.ptt.clone(), task.weak_ticket.upgrade(), task.task_name.clone())
                                    );

                            let Ok((ptg, ptt, ticket_arc_opt, task_name)) = task_by_id
                                else
                                {
                                    if let Some(tx_err) = self.tx_err.as_ref()
                                    {
                                        let _ = tx_err.send(task_by_id.err().unwrap());
                                    }

                                    continue;
                                };
                                

                            // check that PeriodicTaskGuard is ok
                            let Some(ptg_arc) = ptg.upgrade()
                                else
                                {
                                    //let task_name = task.task_name.clone();

                                    // if Arc cannot be upgraded, then remove the task as removed
                                    //task_token.tasks.remove(&timer_fd);
                                    if let Err(e) = task_token.remove_task(&task_name)
                                    {
                                        if let Some(tx_err) = self.tx_err.as_ref()
                                        {
                                            let _ = tx_err.send(e);
                                        }
                                    }

                                    // continue event handling
                                    continue;
                                };

                            // check timer result
                            let overflow_cnt: u64 = 
                                match timer_res
                                {
                                    TimerReadRes::Ok(overfl) => 
                                    {
                                        overfl
                                    },
                                    TimerReadRes::Cancelled => 
                                    {
                                        // the system time was modified, resched instance

                                        self
                                            .global_task_injector
                                            .push(
                                                GlobalTasks::ReschedTask(ptg.clone(), ptt.clone(), None)
                                            );

                                        // timer was reset continue
                                        continue;
                                    },
                                    TimerReadRes::WouldBlock => 
                                    {
                                        // should not happen, silently ignore or panic?
                                        panic!("assertion trap: timer retuned WouldBlock, {}", ptg_arc);
                                    }
                                };

                            // check if ticket presents, if not create
                           // let ticket_arc_opt = task.weak_ticket.upgrade();


                            let ticket = 
                                match ticket_arc_opt
                                {
                                    Some(ticket) => 
                                        ticket,
                                    None => 
                                    {    
                                        let task_thread = 
                                            {
                                                /*let thread_local_hnd = 
                                                    task_token
                                                        .thread_pool
                                                        .get()
                                                        .unwrap()[self.thread_last_id]
                                                        .get_thread_handler();*/

                                               

                                                self.thread_last_id = (self.thread_last_id + 1) % thread_hndl_cnt;

                                                //thread_local_hnd
                                                task_token.clone_thread_handler(self.thread_last_id)
                                            }; 

                                        let ticket =
                                                Arc::new(NewTaskTicket::new(task_thread, ptg.clone()));

                                        let task = 
                                            task_token
                                                .get_task_by_timer_id(timer_fd)
                                                .map_err(|e|
                                                    map_timer_err!(e.get_error_type(), "thread: '{}', {}", self.thread_name, e.get_error_msg())
                                                )?;
                                                
                                        task.weak_ticket = Arc::downgrade(&ticket);

                                        ticket
                                    }
                                };
                            
                            NewTaskTicket::send_task(ticket, overflow_cnt, thread_hndl_cnt);
                        },
                        _ => 
                        {
                            // ignore
                        },
                    }
                } // for
            }
            else if let Err(TryLockError::WouldBlock) = spti_lock_res
            {
                if self.thread_run_flag.load(Ordering::Acquire) == false
                {
                    return Ok(());
                }

                if self.local_thread_inj.is_empty() == true
                {
                    std::thread::park_timeout(Duration::from_secs(2));
                }
            } 
            
        } // while

        return Ok(());
    }
}


/// A common instance which contains the specific thread handler and
/// a [Weak] reference to the loop controlling flag.
#[derive(Debug)]
struct ThreadHandler
{
    /// A thread join handler
    hndl: JoinHandle<TimerResult<()>>,

    /// A local task injector.
    task_injector: Arc<Injector<ThreadTask>>,

    /// A flag which controls the thread loop. Initialized as `true`.
    thread_flag: Weak<AtomicBool>,
}

impl ThreadHandler
{
    fn new(hndl: JoinHandle<TimerResult<()>>, task_injector: Arc<Injector<ThreadTask>>, thread_flag: Weak<AtomicBool>) -> Self
    {            
        return 
            Self
            {
                hndl,
                task_injector,
                thread_flag: thread_flag
            };
    }

    fn stop(&self)
    {
        if let Some(v) = self.thread_flag.upgrade()
        {
            v.store(false, Ordering::Release);
        }
    }

    fn unpark(&self) 
    {
        self.hndl.thread().unpark();
    }

    fn send_task(&self, task: Arc<NewTaskTicket>, unpark: bool)
    {
        self.task_injector.push(ThreadTask::TaskExec(task));

        if unpark == true
        {
            self.hndl.thread().unpark();
        }

        return;
    }

    fn clean_local_queue(&self)
    {
        while let Steal::Success(_) = self.task_injector.steal() {}

        return;
    }
}

/// A instance which contains all information about the threads allocated for the
/// task execution, tasks and timer polling. An instance is shared with all threads.
#[derive(Debug)]
pub struct SharedPeriodicTasks
{
    /// A list of the threads. A [OnceCell] is used to initialize the field once.
    thread_pool: OnceCell<Arc<Vec<Arc<ThreadHandler>>>>,

    /// A [HashMap] of the regestered tasks. Each task is mapped to its timer FD.
    tasks_by_timer_fd: HashMap<TimerId, PeriodicTaskTicket>,

    /// A task name to [TimerId] map.
    tasks_name_to_timer_id: HashMap<String, TimerId>,

    /// A timer event poll. 
    timers_poll: TimerPoll,
}


impl SharedPeriodicTasks
{ 
    fn new() -> TimerResult<Self>
    {

        return Ok( 
            Self 
            { 
                thread_pool: 
                    OnceCell::default(),
                tasks_by_timer_fd: 
                    HashMap::new(),
                tasks_name_to_timer_id: 
                    HashMap::new(),
                timers_poll: 
                    TimerPoll::new()?
            }
        );
    }

    /// Returns the mutable reference to [PeriodicTaskTicket] by the [TimerId].
    fn get_task_by_timer_id(&mut self, timer_fd: TimerId) -> TimerResult<&mut PeriodicTaskTicket>
    {
        return 
            self
                .tasks_by_timer_fd
                .get_mut(&timer_fd)
                .ok_or_else(||
                    map_timer_err!(TimerErrorType::NotFound, "task fd: '{}' was found but task was not found", 
                        timer_fd)
                );
    }

    /// Returns the mutable reference to [PeriodicTaskTicket] by the task name.
    /// 
    /// This function is slower than [Self::get_task_by_timer_id] because it performs
    /// look in two tables.
    fn get_task_by_name(&mut self, task_name: &str) -> TimerResult<&mut PeriodicTaskTicket>
    {
        let res = 
            self
                .tasks_name_to_timer_id
                .get(task_name)
                .ok_or_else(||
                    map_timer_err!(TimerErrorType::NotFound, "task: '{}' was not found", task_name)
                )
                .map(|v|
                    self
                        .tasks_by_timer_fd
                        .get_mut(v)
                        .ok_or_else(||
                            map_timer_err!(TimerErrorType::NotFound, "task: '{}' fd: '{}' was found but task was not found", 
                                task_name, v)
                        )
                )??;

        return Ok(res);
    }

    /// Clones the [ThreadHandler] by the `thread_last_id`.
    /// 
    /// # Panics
    /// 
    /// Will panic if `thread_last_id` is out of range.
    fn clone_thread_handler(&self, thread_last_id: usize) -> Arc<ThreadHandler>
    {
        let thread_local_hnd = 
            self
                .thread_pool
                .get()
                .unwrap()[thread_last_id]
                .clone();

        return thread_local_hnd;
    }

    /// Removes the task by `task_name`.
    fn remove_task(&mut self, task_name: &str) -> TimerResult<PeriodicTaskTicket>
    {
        let Some(task_timer_fd) = self.tasks_name_to_timer_id.remove(task_name)
            else
            {
                timer_err!(TimerErrorType::NotFound, "task: '{}' was not found", task_name);
            };

        let Some(ptt) = self.tasks_by_timer_fd.remove(&task_timer_fd)
            else
            {
                timer_err!(TimerErrorType::NotFound, "task: '{}' fd: '{}' was found but task was not found", 
                    task_name, task_timer_fd);
            };

        return Ok(ptt);
    }

    /// Returns `true` if task with name `task_name` exists.
    fn contains_task(&self, task_name: &str) -> bool
    {
        return self.tasks_name_to_timer_id.contains_key(task_name);
    }

    /// Inserts the `task` into scheduler list.
    fn insert_task(&mut self, period_task_ticket: PeriodicTaskTicket)
    {
        let timer_id = period_task_ticket.sync_timer.get_inner().as_timer_id();

        self.tasks_name_to_timer_id.insert(period_task_ticket.task_name.clone(), timer_id);
        self.tasks_by_timer_fd.insert(timer_id, period_task_ticket);

        return;
    }
}

/// An `inner` of the [SyncPeriodicTasks].
#[derive(Debug)]
pub struct SyncPeriodicTasksInner
{
    /// A [Weak] reference to the [EventFd] of the timer `poll` which is used to interrupt the polling.
    poll_int: PollInterrupt,

    /// A `global` FIFO which is used to send commands to the task executor.
    task_injector: Arc<Injector<GlobalTasks>>,

    /// An error journal
    error_journal: Option<Mutex<Receiver<TimerError>>>,
}

impl SyncPeriodicTasksInner
{
    fn send_global_cmd(&self, glob: GlobalTasks) -> TimerResult<()>
    {
        let poll_int = 
            self.poll_int.aquire()?;
                
        self.task_injector.push(glob);

        poll_int.interrupt_drop()?;

        return Ok(());
    }

    fn clear_global_queue(&self)
    {
        while let Steal::Success(_) = self.task_injector.steal() {}

        return;
    }
}

/// A wrapper for the task which is described in closure.
struct SyncCallableOper
{
    op: Box<dyn FnMut() -> PeriodicTaskResult>,
}

unsafe impl Send for SyncCallableOper {}

impl PeriodicTask for SyncCallableOper
{
    fn exec(&mut self) -> PeriodicTaskResult 
    {
        return (self.op)();
    }
}

/// A `main` instance which spawns the task executor.
/// 
/// ```ignore
/// let spt = SyncPeriodicTasks::new(1.try_into().unwrap()).unwrap();
/// let task1 = TaskStruct1::new(2, send);
/// let task1_ptt = PeriodicTaskTime::Relative(TimerExpMode::<RelativeTime>::new_interval(RelativeTime::new_time(1, 0)));
/// let task1_guard = s.add("task1", task1, task1_ptt).unwrap();
/// ```
/// 
/// This instance should be kept somewhere and dropped only if the task executor WITH
/// all spawned tasks are no longer needed.
#[derive(Debug, Clone)]
pub struct SyncPeriodicTasks
{
    threads: Option<Arc<Vec<Arc<ThreadHandler>>>>,

    inner: Arc<SyncPeriodicTasksInner>,
}

impl Drop for SyncPeriodicTasks
{
    fn drop(&mut self) 
    {
        self.inner.clear_global_queue();

        let mut threads = self.threads.take().unwrap();

        // stop all threads
        for thread in threads.iter()
        {
            thread.stop();
            thread.unpark();
        }

        // interrupt poll
        let _ = self.inner.poll_int.aquire().map(|v| v.interrupt_drop());

        for _ in 0..5
        {
            let threads_unwr = 
                match Arc::try_unwrap(threads)
                {
                    Ok(r) => r,
                    Err(e) =>
                    {
                        threads = e;

                        std::thread::sleep(Duration::from_millis(500));

                        continue;

                    }
                };

            for thread in threads_unwr
            {
                thread.clean_local_queue();

                let Some(thread) = Arc::into_inner(thread)
                else
                {
                    panic!("assertion trap: ~SyncPeriodicTasks, a reference to ThreadHandler left somewhere");
                };

                let _ = thread.hndl.join();
            }

            break;
        }
    }
}


impl SyncPeriodicTasks
{

    
    /// Creates new instance. An amount of threads allocated for the task executor
    /// should be specified. All threads will be started immidiatly. For small
    /// tasks one thread will be enough. For a large amount of tasks, especially it
    /// tasks are waken up oftenly then at least two threads should be allocated.
    /// 
    /// # Arguments
    /// 
    /// * `threads_cnt` - a [NonZeroUsize] amount of threads.
    /// 
    /// # Returns
    /// 
    /// The [Result] is returned as alias [TimerResult].
    pub 
    fn new(threads_cnt: NonZeroUsize, error_report: bool) -> TimerResult<Self>
    {
        let spti = SharedPeriodicTasks::new()?;
        let poll_int = spti.timers_poll.get_poll_interruptor();

        // wrap into the mutex because this instance is shared 
        let spti = Arc::new(Mutex::new(spti));

        let task_injector = Arc::new(Injector::<GlobalTasks>::new());

        let mut thread_hndls: Vec<Arc<ThreadHandler>> = Vec::with_capacity(threads_cnt.get());

        // error report
        let err_journal =
            if error_report == true
            {
                Some(mpsc::channel::<TimerError>())
            }
            else
            {
                None
            };

        // spawn threads, all spawn threads will be parked by default.
        for i in 0..threads_cnt.get()
        {
            let handler = 
                ThreadWorker::new(
                    format!("timer_exec/{}s", i), 
                    task_injector.clone(), 
                    spti.clone(), 
                    poll_int.clone(), 
                    err_journal.as_ref().map(|c| c.0.clone())
                )?;

            thread_hndls.push(Arc::new(handler));
        }

        let thread_hndls = Arc::new(thread_hndls);

        // Lock the instance to initialze the `thread_pool` variable.
        let spti_lock = spti.lock().unwrap();
        spti_lock.thread_pool.get_or_init(|| thread_hndls.clone());

        // get a random thread to unpark it and start the process
        let thread = 
            spti_lock
                .thread_pool
                .get()
                .unwrap()
                .get(random_range(0..threads_cnt.get()))
                .unwrap()
                .clone();

        drop(spti_lock);

        // unpark thread
        thread.unpark();

        let inner = 
            SyncPeriodicTasksInner
            {
                poll_int: 
                    poll_int,
                task_injector: 
                    task_injector,
                error_journal: 
                    err_journal.map(|j| Mutex::new(j.1))
            };

        return Ok( 
            Self 
            { 
                threads: Some(thread_hndls),
                inner: Arc::new(inner),
            }
        );
    }

    /// Adds and spawns the task. The task is defined with generic `T` which is any 
    /// datatype which implements [PeriodicTask].
    /// 
    /// # Arguments
    /// 
    /// * `task_name` - a task name. used only for identification purposes in debug messages.
    /// 
    /// * `task` - a task which should be executed. It should implenet [PeriodicTask].
    /// 
    /// * `task_time` - [PeriodicTaskTime] a time when the task must be spawned.
    /// 
    /// # Returns
    /// 
    /// The [Result] is returned as alias [TimerResult].
    /// 
    /// # Example
    /// 
    /// ```ignore
    /// #[derive(Debug)]
    /// struct TaskStruct1
    /// {
    ///     a1: u64,
    ///     s: Sender<u64>,
    /// }
    /// 
    /// impl TaskStruct1
    /// {
    ///     fn new(a1: u64, s: Sender<u64>) -> Self
    ///     {
    ///         return Self{ a1: a1, s };
    ///     }
    /// }
    /// 
    /// impl PeriodicTask for TaskStruct1
    /// {
    ///     fn exec(&mut self) -> PeriodicTaskResult
    ///     {
    ///         println!("taskstruct1 val: {}", self.a1);
    ///         let _ = self.s.send(self.a1);
    ///         return PeriodicTaskResult::Ok;
    ///     }
    /// }
    /// 
    /// fn main()
    /// {
    ///     let s = SyncPeriodicTasks::new(1.try_into().unwrap()).unwrap();
    ///     
    ///     let (send, recv) = mpsc::channel::<u64>();
    /// 
    ///     let task1 = TaskStruct1::new(2, send);
    /// 
    ///     let task1_ptt = PeriodicTaskTime::exact_time(AbsoluteTime::now() + RelativeTime::new_time(3, 0));
    /// 
    ///     let task1_guard = s.add("task1", task1, task1_ptt).unwrap();
    /// 
    ///     println!("added");
    /// 
    ///     let val = recv.recv();
    /// 
    ///     println!("{:?}", val);
    /// 
    ///     drop(task1_guard);
    /// }
    /// ```
    pub 
    fn add<T>(&self, task_name: impl Into<String>, task: T, task_time: PeriodicTaskTime) -> TimerResult<PeriodicTaskGuard>
    where T: PeriodicTask
    {
        let task_int: PeriodicTaskHndl = Box::new(task);

        let task_name_str: String = task_name.into();

        let period_task_guard = 
            Arc::new(PeriodicTaskGuardInner::new(task_name_str.clone(), task_int)?);


       /* let period_task_ticket = 
            PeriodicTaskTicket::new(task_name_str.clone(), task_time, Arc::downgrade(&period_task_guard));*/

        let (mpsc_send, mpsc_recv) = mpsc::channel();

        self.inner.send_global_cmd(GlobalTasks::AddTask(task_name_str.clone(), task_time, period_task_guard.clone(), Some(mpsc_send)) )?;

        let _ = 
            mpsc_recv
                .recv()
                .map_err(|e|
                    map_timer_err!(TimerErrorType::ExternalError, "mpsc error: {}", e)
                )??;

        let ret = 
            PeriodicTaskGuard
            {
                task_name: task_name_str,
                guard: Some(period_task_guard),
                spt: self.inner.clone()
            };

        return Ok(ret);
    }

    /// Adds and spawns the task in form of a closure.
    /// 
    /// # Arguments
    /// 
    /// * `task_name` - a task name. used only for identification purposes in debug messages.
    /// 
    /// * `task_time` - [PeriodicTaskTime] a time when the task must be spawned.
    /// 
    /// * `clo` - a function to exec on timeout. The function must return [PeriodicTaskResult].
    /// 
    /// # Returns
    /// 
    /// The [Result] is returned as alias [TimerResult].
    /// 
    /// # Example
    /// 
    /// ```ignore
    /// 
    /// let task1_ptt = PeriodicTaskTime::exact_time(AbsoluteTime::now() + RelativeTime::new_time(3, 0));
    /// 
    /// let (send, recv) = mpsc::channel::<u64>();
    /// 
    /// let task1_guard = 
    ///     s.add_closure("task2", task1_ptt, 
    ///         move || 
    ///         { 
    ///             println!("test output"); 
    ///             send.send(1).unwrap();
    /// 
    ///             return PeriodicTaskResult::Ok;
    ///         }
    ///     ).unwrap();
    /// 
    /// let val = recv.recv();
    /// 
    /// println!("{:?}", val);
    /// ```
    pub 
    fn add_closure<F>(&self, task_name: impl Into<String>, task_time: PeriodicTaskTime, clo: F) -> TimerResult<PeriodicTaskGuard>
    where F: 'static + FnMut() -> PeriodicTaskResult + Send
    {
        let closure_task = SyncCallableOper{ op: Box::new(clo) };

        return self.add(task_name, closure_task, task_time);
    }
    /// Checks if any thread have crashed and no longer works.
    /// 
    /// # Returns
    /// 
    /// A [Option] is retuerned with the inner data:
    /// 
    /// * [Option::Some] with the thread name [String] that have quit.
    /// 
    /// * [Option::None] indicating that everthing is fine.
    pub 
    fn check_thread_status(&self) -> Option<String>
    {
        for thread in self.threads.as_ref().unwrap().iter()
        {
            if let None = thread.thread_flag.upgrade()
            {
                return Some(thread.hndl.thread().name().unwrap().to_string());
            }
        }

        return None;
    }

    /// Reads the error from the journal. This function does not block current thread
    /// when receiving, but it will wait for the [Mutex] release.
    /// 
    /// # Returns
    /// 
    /// * If any message is available the [Option::Some] with [TimerError] is returned.
    /// 
    /// * If instance is not available or no errors, the [Option::None] is returned.
    pub 
    fn read_error(&self) -> Option<TimerError>
    {
        let Some(rx) = self.inner.error_journal.as_ref()
            else { return None };

        let Ok(err) = 
            rx
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .recv_timeout(Duration::from_secs(0))
        else { return None };

        return Some(err);
    }
}

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

    use crate::{periodic_task::sync_tasks::{PeriodicTask, PeriodicTaskResult, PeriodicTaskTime, SyncPeriodicTasks}, AbsoluteTime, RelativeTime};

    #[derive(Debug)]
    struct TaskStruct1
    {
        a1: u64,
        s: Sender<u64>,
    }

    impl TaskStruct1
    {
        fn new(a1: u64, s: Sender<u64>) -> Self
        {
            return Self{ a1: a1, s };
        }
    }

    impl PeriodicTask for TaskStruct1
    {
        fn exec(&mut self) -> PeriodicTaskResult
        {
            println!("taskstruct1 val: {}", self.a1);

            let _ = self.s.send(self.a1);

            return PeriodicTaskResult::Ok;
        }
    }

    #[derive(Debug)]
    struct TaskStruct2
    {
        a1: u64,
        s: Sender<u64>,
    }

    impl TaskStruct2
    {
        fn new(a1: u64, s: Sender<u64>) -> Self
        {
            return Self{ a1: a1, s };
        }
    }

    impl PeriodicTask for TaskStruct2
    {
        fn exec(&mut self) -> PeriodicTaskResult
        {
            println!("taskstruct2 val: {}", self.a1);

            self.s.send(self.a1).unwrap();

            return PeriodicTaskResult::TaskReSchedule(PeriodicTaskTime::exact_time(AbsoluteTime::now() + RelativeTime::new_time(2, 0)));
        }
    }

    impl<F> PeriodicTask for F
    where F: 'static + FnMut() + Send + fmt::Debug
    {
        fn exec(&mut self) -> PeriodicTaskResult 
        {
            (self)();
            return PeriodicTaskResult::Ok;
        }
    }

    #[test]
    fn ttt()
    {
        let s = 
            SyncPeriodicTasks::new(1.try_into().unwrap(), true).unwrap();

        let task1_ptt = 
            PeriodicTaskTime
                ::exact_time(AbsoluteTime::now() + RelativeTime::new_time(3, 0));

        let (send, recv) = mpsc::channel::<u64>();
 
        let task1_guard = 
            s.add_closure("task2", task1_ptt, 
                move || 
                { 
                    println!("test output"); 
                    send.send(2).unwrap();

                    return PeriodicTaskResult::Ok;
                }
            ).unwrap();


        println!("added");

        let val = recv.recv_timeout(Duration::from_millis(4000));

        if val.is_err() == true
        {
            let e =  s.read_error();
            println!("ERROR, {:?}",e);

            assert_eq!(true, false, "{:?}", e);
        }

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

        assert_eq!(Ok(2), val);

        drop(task1_guard);
    }

    #[test]
    fn test1_absolute_simple()
    {
        let s = SyncPeriodicTasks::new(1.try_into().unwrap(), false).unwrap();

        let (send, recv) = mpsc::channel::<u64>();

        let task1 = TaskStruct1::new(2, send);
        let task1_ptt = PeriodicTaskTime::exact_time(AbsoluteTime::now() + RelativeTime::new_time(3, 0));
        let task1_guard = s.add("task1", task1, task1_ptt).unwrap();

        println!("added");
        let val = recv.recv();

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

        drop(task1_guard);
    }

    #[test]
    fn test1_relative_simple()
    {
        let s = SyncPeriodicTasks::new(1.try_into().unwrap(), false).unwrap();

        let (send, recv) = mpsc::channel::<u64>();

        let task1 = TaskStruct1::new(2, send);
        let task1_ptt = PeriodicTaskTime::interval(RelativeTime::new_time(1, 0));
        let task1_guard = s.add("task1", task1, task1_ptt).unwrap();
        
        let mut s = Instant::now();

        for i in 0..3
        {
            let val = recv.recv().unwrap();

            let e = s.elapsed();
            s = Instant::now();

            println!("{}: {:?} {:?} {}", i, val, e, e.as_micros());
            
            assert!(999000 < e.as_micros() && e.as_micros() < 10001200);
            assert_eq!(val, 2);
        }

        drop(task1_guard);

        std::thread::sleep(Duration::from_millis(100));

        return;
    }
    
    #[test]
    fn test1_relative_resched_to_abs()
    {
        let s = SyncPeriodicTasks::new(1.try_into().unwrap(), false).unwrap();

        let (send, recv) = mpsc::channel::<u64>();

        let task1 = TaskStruct2::new(2, send);
        let task1_ptt = PeriodicTaskTime::interval(RelativeTime::new_time(1, 0));
        let task1_guard = s.add("task1", task1, task1_ptt).unwrap();

        let s = Instant::now();
        match recv.recv_timeout(Duration::from_millis(1150))
        {
            Ok(rcv_a) => 
            {
                let e = s.elapsed();
                println!("{:?} {}", e, e.as_micros());
                assert_eq!(rcv_a, 2);
                assert!(990051 < e.as_micros() && e.as_micros() < 1020551);
            },
            Err(RecvTimeoutError::Timeout) => 
                panic!("tineout"),
            Err(e) =>
                panic!("{}", e),
        }

        let s = Instant::now();
        match recv.recv_timeout(Duration::from_millis(2100))
        {
            Ok(rcv_a) => 
            {
                let e = s.elapsed();
                println!("{:?} {}", e, e.as_micros());
                assert_eq!(rcv_a, 2);
                assert!(1999642 < e.as_micros() && e.as_micros() < 2008342);
            },
            Err(RecvTimeoutError::Timeout) => 
                panic!("tineout"),
            Err(e) =>
                panic!("{}", e),
        }

        let s = Instant::now();
        match recv.recv_timeout(Duration::from_millis(2100))
        {
            Ok(rcv_a) => 
            {
                let e = s.elapsed();
                println!("{:?} {}", e, e.as_micros());
                assert_eq!(rcv_a, 2);
                assert!(1999642 < e.as_micros() && e.as_micros() < 2003342);
            },
            Err(RecvTimeoutError::Timeout) => 
                panic!("tineout"),
            Err(e) =>
                panic!("{}", e),
        }

        drop(task1_guard);

        std::thread::sleep(Duration::from_millis(100));

        return;
    }

    #[test]
    fn test1_relative_simple_resched()
    {
        let s = SyncPeriodicTasks::new(1.try_into().unwrap(), false).unwrap();

        let (send, recv) = mpsc::channel::<u64>();

        let task1 = TaskStruct1::new(2, send);
        let task1_ptt = 
            PeriodicTaskTime::interval(
                RelativeTime::new_time(1, 0)
            );
        let task1_guard = s.add("task1", task1, task1_ptt).unwrap();
        
        let mut s = Instant::now();

        for i in 0..3
        {
            let val = recv.recv().unwrap();

            let e = s.elapsed();
            s = Instant::now();

            println!("{}: {:?} {:?} {}", i, val, e, e.as_micros());
            
            assert!(990000 < e.as_micros() && e.as_micros() < 10001200);
            assert_eq!(val, 2);
        }

        task1_guard
            .reschedule_task(
                PeriodicTaskTime::exact_time(AbsoluteTime::now() + RelativeTime::new_time(2, 0))
            )
            .unwrap();

        s = Instant::now();
        let val = recv.recv().unwrap();
        let e = s.elapsed();

        println!("resched: {:?} {:?} {}", val, e, e.as_micros());

        assert!(1990000 < e.as_micros() && e.as_micros() < 2003560);

        let val = recv.recv_timeout(Duration::from_secs(3));

        assert_eq!(val.is_err(), true);
        assert_eq!(val.err().unwrap(), RecvTimeoutError::Timeout);

        drop(task1_guard);

        std::thread::sleep(Duration::from_millis(100));

        return;
    }

    #[test]
    fn test1_relative_simple_cancel()
    {
        let s = SyncPeriodicTasks::new(1.try_into().unwrap(), false).unwrap();

        let (send, recv) = mpsc::channel::<u64>();

        let task1 = TaskStruct1::new(0, send.clone());
        let task1_ptt = 
            PeriodicTaskTime::interval(RelativeTime::new_time(1, 0));

        let task2 = TaskStruct1::new(1, send.clone());
        let task2_ptt = 
            PeriodicTaskTime::interval(RelativeTime::new_time(2, 0));

        let task3 = TaskStruct1::new(2, send);
        let task3_ptt = 
            PeriodicTaskTime::interval(RelativeTime::new_time(0, 500_000_000));

        let task1_guard = s.add("task1", task1, task1_ptt).unwrap();
        let task2_guard = s.add("task2", task2, task2_ptt).unwrap();
        let task3_guard = s.add("task3", task3, task3_ptt).unwrap();

        let mut a_cnt: [u8; 3] = [0_u8; 3];

        let end = AbsoluteTime::now() + RelativeTime::new_time(5, 100_000_000);

        while AbsoluteTime::now() < end
        {
            match recv.recv_timeout(Duration::from_millis(1))
            {
                Ok(rcv_a) => 
                    a_cnt[rcv_a as usize] += 1,
                Err(RecvTimeoutError::Timeout) => 
                    continue,
                Err(e) =>
                    panic!("{}", e),
            }

            
        }

        assert_eq!(a_cnt[0], 5);
        assert_eq!(a_cnt[1], 2);
        assert_eq!(a_cnt[2], 10);

        // drop task #3
        task3_guard.suspend_task().unwrap();


        let end = AbsoluteTime::now() + RelativeTime::new_time(5, 100_000_000);

        while AbsoluteTime::now() < end
        {
            match recv.recv_timeout(Duration::from_millis(1))
            {
                Ok(rcv_a) => 
                    a_cnt[rcv_a as usize] += 1,
                Err(RecvTimeoutError::Timeout) => 
                    continue,
                Err(e) =>
                    panic!("{}", e),
            }

            
        }

        assert_eq!(a_cnt[0] > 5, true);
        assert_eq!(a_cnt[1] > 2, true);
        assert!((a_cnt[2] == 10 || a_cnt[2] == 11));

        drop(task1_guard);
        drop(task2_guard);
        drop(task3_guard);

        let end = AbsoluteTime::now() + RelativeTime::new_time(5, 100_000_000);

        while AbsoluteTime::now() < end
        {
            match recv.recv_timeout(Duration::from_millis(1))
            {
                Ok(rcv_a) => 
                    a_cnt[rcv_a as usize] += 1,
                Err(RecvTimeoutError::Timeout) => 
                    continue,
                Err(_) =>
                    break,
            }

            
        }

        assert_eq!(AbsoluteTime::now() < end, true);

        

        return;
    }

    // freebsd text failed with  src/periodic_task/sync_tasks.rs:1784:9 left: 6
    #[test]
    fn test2_multithread_1()
    {
        let s = SyncPeriodicTasks::new(2.try_into().unwrap(), false).unwrap();

        let (send, recv) = mpsc::channel::<u64>();

        let task1 = TaskStruct1::new(0, send.clone());
        let task1_ptt = 
            PeriodicTaskTime::interval(RelativeTime::new_time(1, 0));

        let task2 = TaskStruct1::new(1, send.clone());
        let task2_ptt = 
            PeriodicTaskTime::interval(RelativeTime::new_time(2, 0));

        let task3 = TaskStruct1::new(2, send.clone());
        let task3_ptt = 
            PeriodicTaskTime::interval(RelativeTime::new_time(0, 500_000_000));

        let task4 = TaskStruct1::new(3, send.clone());
        let task4_ptt = 
            PeriodicTaskTime::interval(RelativeTime::new_time(0, 200_000_000));

        let task5 = TaskStruct1::new(4, send.clone());
        let task5_ptt = 
            PeriodicTaskTime::exact_time(AbsoluteTime::now() + RelativeTime::new_time(5, 0));

        let task1_guard = s.add("task1", task1, task1_ptt).unwrap();
        let task2_guard = s.add("task2", task2, task2_ptt).unwrap();
        let task3_guard = s.add("task3", task3, task3_ptt).unwrap();
        let task4_guard = s.add("task4", task4, task4_ptt).unwrap();
        let task5_guard = s.add("task5", task5, task5_ptt).unwrap();


        let mut a_cnt: [u8; 5] = [0_u8; 5];

        let end = AbsoluteTime::now() + RelativeTime::new_time(5, 500_000_000);

        while AbsoluteTime::now() < end
        {
            match recv.recv_timeout(Duration::from_millis(1))
            {
                Ok(rcv_a) => 
                    a_cnt[rcv_a as usize] += 1,
                Err(RecvTimeoutError::Timeout) => 
                    continue,
                Err(e) =>
                    panic!("{}", e),
            }

            
        }

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

        assert!(a_cnt[0] == 5);
        assert!(a_cnt[1] == 2);
        assert!((a_cnt[2] == 10 || a_cnt[2] == 11));
        assert!(a_cnt[3] == 27);
        assert!(a_cnt[4] == 1);

        task5_guard.reschedule_task(PeriodicTaskTime::exact_time(AbsoluteTime::now() + RelativeTime::new_time(0, 500_000_000))).unwrap();

        let end = AbsoluteTime::now() + RelativeTime::new_time(0, 600_000_000);

        while AbsoluteTime::now() < end
        {
            match recv.recv_timeout(Duration::from_millis(1))
            {
                Ok(rcv_a) => 
                    a_cnt[rcv_a as usize] += 1,
                Err(RecvTimeoutError::Timeout) => 
                    continue,
                Err(e) =>
                    panic!("{}", e),
            }
        }

        println!("{:?}", a_cnt);
        assert!(a_cnt[4] == 2);

        drop(task5_guard);
        drop(task4_guard);
        drop(task3_guard);
        drop(task2_guard);

        let end = AbsoluteTime::now() + RelativeTime::new_time(2, 1000);

        while AbsoluteTime::now() < end
        {
            match recv.recv_timeout(Duration::from_millis(1))
            {
                Ok(rcv_a) => 
                    a_cnt[rcv_a as usize] += 1,
                Err(RecvTimeoutError::Timeout) => 
                    continue,
                Err(e) =>
                    panic!("{}", e),
            }
        }


        println!("{:?}", a_cnt);
        assert_eq!(a_cnt[4], 2);
        assert_eq!(a_cnt[0], 8);

        drop(task1_guard);

        std::thread::sleep(Duration::from_millis(10));
        return;
    }
}