rumtk_core/
queue.rs

1/*
2 * rumtk attempts to implement HL7 and medical protocols for interoperability in medicine.
3 * This toolkit aims to be reliable, simple, performant, and standards compliant.
4 * Copyright (C) 2025  Luis M. Santos, M.D.
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2.1 of the License, or (at your option) any later version.
10 *
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14 * Lesser General Public License for more details.
15 *
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, write to the Free Software
18 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
19 */
20pub mod queue {
21    use crate::core::RUMResult;
22    use crate::threading::thread_primitives::*;
23    use crate::{rumtk_init_threads, rumtk_resolve_task, rumtk_spawn_task, threading};
24    use std::future::Future;
25    use std::thread::sleep;
26    use std::time::Duration;
27
28    pub const DEFAULT_SLEEP_DURATION: Duration = Duration::from_millis(1);
29    pub const DEFAULT_QUEUE_CAPACITY: usize = 10;
30    pub const DEFAULT_MICROTASK_QUEUE_CAPACITY: usize = 5;
31
32    pub struct TaskQueue<R> {
33        tasks: AsyncTaskHandles<R>,
34        runtime: &'static SafeTokioRuntime,
35    }
36
37    impl<R> TaskQueue<R>
38    where
39        R: Sync + Send + Clone + 'static,
40    {
41        ///
42        /// This method creates a [`TaskQueue`] instance using sensible defaults.
43        ///
44        /// The `threads` field is computed from the number of cores present in system.
45        ///
46        pub fn default() -> RUMResult<TaskQueue<R>> {
47            Self::new(&threading::threading_functions::get_default_system_thread_count())
48        }
49
50        ///
51        /// Creates an instance of [`ThreadedTaskQueue<T, R>`] in the form of [`SafeThreadedTaskQueue<T, R>`].
52        /// Expects you to provide the count of threads to spawn and the microtask queue size
53        /// allocated by each thread.
54        ///
55        /// This method calls [`Self::with_capacity()`] for the actual object creation.
56        /// The main queue capacity is pre-allocated to [`DEFAULT_QUEUE_CAPACITY`].
57        ///
58        pub fn new(worker_num: &usize) -> RUMResult<TaskQueue<R>> {
59            let tasks = AsyncTaskHandles::with_capacity(DEFAULT_QUEUE_CAPACITY);
60            let runtime = rumtk_init_threads!(&worker_num);
61            Ok(TaskQueue { tasks, runtime })
62        }
63
64        ///
65        /// Add a task to the processing queue. The idea is that you can queue a processor function
66        /// and list of args that will be picked up by one of the threads for processing.
67        ///
68        pub fn add_task<F>(&mut self, task: F)
69        where
70            F: Future<Output = TaskResult<R>> + Send + Sync + 'static,
71            F::Output: Send + 'static,
72        {
73            let handle = rumtk_spawn_task!(&self.runtime, task);
74            self.tasks.push(handle);
75        }
76
77        ///
78        /// This method waits until all queued tasks have been processed from the main queue.
79        ///
80        /// We poll the status of the main queue every [`DEFAULT_SLEEP_DURATION`] ms.
81        ///
82        /// Upon completion,
83        ///
84        /// 1. We collect the results generated (if any).
85        /// 2. We reset the main task and result internal queue states.
86        /// 3. Return the list of results ([`TaskResults<R>`]).
87        ///
88        /// ### Note:
89        /// ```text
90        ///     Results returned here are not guaranteed to be in the same order as the order in which
91        ///     the tasks were queued for work. You will need to pass a type as T that automatically
92        ///     tracks its own id or has a way for you to resort results.
93        /// ```
94        pub fn wait(&mut self) -> TaskResults<R> {
95            while !self.is_completed() {
96                sleep(DEFAULT_SLEEP_DURATION);
97            }
98
99            let results = self.gather();
100            self.reset();
101            results
102        }
103
104        ///
105        /// Check if all work has been completed from the task queue.
106        ///
107        /// This implementation is branchless.
108        ///
109        pub fn is_completed(&self) -> bool {
110            let mut accumulator: usize = 0;
111
112            if self.tasks.is_empty() {
113                return false;
114            }
115
116            for task in self.tasks.iter() {
117                accumulator += task.is_finished() as usize;
118            }
119            (accumulator / self.tasks.len()) > 0
120        }
121
122        ///
123        /// Reset task queue and results queue states.
124        ///
125        pub fn reset(&mut self) {
126            self.tasks.clear();
127        }
128
129        fn gather(&mut self) -> TaskResults<R> {
130            let mut result_queue = TaskResults::<R>::with_capacity(self.tasks.len());
131            for i in 0..self.tasks.len() {
132                let task = self.tasks.pop().unwrap();
133                result_queue.push(rumtk_resolve_task!(&self.runtime, task).unwrap());
134            }
135            result_queue
136        }
137    }
138}
139
140pub mod queue_macros {
141    #[macro_export]
142    macro_rules! rumtk_new_task_queue {
143        ( $worker_num:expr ) => {{
144            use $crate::queue::queue::TaskQueue;
145            TaskQueue::new($worker_num);
146        }};
147    }
148}
rumtk_core/queue.rs

rumtk_core/
queue.rs
