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

rumtk_core/
queue.rs
