rumtk_core/

threading.rs

/*
 * rumtk attempts to implement HL7 and medical protocols for interoperability in medicine.
 * This toolkit aims to be reliable, simple, performant, and standards compliant.
 * Copyright (C) 2025  Luis M. Santos, M.D. <lsantos@medicalmasses.com>
 * Copyright (C) 2025  MedicalMasses L.L.C. <contact@medicalmasses.com>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */

///
/// This module provides all the primitives needed to build a multithreaded application.
///
pub mod thread_primitives {
    pub use std::sync::Mutex as SyncMutex;
    pub use std::sync::MutexGuard as SyncMutexGuard;
    pub use std::sync::RwLock as SyncRwLock;
    use std::sync::{Arc, OnceLock};
    pub use tokio::io;
    pub use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::runtime::Runtime as TokioRuntime;
    pub use tokio::sync::{
        Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard,
        OwnedRwLockReadGuard as AsyncOwnedRwLockMappedReadGuard,
        OwnedRwLockReadGuard as AsyncOwnedRwLockReadGuard,
        OwnedRwLockWriteGuard as AsyncOwnedRwLockWriteGuard, RwLock as AsyncRwLock,
        RwLockMappedWriteGuard as AsyncRwLockMappedWriteGuard,
        RwLockReadGuard as AsyncRwLockReadGuard, RwLockWriteGuard as AsyncRwLockWriteGuard,
    };

    /**************************** Types ***************************************/
    pub type SafeLockReadGuard<T> = AsyncOwnedRwLockReadGuard<T>;
    pub type MappedLockReadGuard<T> = AsyncOwnedRwLockReadGuard<T>;
    pub type SafeLockWriteGuard<T> = AsyncOwnedRwLockWriteGuard<T>;
    pub type SafeLock<T> = Arc<AsyncRwLock<T>>;
    pub type SafeTokioRuntime = OnceLock<TokioRuntime>;
}

pub mod threading_manager {
    use crate::core::{RUMResult, RUMVec};
    use crate::strings::rumtk_format;
    use crate::threading::thread_primitives::SafeLock;
    use crate::threading::threading_functions::{async_sleep, sleep};
    use crate::types::{RUMHashMap, RUMID};
    use crate::{rumtk_init_threads, rumtk_resolve_task, rumtk_spawn_task, threading};
    use std::future::Future;
    use std::sync::Arc;
    pub use std::sync::RwLock as SyncRwLock;
    use tokio::task::JoinHandle;

    const DEFAULT_SLEEP_DURATION: f32 = 0.001f32;
    const DEFAULT_TASK_CAPACITY: usize = 100;

    pub type AsyncHandle<T> = JoinHandle<T>;
    pub type TaskItems<T> = RUMVec<T>;
    /// This type aliases a vector of T elements that will be used for passing arguments to the task processor.
    pub type TaskArgs<T> = TaskItems<T>;
    /// Function signature defining the interface of task processing logic.
    pub type SafeTaskArgs<T> = SafeLock<TaskItems<T>>;
    pub type AsyncTaskHandle<R> = AsyncHandle<TaskResult<R>>;
    pub type AsyncTaskHandles<R> = Vec<AsyncTaskHandle<R>>;
    //pub type TaskProcessor<T, R, Fut: Future<Output = TaskResult<R>>> = impl FnOnce(&SafeTaskArgs<T>) -> Fut;
    pub type TaskID = RUMID;

    #[derive(Debug, Clone, Default)]
    pub struct Task<R> {
        pub id: TaskID,
        pub finished: bool,
        pub result: Option<R>,
    }

    pub type SafeTask<R> = Arc<Task<R>>;
    type SafeInternalTask<R> = Arc<SyncRwLock<Task<R>>>;
    pub type TaskTable<R> = RUMHashMap<TaskID, SafeInternalTask<R>>;
    pub type SafeAsyncTaskTable<R> = SafeLock<TaskTable<R>>;
    pub type SafeSyncTaskTable<R> = Arc<SyncRwLock<TaskTable<R>>>;
    pub type TaskBatch = RUMVec<TaskID>;
    /// Type to use to define how task results are expected to be returned.
    pub type TaskResult<R> = RUMResult<SafeTask<R>>;
    pub type TaskResults<R> = TaskItems<TaskResult<R>>;

    ///
    /// Manages asynchronous tasks submitted as micro jobs from synchronous code. This type essentially
    /// gives the multithreading, asynchronous superpowers to synchronous logic.
    ///
    /// ## Example Usage
    ///
    /// ```
    /// use std::sync::{Arc};
    /// use tokio::sync::RwLock as AsyncRwLock;
    /// use rumtk_core::core::RUMResult;
    /// use rumtk_core::strings::RUMString;
    /// use rumtk_core::threading::threading_manager::{SafeTaskArgs, TaskItems, TaskManager};
    /// use rumtk_core::{rumtk_create_task, };
    ///
    /// let expected = vec![
    ///     RUMString::from("Hello"),
    ///     RUMString::from("World!"),
    ///     RUMString::from("Overcast"),
    ///     RUMString::from("and"),
    ///     RUMString::from("Sad"),
    ///  ];
    ///
    /// type TestResult = RUMResult<Vec<RUMString>>;
    /// let mut queue: TaskManager<TestResult> = TaskManager::new(&5).unwrap();
    ///
    /// let locked_args = AsyncRwLock::new(expected.clone());
    /// let task_args = SafeTaskArgs::<RUMString>::new(locked_args);
    /// let processor = rumtk_create_task!(
    ///     async |args: &SafeTaskArgs<RUMString>| -> TestResult {
    ///         let owned_args = Arc::clone(args);
    ///         let locked_args = owned_args.read().await;
    ///         let mut results = TaskItems::<RUMString>::with_capacity(locked_args.len());
    ///
    ///         for arg in locked_args.iter() {
    ///             results.push(RUMString::new(arg));
    ///         }
    ///
    ///         Ok(results)
    ///     },
    ///     task_args
    /// );
    ///
    /// queue.add_task::<_>(processor);
    /// let results = queue.wait();
    ///
    /// let mut result_data = Vec::<RUMString>::with_capacity(5);
    /// for r in results {
    ///     for v in r.unwrap().result.clone().unwrap().iter() {
    ///         for value in v.iter() {
    ///             result_data.push(value.clone());
    ///         }
    ///     }
    ///  }
    ///
    /// assert_eq!(result_data, expected, "Results do not match expected!");
    ///
    /// ```
    ///
    #[derive(Debug, Clone, Default)]
    pub struct TaskManager<R> {
        tasks: SafeSyncTaskTable<R>,
        workers: usize,
    }

    impl<R> TaskManager<R>
    where
        R: Sync + Send + Clone + 'static,
    {
        ///
        /// This method creates a [`TaskManager`] instance using sensible defaults.
        ///
        /// The `threads` field is computed from the number of cores present in system.
        ///
        pub fn default() -> RUMResult<TaskManager<R>> {
            Self::new(&threading::threading_functions::get_default_system_thread_count())
        }

        ///
        /// Creates an instance of [`TaskManager<R>`](TaskManager<R>).
        /// Expects you to provide the count of threads to spawn and the microtask queue size
        /// allocated by each thread.
        ///
        /// This method calls [`TaskTable::with_capacity()`](TaskTable::with_capacity) for the actual object creation.
        /// The main queue capacity is pre-allocated to [`DEFAULT_TASK_CAPACITY`](DEFAULT_TASK_CAPACITY).
        ///
        pub fn new(worker_num: &usize) -> RUMResult<TaskManager<R>> {
            let tasks = SafeSyncTaskTable::<R>::new(SyncRwLock::new(TaskTable::with_capacity(
                DEFAULT_TASK_CAPACITY,
            )));
            Ok(TaskManager::<R> {
                tasks,
                workers: worker_num.to_owned(),
            })
        }

        ///
        /// Add a task to the processing queue. The idea is that you can queue a processor function
        /// and list of args that will be picked up by one of the threads for processing.
        ///
        /// This is the async counterpart
        ///
        pub async fn add_task_async<F>(&mut self, task: F) -> TaskID
        where
            F: Future<Output = R> + Send + Sync + 'static,
            F::Output: Send + 'static,
        {
            let id = TaskID::new_v4();
            Self::_add_task_async(id.clone(), self.tasks.clone(), task).await
        }

        ///
        /// See [`Self::add_task_async`]
        ///
        /// Unlike `add_task`, this method does not block which is key to avoiding panicking
        /// the tokio runtim if trying to add task to queue from a normal function called from an
        /// async environment.
        ///
        /// ## Example
        ///
        /// ```
        /// use rumtk_core::threading::threading_manager::{TaskManager};
        /// use rumtk_core::{rumtk_init_threads, strings::rumtk_format};
        /// use std::sync::{Arc};
        /// use tokio::sync::Mutex;
        ///
        /// type JobManager = Arc<Mutex<TaskManager<usize>>>;
        ///
        /// async fn called_fn() -> usize {
        ///     5
        /// }
        ///
        /// fn push_job(manager: &mut TaskManager<usize>) -> usize {
        ///     manager.spawn_task(called_fn());
        ///     1
        /// }
        ///
        /// async fn call_sync_fn(mut manager: JobManager) -> usize {
        ///     let mut owned = manager.lock().await;
        ///     push_job(&mut owned)
        /// }
        ///
        /// let workers = 5;
        /// let mut manager = Arc::new(Mutex::new(TaskManager::new(&workers).unwrap()));
        ///
        /// manager.blocking_lock().spawn_task(call_sync_fn(manager.clone()));
        ///
        /// let result_raw = manager.blocking_lock().wait();
        ///
        /// ```
        ///
        pub fn spawn_task<F>(&mut self, task: F) -> RUMResult<TaskID>
        where
            F: Future<Output = R> + Send + Sync + 'static,
            F::Output: Send + Sized + 'static,
        {
            let id = TaskID::new_v4();
            let rt = rumtk_init_threads!(self.workers);
            rumtk_spawn_task!(
                rt,
                Self::_add_task_async(id.clone(), self.tasks.clone(), task)
            );
            Ok(id)
        }

        ///
        /// See [add_task_async](Self::add_task_async)
        ///
        pub fn add_task<F>(&mut self, task: F) -> RUMResult<TaskID>
        where
            F: Future<Output = R> + Send + Sync + 'static,
            F::Output: Send + Sized + 'static,
        {
            let id = TaskID::new_v4();
            let tasks = self.tasks.clone();
            Ok(rumtk_resolve_task!(Self::_add_task_async(id.clone(), tasks, task)))
        }

        async fn _add_task_async<F>(id: TaskID, tasks: SafeSyncTaskTable<R>, task: F) -> TaskID
        where
            F: Future<Output = R> + Send + Sync + 'static,
            F::Output: Send + Sized + 'static,
        {
            let mut safe_task = Arc::new(SyncRwLock::new(Task::<R> {
                id: id.clone(),
                finished: false,
                result: None,
            }));
            tasks.write().unwrap().insert(id.clone(), safe_task.clone());

            let task_wrapper = async move || {
                // Run the task
                let result = task.await;

                // Cleanup task
                let mut lock = safe_task.write().unwrap();
                lock.result = Some(result);
                lock.finished = true;
            };

            tokio::spawn(task_wrapper());

            id
        }

        ///
        /// See [wait_async](Self::wait_async)
        ///
        /// Duplicated here because we can't request the tokio runtime to do a quick exec for us if
        /// this function happens to be called from the async context.
        ///
        pub fn wait(&mut self) -> TaskResults<R> {
            let task_batch = self
                .tasks
                .read()
                .unwrap()
                .keys()
                .cloned()
                .collect::<Vec<_>>();
            self.wait_on_batch(&task_batch)
        }

        ///
        /// See [wait_on_batch_async](Self::wait_on_batch_async)
        ///
        /// Duplicated here because we can't request the tokio runtime to do a quick exec for us if
        /// this function happens to be called from the async context.
        ///
        pub fn wait_on_batch(&mut self, tasks: &TaskBatch) -> TaskResults<R> {
            let mut results = TaskResults::<R>::default();
            for task in tasks {
                results.push(self.wait_on(task));
            }
            results
        }

        ///
        /// See [wait_on_async](Self::wait_on_async)
        ///
        /// Duplicated here because we can't request the tokio runtime to do a quick exec for us if
        /// this function happens to be called from the async context.
        ///
        pub fn wait_on(&mut self, task_id: &TaskID) -> TaskResult<R> {
            let task = match self.tasks.write().unwrap().remove(task_id) {
                Some(task) => task.clone(),
                None => return Err(rumtk_format!("No task with id {}", task_id)),
            };

            while !task.read().unwrap().finished {
                sleep(DEFAULT_SLEEP_DURATION);
            }

            let x = Ok(Arc::new(task.write().unwrap().clone()));
            x
        }

        ///
        /// This method waits until a queued task with [TaskID](TaskID) has been processed from the main queue.
        ///
        /// We poll the status of the task every [DEFAULT_SLEEP_DURATION](DEFAULT_SLEEP_DURATION) ms.
        ///
        /// Upon completion,
        ///
        /// 2. Return the result ([TaskResults<R>](TaskResults)).
        ///
        /// This operation consumes the task.
        ///
        /// ### Note:
        /// ```text
        ///     Results returned here are not guaranteed to be in the same order as the order in which
        ///     the tasks were queued for work. You will need to pass a type as T that automatically
        ///     tracks its own id or has a way for you to resort results.
        /// ```
        pub async fn wait_on_async(&mut self, task_id: &TaskID) -> TaskResult<R> {
            let task = match self.tasks.write().unwrap().remove(task_id) {
                Some(task) => task.clone(),
                None => return Err(rumtk_format!("No task with id {}", task_id)),
            };

            while !task.read().unwrap().finished {
                async_sleep(DEFAULT_SLEEP_DURATION).await;
            }

            let x = Ok(Arc::new(task.write().unwrap().clone()));
            x
        }

        ///
        /// This method waits until a set of queued tasks with [TaskID](TaskID) has been processed from the main queue.
        ///
        /// We poll the status of the task every [DEFAULT_SLEEP_DURATION](DEFAULT_SLEEP_DURATION) ms.
        ///
        /// Upon completion,
        ///
        /// 1. We collect the results generated (if any).
        /// 2. Return the list of results ([TaskResults<R>](TaskResults)).
        ///
        /// ### Note:
        /// ```text
        ///     Results returned here are not guaranteed to be in the same order as the order in which
        ///     the tasks were queued for work. You will need to pass a type as T that automatically
        ///     tracks its own id or has a way for you to resort results.
        /// ```
        pub async fn wait_on_batch_async(&mut self, tasks: &TaskBatch) -> TaskResults<R> {
            let mut results = TaskResults::<R>::default();
            for task in tasks {
                results.push(self.wait_on_async(task).await);
            }
            results
        }

        ///
        /// This method waits until all queued tasks have been processed from the main queue.
        ///
        /// We poll the status of the main queue every [DEFAULT_SLEEP_DURATION](DEFAULT_SLEEP_DURATION) ms.
        ///
        /// Upon completion,
        ///
        /// 1. We collect the results generated (if any).
        /// 2. We reset the main task and result internal queue states.
        /// 3. Return the list of results ([TaskResults<R>](TaskResults)).
        ///
        /// This operation consumes all the tasks.
        ///
        /// ### Note:
        /// ```text
        ///     Results returned here are not guaranteed to be in the same order as the order in which
        ///     the tasks were queued for work. You will need to pass a type as T that automatically
        ///     tracks its own id or has a way for you to resort results.
        /// ```
        pub async fn wait_async(&mut self) -> TaskResults<R> {
            let task_batch = self
                .tasks
                .read()
                .unwrap()
                .keys()
                .cloned()
                .collect::<Vec<_>>();
            self.wait_on_batch_async(&task_batch).await
        }

        ///
        /// Check if all work has been completed from the task queue.
        ///
        /// ## Examples
        ///
        /// ### Sync Usage
        ///
        ///```
        /// use rumtk_core::threading::threading_manager::TaskManager;
        ///
        /// let manager = TaskManager::<usize>::new(&4).unwrap();
        ///
        /// let all_done = manager.is_all_completed();
        ///
        /// assert_eq!(all_done, true, "Empty TaskManager reports tasks are not completed!");
        ///
        /// ```
        ///
        pub fn is_all_completed(&self) -> bool {
            self._is_all_completed_async()
        }

        pub async fn is_all_completed_async(&self) -> bool {
            self._is_all_completed_async()
        }

        fn _is_all_completed_async(&self) -> bool {
            for (_, task) in self.tasks.read().unwrap().iter() {
                if !task.read().unwrap().finished {
                    return false;
                }
            }

            true
        }

        ///
        /// Check if a task completed
        ///
        pub fn is_finished(&self, id: &TaskID) -> bool {
            match self.tasks.read().unwrap().get(id) {
                Some(t) => t.read().unwrap().finished,
                None => false,
            }
        }

        pub async fn is_finished_async(&self, id: &TaskID) -> bool {
            match self.tasks.read().unwrap().get(id) {
                Some(task) => task.read().unwrap().finished,
                None => true,
            }
        }

        ///
        /// Alias for [wait](TaskManager::wait).
        ///
        fn gather(&mut self) -> TaskResults<R> {
            self.wait()
        }
    }
}

///
/// This module contains a few helper.
///
/// For example, you can find a function for determining number of threads available in system.
/// The sleep family of functions are also here.
///
pub mod threading_functions {
    use crate::core::RUMResult;
    use crate::net::tcp::{AsyncOwnedRwLockReadGuard, AsyncOwnedRwLockWriteGuard, SafeLockReadGuard, SafeLockWriteGuard, SafeTokioRuntime};
    use crate::threading::thread_primitives::{AsyncRwLock, SafeLock};
    use num_cpus;
    use std::future::Future;
    use std::sync::Arc;
    use std::thread::{available_parallelism, sleep as std_sleep};
    use std::time::Duration;
    use tokio::runtime::Runtime;
    use tokio::time::sleep as tokio_sleep;
    /**************************** Globals **************************************/
    static mut DEFAULT_RUNTIME: SafeTokioRuntime = SafeTokioRuntime::new();

    pub const NANOS_PER_SEC: u64 = 1000000000;
    pub const MILLIS_PER_SEC: u64 = 1000;
    pub const MICROS_PER_SEC: u64 = 1000000;
    const DEFAULT_SLEEP_DURATION: f32 = 0.001;
    /**************************** Helpers **************************************/
    pub fn init_runtime<'a>(workers: usize) -> &'a Runtime {
        unsafe {
            let runtime = DEFAULT_RUNTIME.get_or_init(|| {
                let mut builder = tokio::runtime::Builder::new_multi_thread();
                builder.worker_threads(workers);
                builder.enable_all();
                match builder.build() {
                    Ok(handle) => handle,
                    Err(e) => panic!(
                        "Unable to initialize threading tokio runtime because {}!",
                        &e
                    ),
                }
            });
            runtime
        }
    }

    pub fn get_default_system_thread_count() -> usize {
        let cpus: usize = num_cpus::get();
        let parallelism = match available_parallelism() {
            Ok(n) => n.get(),
            Err(_) => 0,
        };

        if parallelism >= cpus {
            parallelism
        } else {
            cpus
        }
    }

    pub fn sleep(s: f32) {
        let ns = s * NANOS_PER_SEC as f32;
        let rounded_ns = ns.round() as u64;
        let duration = Duration::from_nanos(rounded_ns);
        std_sleep(duration);
    }

    pub async fn async_sleep(s: f32) {
        let ns = s * NANOS_PER_SEC as f32;
        let rounded_ns = ns.round() as u64;
        let duration = Duration::from_nanos(rounded_ns);
        tokio_sleep(duration).await;
    }

    ///
    /// Given a closure task, push it onto the current `tokio` runtime for execution.
    /// Every [DEFAULT_SLEEP_DURATION] seconds, we check if the task has concluded.
    /// Once the task has concluded, we call [tokio::block_on](tokio::task::block_in_place) to resolve and extract the task
    /// result.
    ///
    /// Because this helper function can fail, the return value is wrapped inside a [RUMResult].
    ///
    /// ## Example
    ///
    /// ```
    /// use rumtk_core::threading::threading_functions::{init_runtime, block_on_task};
    ///
    /// const Hello: &str = "World!";
    ///
    /// init_runtime(5);
    ///
    /// let result = block_on_task(async {
    ///     Hello
    /// });
    ///
    /// assert_eq!(Hello, result, "Result mismatches expected! {} vs. {}", Hello, result);
    /// ```
    ///
    pub fn block_on_task<R, F>(task: F) -> R
    where
        F: Future<Output = R> + Send + 'static,
        F::Output: Send + 'static,
    {
        let rt = init_runtime(get_default_system_thread_count());
        // You need to wrap our call to block_on with a call to tokio::task::block_in_place to force 
        // cleanup of async executor and therefore avoid panics from the tokio runtime!
        // Per Tokio's documentation, spawn_blocking would be better since it moves the task to an 
        // executor meant for blocking tasks instead of moving tasks out of the current thread and 
        // converting the thread into a clocking executor. The reason we don't do that is because 
        // the call to this function expects to block the current thread until completion and then 
        // return the result. If there's an issue with IO, revisit this function.
        //
        // https://docs.rs/tokio/latest/tokio/task/fn.block_in_place.html
        // https://docs.rs/tokio/latest/tokio/runtime/struct.Runtime.html#method.block_on
        // https://docs.rs/tokio/latest/tokio/runtime/struct.Handle.html#method.spawn_blocking
        // https://docs.rs/tokio/latest/tokio/runtime/struct.Handle.html#method.spawn_blocking
        tokio::task::block_in_place(move || {
            rt.block_on(task)
        })
    }

    pub fn new_lock<T>(data: T) -> SafeLock<T> {
        Arc::new(AsyncRwLock::new(data))
    }

    ///
    /// This function gives you read access to underlying structure.
    ///
    /// Helper function for executing microtask immediately after locking the spin lock. This function
    /// should be used in situations in which you want to minimize the risk of Time of Check Time of
    /// Use security bugs.
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::core::RUMResult;
    /// use rumtk_core::threading::thread_primitives::SafeLock;
    /// use rumtk_core::threading::threading_functions::{new_lock, process_read_critical_section};
    ///
    /// let data = 5;
    /// let lock = new_lock(data.clone());
    /// let result = process_read_critical_section(lock, |guard| -> RUMResult<i32> {
    ///     Ok(*guard)
    /// }).unwrap();
    ///
    /// assert_eq!(result, data, "Failed to execute critical section through which we retrieve the locked data!");
    /// ```
    ///
    pub fn process_read_critical_section<T, R, F>(
        lock: SafeLock<T>,
        critical_section: F,
    ) -> R
    where 
        F: Fn(SafeLockReadGuard<T>) -> R, T: Send + Sync + 'static,
    {
        tokio::task::block_in_place(move || {
            let read_guard = lock_read(lock);
            critical_section(read_guard)
        })
    }

    ///
    /// This function gives you write access to underlying structure.
    ///
    /// Helper function for executing microtask immediately after locking the spin lock. This function
    /// should be used in situations in which you want to minimize the risk of Time of Check Time of
    /// Use security bugs.
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::core::RUMResult;
    /// use rumtk_core::threading::thread_primitives::SafeLock;
    /// use rumtk_core::threading::threading_functions::{new_lock, process_write_critical_section};
    ///
    /// let data = 5;
    /// let lock = new_lock(data.clone());
    /// let new_data = 10;
    /// let result = process_write_critical_section(lock, |mut guard| -> RUMResult<i32> {
    ///     *guard = new_data;
    ///     Ok(*guard)
    /// }).unwrap();
    ///
    /// assert_eq!(result, new_data, "Failed to execute critical section through which we modify the locked data!");
    /// ```
    ///
    pub fn process_write_critical_section<T, R, F>(
        lock: SafeLock<T>,
        critical_section: F,
    ) -> R
    where
        F: Fn(SafeLockWriteGuard<T>) -> R, T: Send + Sync + 'static,
    {
        tokio::task::block_in_place(move || {
            let write_guard = lock_write(lock);
            critical_section(write_guard)
        })
    }

    ///
    /// Obtain read guard to standard spin lock such that you have a more ergonomic interface to
    /// locked data.
    ///
    /// It is preferable to use [process_read_critical_section] when you must process
    /// critical logic that is sensitive to time of check time of use security bugs!
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::threading::thread_primitives::SafeLock;
    /// use rumtk_core::threading::threading_functions::{new_lock, lock_read};
    ///
    /// let data = 5;
    /// let lock = new_lock(data.clone());
    /// let result = *lock_read(lock);
    ///
    /// assert_eq!(result, data, "Failed to access the locked data!");
    /// ```
    ///
    pub fn lock_read<T: Send + Sync + 'static>(lock: SafeLock<T>) -> AsyncOwnedRwLockReadGuard<T> {
        block_on_task(async move {
            lock.read_owned().await
        })
    }

    ///
    /// Obtain write guard to standard spin lock such that you have a more ergonomic interface to
    /// locked data.
    ///
    /// It is preferable to use [process_write_critical_section] when you must process
    /// critical logic that is sensitive to time of check time of use security bugs!
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::threading::thread_primitives::SafeLock;
    /// use rumtk_core::threading::threading_functions::{new_lock, lock_read, lock_write};
    ///
    /// let data = 5;
    /// let lock = new_lock(data.clone());
    /// let new_data = 10;
    ///
    /// *lock_write(lock.clone()) = new_data;
    ///
    /// let result = *lock_read(lock);
    ///
    /// assert_eq!(result, new_data, "Failed to modify the locked data!");
    /// ```
    ///
    pub fn lock_write<T: Send + Sync + 'static>(lock: SafeLock<T>) -> AsyncOwnedRwLockWriteGuard<T> {
        block_on_task(async move {
            lock.write_owned().await
        })
    }
}

///
/// Main API for interacting with the threading back end. Remember, we use tokio as our executor.
/// This means that by default, all jobs sent to the thread pool have to be async in nature.
/// These macros make handling of these jobs at the sync/async boundary more convenient.
///
pub mod threading_macros {
    use crate::threading::thread_primitives;
    use crate::threading::threading_manager::SafeTaskArgs;

    ///
    /// First, let's make sure we have *tokio* initialized at least once. The runtime created here
    /// will be saved to the global context so the next call to this macro will simply grab a
    /// reference to the previously initialized runtime.
    ///
    /// Passing nothing will default to initializing a runtime using the default number of threads
    /// for this system. This is typically equivalent to number of cores/threads for your CPU.
    ///
    /// Passing `threads` number will yield a runtime that allocates that many threads.
    ///
    ///
    /// ## Examples
    ///
    /// ```
    ///     use rumtk_core::{rumtk_init_threads, rumtk_resolve_task, rumtk_create_task_args, rumtk_create_task, rumtk_spawn_task};
    ///     use rumtk_core::core::RUMResult;
    ///     use rumtk_core::threading::threading_manager::SafeTaskArgs;
    ///
    ///     async fn test(args: &SafeTaskArgs<i32>) -> RUMResult<Vec<i32>> {
    ///         let mut result = Vec::<i32>::new();
    ///         for arg in args.read().await.iter() {
    ///             result.push(*arg);
    ///         }
    ///         Ok(result)
    ///     }
    ///
    ///     let args = rumtk_create_task_args!(1);                               // Creates a vector of i32s
    ///     let task = rumtk_create_task!(test, args);                           // Creates a standard task which consists of a function or closure accepting a Vec<T>
    ///     let result = rumtk_resolve_task!(task); // Spawn's task and waits for it to conclude.
    /// ```
    ///
    /// ```
    ///     use rumtk_core::{rumtk_init_threads, rumtk_resolve_task, rumtk_create_task_args, rumtk_create_task, rumtk_spawn_task};
    ///     use rumtk_core::core::RUMResult;
    ///     use rumtk_core::threading::threading_manager::SafeTaskArgs;
    ///
    ///     async fn test(args: &SafeTaskArgs<i32>) -> RUMResult<Vec<i32>> {
    ///         let mut result = Vec::<i32>::new();
    ///         for arg in args.read().await.iter() {
    ///             result.push(*arg);
    ///         }
    ///         Ok(result)
    ///     }
    ///
    ///     let thread_count: usize = 10;
    ///     let args = rumtk_create_task_args!(1);
    ///     let task = rumtk_create_task!(test, args);
    ///     let result = rumtk_resolve_task!(task);
    /// ```
    #[macro_export]
    macro_rules! rumtk_init_threads {
        ( ) => {{
            use $crate::threading::threading_functions::{
                get_default_system_thread_count, init_runtime,
            };
            init_runtime(get_default_system_thread_count())
        }};
        ( $threads:expr ) => {{
            use $crate::rumtk_cache_fetch;
            use $crate::threading::threading_functions::init_runtime;
            init_runtime($threads)
        }};
    }

    ///
    /// Puts task onto the runtime queue.
    ///
    /// The parameters to this macro are a reference to the runtime (`rt`) and a future (`func`).
    ///
    /// The return is a [thread_primitives::JoinHandle<T>] instance. If the task was a standard
    /// framework task, you will get [thread_primitives::AsyncTaskHandle] instead.
    ///
    #[macro_export]
    macro_rules! rumtk_spawn_task {
        ( $func:expr ) => {{
            use $crate::rumtk_init_threads;
            let rt = rumtk_init_threads!().expect("Runtime is not initialized!");
            rt.spawn($func)
        }};
        ( $rt:expr, $func:expr ) => {{
            $rt.spawn($func)
        }};
    }

    ///
    /// Using the initialized runtime, wait for the future to resolve in a thread blocking manner!
    ///
    /// If you pass a reference to the runtime (`rt`) and an async closure (`func`), we await the
    /// async closure without passing any arguments.
    ///
    /// You can pass a third argument to this macro in the form of any number of arguments (`arg_item`).
    /// In such a case, we pass those arguments to the call on the async closure and await on results.
    ///
    #[macro_export]
    macro_rules! rumtk_wait_on_task {
        ( $func:expr ) => {{
            use $crate::threading::threading_functions::block_on_task;
            block_on_task(async move {
                $func().await
            })
        }};
        ( $func:expr, $($arg_items:expr),+ ) => {{
            use $crate::threading::threading_functions::block_on_task;
            block_on_task(async move {
                $func($($arg_items),+).await
            })
        }};
    }

    ///
    /// This macro awaits a future.
    ///
    /// The arguments are a reference to the runtime (`rt) and a future.
    ///
    /// If there is a result, you will get the result of the future.
    ///
    /// ## Examples
    ///
    /// ```
    ///     use rumtk_core::{rumtk_init_threads, rumtk_resolve_task, rumtk_create_task_args, rumtk_create_task, rumtk_spawn_task};
    ///     use rumtk_core::core::RUMResult;
    ///     use rumtk_core::threading::threading_manager::SafeTaskArgs;
    ///
    ///     async fn test(args: &SafeTaskArgs<i32>) -> RUMResult<Vec<i32>> {
    ///         let mut result = Vec::<i32>::new();
    ///         for arg in args.read().await.iter() {
    ///             result.push(*arg);
    ///         }
    ///         Ok(result)
    ///     }
    ///
    ///     let args = rumtk_create_task_args!(1);
    ///     let task = rumtk_create_task!(test, args);
    ///     let result = rumtk_resolve_task!(task);
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_resolve_task {
        ( $future:expr ) => {{
            use $crate::threading::threading_functions::block_on_task;
            // Fun tidbit, the expression rumtk_resolve_task!(&rt, rumtk_spawn_task!(&rt, task)), where
            // rt is the tokio runtime yields async move { { &rt.spawn(task) } }. However, the whole thing
            // is technically moved into the async closure and captured so things like mutex guards
            // technically go out of the outer scope. As a result that expression fails to compile even
            // though the intent is for rumtk_spawn_task to resolve first and its result get moved
            // into the async closure. To ensure that happens regardless of given expression, we do
            // a variable assignment below to force the "future" macro expressions to resolve before
            // moving into the closure. DO NOT REMOVE OR "SIMPLIFY" THE let future = $future LINE!!!
            //let future = $future;
            block_on_task(async move { $future.await })
        }};
    }

    #[macro_export]
    macro_rules! rumtk_resolve_task_from_async {
        ( $rt:expr, $future:expr ) => {{
            let handle = $rt.spawn_blocking(async move { future.await })
        }};
    }

    ///
    /// This macro creates an async body that calls the async closure and awaits it.
    ///
    /// ## Example
    ///
    /// ```
    /// use std::sync::{Arc, RwLock};
    /// use tokio::sync::RwLock as AsyncRwLock;
    /// use rumtk_core::strings::RUMString;
    /// use rumtk_core::threading::threading_manager::{SafeTaskArgs, TaskItems};
    ///
    /// pub type SafeTaskArgs2<T> = Arc<RwLock<TaskItems<T>>>;
    /// let expected = vec![
    ///     RUMString::from("Hello"),
    ///     RUMString::from("World!"),
    ///     RUMString::from("Overcast"),
    ///     RUMString::from("and"),
    ///     RUMString::from("Sad"),
    ///  ];
    /// let locked_args = AsyncRwLock::new(expected.clone());
    /// let task_args = SafeTaskArgs::<RUMString>::new(locked_args);
    ///
    ///
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_create_task {
        ( $func:expr ) => {{
            async move {
                let f = $func;
                f().await
            }
        }};
        ( $func:expr, $args:expr ) => {{
            let f = $func;
            async move { f(&$args).await }
        }};
    }

    ///
    /// Creates an instance of [SafeTaskArgs](SafeTaskArgs) with the arguments passed.
    ///
    /// ## Note
    ///
    /// All arguments must be of the same type
    ///
    #[macro_export]
    macro_rules! rumtk_create_task_args {
        ( ) => {{
            use $crate::threading::threading_manager::{TaskArgs, SafeTaskArgs, TaskItems};
            use $crate::threading::thread_primitives::AsyncRwLock;
            SafeTaskArgs::new(AsyncRwLock::new(vec![]))
        }};
        ( $($args:expr),+ ) => {{
            use $crate::threading::threading_manager::{SafeTaskArgs};
            use $crate::threading::thread_primitives::AsyncRwLock;
            SafeTaskArgs::new(AsyncRwLock::new(vec![$($args),+]))
        }};
    }

    ///
    /// Convenience macro for packaging the task components and launching the task in one line.
    ///
    /// One of the advantages is that you can generate a new `tokio` runtime by specifying the
    /// number of threads at the end. This is optional. Meaning, we will default to the system's
    /// number of threads if that value is not specified.
    ///
    /// Between the `func` parameter and the optional `threads` parameter, you can specify a
    /// variable number of arguments to pass to the task. each argument must be of the same type.
    /// If you wish to pass different arguments with different types, please define an abstract type
    /// whose underlying structure is a tuple of items and pass that instead.
    ///
    /// ## Examples
    ///
    /// ### With Default Thread Count
    /// ```
    ///     use rumtk_core::{rumtk_exec_task};
    ///     use rumtk_core::core::RUMResult;
    ///     use rumtk_core::threading::threading_manager::SafeTaskArgs;
    ///
    ///     async fn test(args: &SafeTaskArgs<i32>) -> RUMResult<Vec<i32>> {
    ///         let mut result = Vec::<i32>::new();
    ///         for arg in args.read().await.iter() {
    ///             result.push(*arg);
    ///         }
    ///         Ok(result)
    ///     }
    ///
    ///     let result = rumtk_exec_task!(test, vec![5]).unwrap();
    ///     assert_eq!(&result.clone(), &vec![5], "Results mismatch");
    ///     assert_ne!(&result.clone(), &vec![5, 10], "Results do not mismatch as expected!");
    /// ```
    ///
    /// ### With Custom Thread Count
    /// ```
    ///     use rumtk_core::{rumtk_exec_task};
    ///     use rumtk_core::core::RUMResult;
    ///     use rumtk_core::threading::threading_manager::SafeTaskArgs;
    ///
    ///     async fn test(args: &SafeTaskArgs<i32>) -> RUMResult<Vec<i32>> {
    ///         let mut result = Vec::<i32>::new();
    ///         for arg in args.read().await.iter() {
    ///             result.push(*arg);
    ///         }
    ///         Ok(result)
    ///     }
    ///
    ///     let result = rumtk_exec_task!(test, vec![5], 5).unwrap();
    ///     assert_eq!(&result.clone(), &vec![5], "Results mismatch");
    ///     assert_ne!(&result.clone(), &vec![5, 10], "Results do not mismatch as expected!");
    /// ```
    ///
    /// ### With Async Function Body
    /// ```
    ///     use rumtk_core::{rumtk_exec_task};
    ///     use rumtk_core::core::RUMResult;
    ///     use rumtk_core::threading::threading_manager::SafeTaskArgs;
    ///
    ///     let result = rumtk_exec_task!(
    ///     async move |args: &SafeTaskArgs<i32>| -> RUMResult<Vec<i32>> {
    ///         let mut result = Vec::<i32>::new();
    ///         for arg in args.read().await.iter() {
    ///             result.push(*arg);
    ///         }
    ///         Ok(result)
    ///     },
    ///     vec![5]).unwrap();
    ///     assert_eq!(&result.clone(), &vec![5], "Results mismatch");
    ///     assert_ne!(&result.clone(), &vec![5, 10], "Results do not mismatch as expected!");
    /// ```
    ///
    /// ### With Async Function Body and No Args
    /// ```
    ///     use rumtk_core::{rumtk_exec_task};
    ///     use rumtk_core::core::RUMResult;
    ///     use rumtk_core::threading::threading_manager::SafeTaskArgs;
    ///
    ///     let result = rumtk_exec_task!(
    ///     async || -> RUMResult<Vec<i32>> {
    ///         let mut result = Vec::<i32>::new();
    ///         Ok(result)
    ///     }).unwrap();
    ///     let empty = Vec::<i32>::new();
    ///     assert_eq!(&result.clone(), &empty, "Results mismatch");
    ///     assert_ne!(&result.clone(), &vec![5, 10], "Results do not mismatch as expected!");
    /// ```
    ///
    /// ## Equivalent To
    ///
    /// ```no_run
    ///     use rumtk_core::{rumtk_init_threads, rumtk_resolve_task, rumtk_create_task_args, rumtk_create_task, rumtk_spawn_task};
    ///     use rumtk_core::core::RUMResult;
    ///     use rumtk_core::threading::threading_manager::SafeTaskArgs;
    ///
    ///     async fn test(args: &SafeTaskArgs<i32>) -> RUMResult<Vec<i32>> {
    ///         let mut result = Vec::<i32>::new();
    ///         for arg in args.read().await.iter() {
    ///             result.push(*arg);
    ///         }
    ///         Ok(result)
    ///     }
    ///
    ///     let args = rumtk_create_task_args!(1);
    ///     let task = rumtk_create_task!(test, args);
    ///     let result = rumtk_resolve_task!(task);
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_exec_task {
        ($func:expr ) => {{
            use $crate::{
                rumtk_create_task, rumtk_create_task_args, rumtk_init_threads, rumtk_resolve_task,
            };
            let task = rumtk_create_task!($func);
            rumtk_resolve_task!(task)
        }};
        ($func:expr, $args:expr ) => {{
            use $crate::threading::threading_functions::get_default_system_thread_count;
            rumtk_exec_task!($func, $args, get_default_system_thread_count())
        }};
        ($func:expr, $args:expr , $threads:expr ) => {{
            use $crate::threading::thread_primitives::AsyncRwLock;
            use $crate::{
                rumtk_create_task, rumtk_create_task_args, rumtk_init_threads, rumtk_resolve_task,
            };
            let args = SafeTaskArgs::new(AsyncRwLock::new($args));
            let task = rumtk_create_task!($func, args);
            rumtk_resolve_task!(task)
        }};
    }

    ///
    /// Sleep a duration of time in a sync context, so no await can be call on the result.
    ///
    /// You can pass any value that can be cast to f32.
    ///
    /// The precision is up to nanoseconds and it is depicted by the number of decimal places.
    ///
    /// ## Examples
    ///
    /// ```
    ///     use rumtk_core::rumtk_sleep;
    ///     rumtk_sleep!(1);           // Sleeps for 1 second.
    ///     rumtk_sleep!(0.001);       // Sleeps for 1 millisecond
    ///     rumtk_sleep!(0.000001);    // Sleeps for 1 microsecond
    ///     rumtk_sleep!(0.000000001); // Sleeps for 1 nanosecond
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_sleep {
        ( $dur:expr) => {{
            use $crate::threading::threading_functions::sleep;
            sleep($dur as f32)
        }};
    }

    ///
    /// Sleep for some duration of time in an async context. Meaning, we can be awaited.
    ///
    /// You can pass any value that can be cast to f32.
    ///
    /// The precision is up to nanoseconds and it is depicted by the number of decimal places.
    ///
    /// ## Examples
    ///
    /// ```
    ///     use rumtk_core::{rumtk_async_sleep, rumtk_exec_task};
    ///     use rumtk_core::core::RUMResult;
    ///     rumtk_exec_task!( async || -> RUMResult<()> {
    ///             rumtk_async_sleep!(1).await;           // Sleeps for 1 second.
    ///             rumtk_async_sleep!(0.001).await;       // Sleeps for 1 millisecond
    ///             rumtk_async_sleep!(0.000001).await;    // Sleeps for 1 microsecond
    ///             rumtk_async_sleep!(0.000000001).await; // Sleeps for 1 nanosecond
    ///             Ok(())
    ///         }
    ///     );
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_async_sleep {
        ( $dur:expr) => {{
            use $crate::threading::threading_functions::async_sleep;
            async_sleep($dur as f32)
        }};
    }

    ///
    ///
    ///
    #[macro_export]
    macro_rules! rumtk_new_task_queue {
        ( $worker_num:expr ) => {{
            use $crate::threading::threading_manager::TaskManager;
            TaskManager::new($worker_num);
        }};
    }

    ///
    /// Creates a new safe lock to guard the given data. This interface was created to cleanup lock
    /// management for consumers of framework!
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::{rumtk_new_lock};
    ///
    /// let data = 5;
    /// let lock = rumtk_new_lock!(data);
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_new_lock {
        ( $data:expr ) => {{
            use $crate::threading::threading_functions::new_lock;
            new_lock($data)
        }};
    }

    ///
    /// Using a standard spin lock [SafeLock](thread_primitives::SafeLock), lock it and execute the
    /// critical section. The critical section itself is a synchronous function or closure. In this case,
    /// the critical section simply retrieves a value from a guarded dataset.
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::core::RUMResult;
    /// use rumtk_core::{rumtk_new_lock, rumtk_critical_section_read};
    ///
    /// let data = 5;
    /// let lock = rumtk_new_lock!(data);
    /// let result = rumtk_critical_section_read!(
    ///     lock,
    ///     |guard| -> RUMResult<i32> {
    ///         let result: i32 = *guard;
    ///         Ok(result)
    ///     }
    /// ).expect("No errors locking!");
    ///
    /// assert_eq!(result, data, "Critical section yielded invalid result!");
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_critical_section_read {
        ( $lock:expr, $function:expr ) => {{
            use $crate::threading::threading_functions::process_read_critical_section;
            process_read_critical_section($lock, $function)
        }};
    }

    ///
    /// Using a standard spin lock [SafeLock](thread_primitives::SafeLock), lock it and execute the
    /// critical section. The critical section itself is a synchronous function or closure. In this case,
    /// the critical section attempts to modify the internal state of a guarded dataset.
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::{rumtk_new_lock, rumtk_critical_section_write};
    ///
    /// let data = 5;
    /// let new_data = 10;
    /// let lock = rumtk_new_lock!(data);
    /// let result = rumtk_critical_section_write!(
    ///     lock,
    ///     |mut guard| {
    ///         *guard = new_data;
    ///     }
    /// );
    ///
    /// assert_eq!(result, (), "Critical section yielded invalid result!");
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_critical_section_write {
        ( $lock:expr, $function:expr ) => {{
            use $crate::threading::threading_functions::process_write_critical_section;
            process_write_critical_section($lock, $function)
        }};
    }

    ///
    /// Framework interface to obtain a `read` guard to the locked data.
    /// To access the internal data, you will need to dereference the guard (`*guard`).
    ///
    /// It is preferred to use [rumtk_critical_section_read] if you need to avoid `time of check time
    /// of use` security bugs.
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::{rumtk_new_lock, rumtk_lock_read};
    ///
    /// let data = 5;
    /// let lock = rumtk_new_lock!(data.clone());
    /// let result = *rumtk_lock_read!(lock);
    ///
    /// assert_eq!(result, data, "Failed to access locked data.");
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_lock_read {
        ( $lock:expr ) => {{
            use $crate::threading::threading_functions::lock_read;
            lock_read($lock.clone())
        }};
    }

    ///
    /// Framework interface to obtain a `write` guard to the locked data.
    /// To access the internal data, you will need to dereference the guard (`*guard`).
    ///
    /// It is preferred to use [rumtk_critical_section_write] if you need to avoid `time of check time
    /// of use` security bugs.
    ///
    /// ## Example
    /// ```
    /// use rumtk_core::{rumtk_new_lock, rumtk_lock_read, rumtk_lock_write};
    ///
    /// let data = 5;
    /// let lock = rumtk_new_lock!(data.clone());
    /// let new_data = 10;
    ///
    /// *rumtk_lock_write!(lock) = new_data;
    /// let result = *rumtk_lock_read!(lock);
    ///
    /// assert_eq!(result, new_data, "Failed to modify locked data.");
    /// ```
    ///
    #[macro_export]
    macro_rules! rumtk_lock_write {
        ( $lock:expr ) => {{
            use $crate::threading::threading_functions::lock_write;
            lock_write($lock.clone())
        }};
    }
}