neptune_job_queue/lib.rs
1//! This module implements a prioritized, heterogenous job queue that sends
2//! completed job results of arbitrary type to the initiator/caller.
3//!
4//! This is intended for running heavy multi-threaded jobs that should be run
5//! one at a time to avoid resource contention. By using this queue, multiple
6//! (async) tasks can initiate these tasks and wait for results without need
7//! of any other synchronization.
8//!
9//! note: Other rust job queues investigated cerca 2024 either did not support
10//! waiting for job results or else were overly complicated, requiring backend
11//! database, etc.
12//!
13//! Both blocking and non-blocking (async) jobs are supported. Non-blocking jobs
14//! are called inside spawn_blocking() in order to execute on tokio's blocking
15//! thread-pool. Async jobs are simply awaited.
16//!
17//! It supports prioritizing Jobs. The order of job execution is not a simple
18//! FIFO or LIFO but rather depends on the assigned priority of each job.
19//! Job priority level can be specified via any type that implements [Ord]
20//! such as a custom enum.
21//!
22//! There is no upper limit on the number of jobs. (except RAM).
23//!
24//! Jobs may be of mixed (heterogenous) types in a single [JobQueue] instance.
25//! Any type that implements the [Job](traits::Job) trait may be a job.
26//!
27//! Jobs may be async or blocking. Both types can be run in the same JobQueue
28//! instance concurrently.
29//!
30//! Job results also may be of any type. Typically each type of Job will return
31//! a single concrete result type. A [JobResultWrapper] is provided to
32//! facilitate this usage pattern.
33//!
34//! Each Job has an associated [JobHandle] that is used to await or cancel the
35//! job. If the `JobHandle` is dropped, the job will be cancelled.
36//!
37//! ## hello job-queue world.
38//!
39//! Here we demonstrate the most basic usage by creating a `HelloJobAsync` job
40//! and running it once in the JobQueue.
41//!
42//! We choose an async job for this example because it's a little bit simpler.
43//! We don't have to check for job-cancellation in the job itself.
44//!
45//! ```
46//! use neptune_job_queue::JobResultWrapper;
47//! use neptune_job_queue::JobQueue;
48//! use neptune_job_queue::traits::*;
49//!
50//! // define our custom job type that just returns "hello <name>"
51//! pub struct HelloJobAsync(String);
52//!
53//! // implement Job trait.
54//! #[async_trait::async_trait]
55//! impl Job for HelloJobAsync {
56//! // indicate that we are an async Job
57//! fn is_async(&self) -> bool {
58//! true
59//! }
60//!
61//! // as an async job we must impl run_async() or run_async_cancellable()
62//! async fn run_async(&self) -> Box<dyn JobResult> {
63//! let job_result = format!("hello {}", self.0);
64//! JobResultWrapper::<String>::new(job_result).into()
65//! }
66//! }
67//!
68//! #[tokio::main]
69//! async fn main() -> anyhow::Result<()> {
70//! // we choose a simple u8 for prioritizing jobs in this queue.
71//! type QueuePriority = u8;
72//!
73//! // start the JobQueue running.
74//! let mut job_queue = JobQueue::<QueuePriority>::start();
75//!
76//! let job = HelloJobAsync("world".to_string());
77//! let job_handle = job_queue.add_job_mut(job, 1)?;
78//!
79//! // await job to complete and obtain the (wrapped) job result
80//! let completion = job_handle.await?;
81//! let output = JobResultWrapper::<String>::try_from(completion)?.into_inner();
82//!
83//! assert_eq!("hello world", output);
84//!
85//! Ok(())
86//! }
87//! ```
88//!
89//! ## async vs blocking Job.
90//!
91//! To demonstrate the difference, we will implement an example job first as an
92//! async job and then as a blocking job.
93//!
94//! The example job finds all the prime numbers in a provided range.
95//!
96//! We create 100 jobs, each searching a range of 100 numbers. So the first
97//! 10000 integers are searched by all jobs.
98//!
99//! We use a `JobQueue<QueueJobPriority>` where `QueueJobPriority` is an enum we
100//! define that simply has `Low` and `High` variants.
101//!
102//! The jobs are added to the queue in ascending order but each is assigned a
103//! random job priority (either High or Low). High priority jobs will process
104//! first, thus queue-processing order will not match the order of adding jobs.
105//!
106//! In this example, job results are obtained by awaiting each JobHandle in the
107//! order it was added. Thus results are obtained in FIFO order despite the
108//! out-of-order processing.
109//!
110//! Alternatively a JoinSet or join_all() could be used to await all job-handles
111//! simultaneously and obtain results in queue-processing order as they complete.
112//!
113//! Likewise in an application with many concurrent tasks, each task might be
114//! submitting a job and immediately awaiting the JobHandle. In that scenario
115//! whichever task has submitted the highest priority job will obtain results
116//! first.
117//!
118//! ### Async Job considerations
119//!
120//! 1. the Job::is_async() impl returns true.
121//! 2. Job::run_async() or Job::run_async_cancellable() must be implemented.
122//!
123//! It is important the job yield regularly to the async runtime. Our
124//! processing is inherently blocking, so we accomplish this simply by making
125//! the is_prime() fn async, which is called in every loop iteration.
126//!
127//! ### Blocking job considerations
128//!
129//! 1. the Job::is_async() impl returns false.
130//! 2. Job::run() must be implemented.
131//! 3. it is necessary to regularly poll for a job-cancellation message in the
132//! job's main processing loop.
133//!
134//! ### Example
135//!
136//! ```
137//! use neptune_job_queue::JobCompletion;
138//! use neptune_job_queue::JobResultWrapper;
139//! use neptune_job_queue::JobQueue;
140//! use neptune_job_queue::channels::JobCancelReceiver;
141//! use neptune_job_queue::traits::*;
142//! use rand::Rng;
143//!
144//! // ### First lets define some common types ###
145//! // -------------------------------------------
146//!
147//! // define job priority levels for this job-queue.
148//! #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
149//! enum QueueJobPriority {
150//! Low = 1,
151//! High = 2,
152//! }
153//! impl QueueJobPriority {
154//! pub fn random() -> Self {
155//! let variants = [QueueJobPriority::Low, QueueJobPriority::High];
156//! variants[rand::rng().random_range(0..variants.len())]
157//! }
158//! }
159//!
160//! // define type alias for a wrapper around the data returned by our
161//! // job type. The wrapper is not required, but simplifies
162//! // conversions.
163//! type FindPrimesJobResult = JobResultWrapper<Vec<u64>>;
164//!
165//!
166//! // ### Now we define an async Job ###
167//! // ----------------------------------
168//!
169//! // define our custom job type that finds prime numbers within a range
170//! #[derive(Debug)]
171//! pub struct FindPrimesJobAsync {
172//! start: u64,
173//! len: u64,
174//! }
175//!
176//! // The prime-number finding algorithm can be described as:
177//! // Trial Division with Square Root Limit and 6k ± 1 Optimization
178//! //
179//! // we make the functions async because our "impl Job"
180//! // defines this as an async job and thus the runtime needs
181//! // some await points for cancellation and cooperating with
182//! // other async tasks.
183//! impl FindPrimesJobAsync {
184//! async fn is_prime(num: u64) -> bool {
185//! if num <= 1 {
186//! return false;
187//! }
188//! if num <= 3 {
189//! return true;
190//! }
191//! if num % 2 == 0 || num % 3 == 0 {
192//! return false;
193//! }
194//! let mut i = 5;
195//! while i * i <= num {
196//! if num % i == 0 || num % (i + 2) == 0 {
197//! return false;
198//! }
199//! i += 6;
200//! }
201//! true
202//! }
203//!
204//! async fn find_primes(&self) -> Vec<u64> {
205//! let mut primes = Vec::new();
206//! for num in self.start..=self.start + self.len {
207//! if Self::is_prime(num).await {
208//! primes.push(num);
209//! }
210//! }
211//!
212//! primes
213//! }
214//! }
215//!
216//! // implement Job trait.
217//! #[async_trait::async_trait]
218//! impl Job for FindPrimesJobAsync {
219//! // we are an async Job, so we must impl the run_async method
220//! fn is_async(&self) -> bool {
221//! true
222//! }
223//!
224//! async fn run_async(&self) -> Box<dyn JobResult> {
225//! let found_primes = self.find_primes().await;
226//! FindPrimesJobResult::new(found_primes).into()
227//! }
228//! }
229//!
230//! // ### Define an equivalent blocking job ###
231//! // -----------------------------------------
232//!
233//! // define our custom job type that finds prime numbers within a range
234//! #[derive(Debug)]
235//! pub struct FindPrimesJob {
236//! start: u64,
237//! len: u64,
238//! }
239//!
240//! // The prime-number finding algorithm can be described as:
241//! // Trial Division with Square Root Limit and 6k ± 1 Optimization
242//! //
243//! // None of the functions are async because our "impl Job"
244//! // defines this as blocking job. It will be run in tokio's blocking
245//! // threadpool via a spawn_blocking() call in the job-queue.
246//! impl FindPrimesJob {
247//! fn is_prime(num: u64) -> bool {
248//! if num <= 1 {
249//! return false;
250//! }
251//! if num <= 3 {
252//! return true;
253//! }
254//! if num % 2 == 0 || num % 3 == 0 {
255//! return false;
256//! }
257//! let mut i = 5;
258//! while i * i <= num {
259//! if num % i == 0 || num % (i + 2) == 0 {
260//! return false;
261//! }
262//! i += 6;
263//! }
264//! true
265//! }
266//! }
267//!
268//! // implement Job trait.
269//! #[async_trait::async_trait]
270//! impl Job for FindPrimesJob {
271//! // we are *not* an async Job.
272//! fn is_async(&self) -> bool {
273//! false
274//! }
275//!
276//! // as a blocking job we must impl the run() method
277//! fn run(&self, cancel_rx: JobCancelReceiver) -> JobCompletion {
278//! let mut primes = Vec::new();
279//!
280//! // this is the main processing loop of our job, so it should poll for
281//! // a cancellation message. It could be more efficient and poll
282//! // every 100 iterations or n milliseconds, etc.
283//! for num in self.start..=self.start + self.len {
284//!
285//! match cancel_rx.has_changed() {
286//! Ok(changed) if changed => return JobCompletion::Cancelled,
287//! Err(_) => return JobCompletion::Cancelled,
288//! _ => {}
289//! }
290//!
291//! if Self::is_prime(num) {
292//! primes.push(num);
293//! }
294//! }
295//!
296//! JobCompletion::Finished(FindPrimesJobResult::new(primes).into())
297//! }
298//! }
299//!
300//! // Now let's run these jobs.
301//! // -------------------------
302//!
303//! // Note that we will be mixing jobs of two different types.
304//! // Result processing is simplified because they both return the same
305//! // result type but that's not necessary.
306//!
307//! #[tokio::main]
308//! async fn main() -> anyhow::Result<()> {
309//! // setup
310//! const NUM_PRIMES_PER_JOB: u64 = 100;
311//! let mut job_handles = vec![];
312//!
313//! // start the JobQueue running.
314//! let mut job_queue = JobQueue::<QueueJobPriority>::start();
315//!
316//! // start 100 jobs, each searching 100 numbers for primes, with random job priorities
317//! // note that jobs begin processing right away while this loop is running.
318//! for n in 0..100 {
319//! let job_handle = if n % 2 == 0 {
320//! let job = FindPrimesJob {
321//! start: n * NUM_PRIMES_PER_JOB,
322//! len: NUM_PRIMES_PER_JOB,
323//! };
324//! job_queue.add_job_mut(job, QueueJobPriority::random())?
325//! } else {
326//! let job = FindPrimesJobAsync {
327//! start: n * NUM_PRIMES_PER_JOB,
328//! len: NUM_PRIMES_PER_JOB,
329//! };
330//! job_queue.add_job_mut(job, QueueJobPriority::random())?
331//! };
332//!
333//! job_handles.push(job_handle);
334//! }
335//!
336//! // await all the jobs to complete. note that:
337//! // 1. jobs will be processed in a different order than they were added due to the random priorities
338//! // 2. we are awaiting the job_handles in the order of adding, thus results are printed
339//! // sequentially from lowest primes to highest.
340//! // 3. if we moved the println!() inside FindPrimesJob::run_async() we would see the
341//! // order of processing, with prime ranges out-of-order.
342//! let mut max: u64 = 0;
343//! for job_handle in job_handles {
344//! let job_id = job_handle.job_id();
345//!
346//! // await job to complete and obtain the (wrapped) job result
347//! let completion = job_handle.await?;
348//! let found_primes = FindPrimesJobResult::try_from(completion)?.into_inner();
349//!
350//! // check for last (highest) prime in the result set
351//! if let Some(last_found) = found_primes.last() {
352//! // verify that max of each set is larger than previous set.
353//! // which indicates that job results are in same order as jobs were added.
354//! assert!(*last_found > max);
355//!
356//! max = std::cmp::max(max, *last_found);
357//! }
358//!
359//! println!(
360//! "job {} found {} primes: {:?}",
361//! job_id,
362//! found_primes.len(),
363//! found_primes
364//! );
365//! }
366//!
367//! // verify
368//! assert_eq!(9973, max); // 9973 is the largest prime number below 10000
369//!
370//! Ok(())
371//! }
372//! ```
373
374// please note that the job_queue module has zero neptune-core specific
375// code in it. It is intended/planned to move job_queue into its own
376// crate in the (near) future.
377
378pub mod channels;
379pub mod errors;
380mod job_completion;
381mod job_handle;
382mod job_id;
383mod job_result_wrapper;
384mod queue;
385pub mod traits;
386
387pub use job_completion::JobCompletion;
388pub use job_handle::JobHandle;
389pub use job_id::JobId;
390pub use job_result_wrapper::JobResultWrapper;
391pub use queue::JobQueue;