pharos_runtime_core/
worker_pool.rs

1//! Bounded native worker-pool primitives for runtime request orchestration.
2//!
3//! The pool owns its channel and worker joins, rejects work after shutdown, and
4//! never uses an unbounded queue or an unsafe FFI bridge.
5
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::sync::{Arc, Mutex, mpsc};
8use std::thread::{self, JoinHandle};
9use std::time::Duration;
10
11const MAX_WORKERS: usize = 128;
12const MAX_QUEUE_CAPACITY: usize = 16_384;
13/// Approximate queue-memory accounting unit used for bounded work sizing.
14///
15/// # Examples
16///
17/// ```rust,ignore
18/// let _ = pharos_runtime_core::worker_pool::PHAROS_NATIVE_WORK_QUEUE_BYTES_PER_SLOT;
19/// ```
20///
21/// # Related
22///
23/// - [`crate::worker_pool`] for the surrounding module surface.
24pub const PHAROS_NATIVE_WORK_QUEUE_BYTES_PER_SLOT: usize = 4096;
25
26#[derive(Debug, Clone, Copy, Eq, PartialEq)]
27/// Errors returned by bounded worker-pool lifecycle operations.
28///
29/// # Variants
30///
31/// - `InvalidConfiguration`: Supported `WorkerPoolError` variant.
32/// - `Shutdown`: Supported `WorkerPoolError` variant.
33/// - `QueueFull`: Supported `WorkerPoolError` variant.
34/// - `Poisoned`: Supported `WorkerPoolError` variant.
35/// - `WorkerPanicked`: Supported `WorkerPoolError` variant.
36///
37/// # Examples
38///
39/// ```rust,ignore
40/// let _ = pharos_runtime_core::worker_pool::WorkerPoolError::InvalidConfiguration;
41/// ```
42///
43/// # Related
44///
45/// - [`crate::worker_pool`] for constructors, helpers, and neighboring runtime types.
46pub enum WorkerPoolError {
47    /// The requested size is outside the safe configured range.
48    InvalidConfiguration,
49    /// The queue has been closed or shutdown has been requested.
50    Shutdown,
51    /// The bounded queue is full and the caller must apply overload behavior.
52    QueueFull,
53    /// A lifecycle mutex was poisoned by a worker or caller panic.
54    Poisoned,
55    /// A worker panicked while joining.
56    WorkerPanicked,
57}
58
59/// A bounded worker pool that executes an owned callback for each item.
60///
61/// # Fields
62///
63/// - This type stores module-owned state for [`crate::worker_pool::WorkerPool`].
64///
65/// # Examples
66///
67/// ```rust,ignore
68/// let _ = std::mem::size_of::<pharos_runtime_core::worker_pool::WorkerPool>();
69/// ```
70///
71/// # Related
72///
73/// - [`crate::worker_pool`] for constructors, helpers, and neighboring runtime types.
74pub struct WorkerPool<T: Send + 'static> {
75    sender: Mutex<Option<mpsc::SyncSender<T>>>,
76    shutdown: Arc<AtomicBool>,
77    workers: Mutex<Vec<JoinHandle<()>>>,
78}
79
80impl<T: Send + 'static> WorkerPool<T> {
81    /// Creates a pool with a bounded queue and a shared item handler.
82    ///
83    /// # Parameters
84    ///
85    /// - `worker_count`: Input consumed by [`crate::worker_pool::new`].
86    /// - `queue_capacity`: Input consumed by [`crate::worker_pool::new`].
87    /// - `handler`: Input consumed by [`crate::worker_pool::new`].
88    ///
89    /// # Returns
90    ///
91    /// Returns the successful value produced by [`crate::worker_pool::new`] or an error when the operation fails.
92    ///
93    /// # Errors
94    ///
95    /// Returns an error value when [`crate::worker_pool::new`] cannot complete successfully.
96    ///
97    /// # Performance
98    ///
99    /// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::worker_pool::new`].
100    ///
101    /// # Examples
102    ///
103    /// ```rust,ignore
104    /// let _result = pharos_runtime_core::worker_pool::new(todo!(), todo!(), todo!());
105    /// ```
106    ///
107    /// # Related
108    ///
109    /// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
110    pub fn new(
111        worker_count: usize,
112        queue_capacity: usize,
113        handler: Arc<dyn Fn(T) + Send + Sync + 'static>,
114    ) -> Result<Self, WorkerPoolError> {
115        if worker_count == 0
116            || worker_count > MAX_WORKERS
117            || queue_capacity == 0
118            || queue_capacity > MAX_QUEUE_CAPACITY
119        {
120            return Err(WorkerPoolError::InvalidConfiguration);
121        }
122        let (sender, receiver) = mpsc::sync_channel(queue_capacity);
123        let receiver = Arc::new(Mutex::new(receiver));
124        let shutdown = Arc::new(AtomicBool::new(false));
125        let mut workers = Vec::with_capacity(worker_count);
126        for index in 0..worker_count {
127            let receiver = Arc::clone(&receiver);
128            let shutdown = Arc::clone(&shutdown);
129            let handler = Arc::clone(&handler);
130            let worker = thread::Builder::new()
131                .name(format!("pharos-worker-{index}"))
132                .spawn(move || {
133                    loop {
134                        let item = match receiver.lock() {
135                            Ok(lock) => lock.recv_timeout(Duration::from_millis(100)),
136                            Err(_) => break,
137                        };
138                        match item {
139                            Ok(item) => handler(item),
140                            Err(mpsc::RecvTimeoutError::Timeout)
141                                if shutdown.load(Ordering::Acquire) =>
142                            {
143                                break;
144                            }
145                            Err(mpsc::RecvTimeoutError::Timeout) => continue,
146                            Err(mpsc::RecvTimeoutError::Disconnected) => break,
147                        }
148                    }
149                })
150                .map_err(|_| WorkerPoolError::Shutdown)?;
151            workers.push(worker);
152        }
153        Ok(Self {
154            sender: Mutex::new(Some(sender)),
155            shutdown,
156            workers: Mutex::new(workers),
157        })
158    }
159
160    /// Enqueues one item without allowing the queue to grow without bound.
161    ///
162    /// # Parameters
163    ///
164    /// - `item`: Input consumed by [`crate::worker_pool::enqueue`].
165    ///
166    /// # Returns
167    ///
168    /// Returns the successful value produced by [`crate::worker_pool::enqueue`] or an error when the operation fails.
169    ///
170    /// # Errors
171    ///
172    /// Returns an error value when [`crate::worker_pool::enqueue`] cannot complete successfully.
173    ///
174    /// # Examples
175    ///
176    /// ```rust,ignore
177    /// let _result = pharos_runtime_core::worker_pool::enqueue(todo!());
178    /// ```
179    ///
180    /// # Related
181    ///
182    /// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
183    pub fn enqueue(&self, item: T) -> Result<(), WorkerPoolError> {
184        self.enqueue_with_rejection(item, |_| {})
185    }
186
187    /// Enqueues one item and returns ownership to a bounded overload handler when full.
188    ///
189    /// # Parameters
190    ///
191    /// - `item`: Input consumed by [`crate::worker_pool::enqueue_with_rejection`].
192    /// - `reject`: Input consumed by [`crate::worker_pool::enqueue_with_rejection`].
193    ///
194    /// # Returns
195    ///
196    /// Returns the successful value produced by [`crate::worker_pool::enqueue_with_rejection`] or an error when the operation fails.
197    ///
198    /// # Errors
199    ///
200    /// Returns an error value when [`crate::worker_pool::enqueue_with_rejection`] cannot complete successfully.
201    ///
202    /// # Examples
203    ///
204    /// ```rust,ignore
205    /// let _result = pharos_runtime_core::worker_pool::enqueue_with_rejection(todo!(), todo!());
206    /// ```
207    ///
208    /// # Related
209    ///
210    /// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
211    pub fn enqueue_with_rejection<F>(&self, item: T, reject: F) -> Result<(), WorkerPoolError>
212    where
213        F: FnOnce(T),
214    {
215        if self.shutdown.load(Ordering::Acquire) {
216            reject(item);
217            return Err(WorkerPoolError::Shutdown);
218        }
219        let sender = self
220            .sender
221            .lock()
222            .map_err(|_| WorkerPoolError::Poisoned)?
223            .clone()
224            .ok_or(WorkerPoolError::Shutdown)?;
225        match sender.try_send(item) {
226            Ok(()) => Ok(()),
227            Err(mpsc::TrySendError::Full(item)) => {
228                reject(item);
229                Err(WorkerPoolError::QueueFull)
230            }
231            Err(mpsc::TrySendError::Disconnected(item)) => {
232                reject(item);
233                Err(WorkerPoolError::Shutdown)
234            }
235        }
236    }
237
238    /// Requests shutdown and closes the queue to new work.
239    ///
240    /// # Returns
241    ///
242    /// Returns the successful value produced by [`crate::worker_pool::request_shutdown`] or an error when the operation fails.
243    ///
244    /// # Errors
245    ///
246    /// Returns an error value when [`crate::worker_pool::request_shutdown`] cannot complete successfully.
247    ///
248    /// # Examples
249    ///
250    /// ```rust,ignore
251    /// let _result = pharos_runtime_core::worker_pool::request_shutdown();
252    /// ```
253    ///
254    /// # Related
255    ///
256    /// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
257    pub fn request_shutdown(&self) -> Result<(), WorkerPoolError> {
258        self.shutdown.store(true, Ordering::Release);
259        let mut sender = self.sender.lock().map_err(|_| WorkerPoolError::Poisoned)?;
260        sender.take();
261        Ok(())
262    }
263
264    /// Joins all workers after shutdown has been requested.
265    ///
266    /// # Returns
267    ///
268    /// Returns the successful value produced by [`crate::worker_pool::join`] or an error when the operation fails.
269    ///
270    /// # Errors
271    ///
272    /// Returns an error value when [`crate::worker_pool::join`] cannot complete successfully.
273    ///
274    /// # Performance
275    ///
276    /// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::worker_pool::join`].
277    ///
278    /// # Examples
279    ///
280    /// ```rust,ignore
281    /// let _result = pharos_runtime_core::worker_pool::join();
282    /// ```
283    ///
284    /// # Related
285    ///
286    /// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
287    pub fn join(&self) -> Result<(), WorkerPoolError> {
288        self.request_shutdown()?;
289        let mut workers = self.workers.lock().map_err(|_| WorkerPoolError::Poisoned)?;
290        let handles = std::mem::take(&mut *workers);
291        for handle in handles {
292            if handle.join().is_err() {
293                return Err(WorkerPoolError::WorkerPanicked);
294            }
295        }
296        Ok(())
297    }
298
299    /// Shuts down and joins the pool, releasing all worker handles.
300    ///
301    /// # Returns
302    ///
303    /// Returns the successful value produced by [`crate::worker_pool::destroy`] or an error when the operation fails.
304    ///
305    /// # Errors
306    ///
307    /// Returns an error value when [`crate::worker_pool::destroy`] cannot complete successfully.
308    ///
309    /// # Examples
310    ///
311    /// ```rust,ignore
312    /// let _result = pharos_runtime_core::worker_pool::destroy();
313    /// ```
314    ///
315    /// # Related
316    ///
317    /// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
318    pub fn destroy(self) -> Result<(), WorkerPoolError> {
319        self.join()
320    }
321}
322
323impl<T: Send + 'static> Drop for WorkerPool<T> {
324    /// Performs drop.
325    fn drop(&mut self) {
326        self.shutdown.store(true, Ordering::Release);
327        if let Ok(mut sender) = self.sender.lock() {
328            sender.take();
329        }
330        if let Ok(mut workers) = self.workers.lock() {
331            for handle in std::mem::take(&mut *workers) {
332                let _ = handle.join();
333            }
334        }
335    }
336}
337
338/// Executes one worker thread main-loop with bounded recv polling and cooperative shutdown.
339///
340/// # Parameters
341///
342/// - `receiver`: Input consumed by [`crate::worker_pool::worker_pool_thread_main`].
343/// - `shutdown`: Input consumed by [`crate::worker_pool::worker_pool_thread_main`].
344/// - `handler`: Input consumed by [`crate::worker_pool::worker_pool_thread_main`].
345///
346/// # Returns
347///
348/// Returns after [`crate::worker_pool::worker_pool_thread_main`] applies its in-place effect.
349///
350/// # Examples
351///
352/// ```rust,ignore
353/// let _result = pharos_runtime_core::worker_pool::worker_pool_thread_main(todo!(), todo!(), todo!());
354/// ```
355///
356/// # Related
357///
358/// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
359pub fn worker_pool_thread_main<T: Send + 'static>(
360    receiver: Arc<Mutex<mpsc::Receiver<T>>>,
361    shutdown: Arc<AtomicBool>,
362    handler: Arc<dyn Fn(T) + Send + Sync + 'static>,
363) {
364    loop {
365        let item = match receiver.lock() {
366            Ok(lock) => lock.recv_timeout(Duration::from_millis(100)),
367            Err(_) => break,
368        };
369        match item {
370            Ok(item) => handler(item),
371            Err(mpsc::RecvTimeoutError::Timeout) if shutdown.load(Ordering::Acquire) => break,
372            Err(mpsc::RecvTimeoutError::Timeout) => continue,
373            Err(mpsc::RecvTimeoutError::Disconnected) => break,
374        }
375    }
376}
377
378/// Creates the native string-work worker pool with bounded resources.
379///
380/// # Parameters
381///
382/// - `worker_count`: Input consumed by [`crate::worker_pool::worker_pool_init`].
383/// - `queue_capacity`: Input consumed by [`crate::worker_pool::worker_pool_init`].
384/// - `handler`: Input consumed by [`crate::worker_pool::worker_pool_init`].
385///
386/// # Returns
387///
388/// Returns the successful value produced by [`crate::worker_pool::worker_pool_init`] or an error when the operation fails.
389///
390/// # Errors
391///
392/// Returns an error value when [`crate::worker_pool::worker_pool_init`] cannot complete successfully.
393///
394/// # Examples
395///
396/// ```rust,ignore
397/// let _result = pharos_runtime_core::worker_pool::worker_pool_init(todo!(), todo!(), todo!());
398/// ```
399///
400/// # Related
401///
402/// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
403pub fn worker_pool_init(
404    worker_count: usize,
405    queue_capacity: usize,
406    handler: Arc<dyn Fn(String) + Send + Sync + 'static>,
407) -> Result<WorkerPool<String>, WorkerPoolError> {
408    WorkerPool::new(worker_count, queue_capacity, handler)
409}
410
411/// Enqueues one string request and fails when the bounded queue is closed.
412///
413/// # Parameters
414///
415/// - `pool`: Input consumed by [`crate::worker_pool::worker_pool_enqueue`].
416/// - `item`: Input consumed by [`crate::worker_pool::worker_pool_enqueue`].
417///
418/// # Returns
419///
420/// Returns the successful value produced by [`crate::worker_pool::worker_pool_enqueue`] or an error when the operation fails.
421///
422/// # Errors
423///
424/// Returns an error value when [`crate::worker_pool::worker_pool_enqueue`] cannot complete successfully.
425///
426/// # Examples
427///
428/// ```rust,ignore
429/// let _result = pharos_runtime_core::worker_pool::worker_pool_enqueue(todo!(), todo!());
430/// ```
431///
432/// # Related
433///
434/// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
435pub fn worker_pool_enqueue(pool: &WorkerPool<String>, item: String) -> Result<(), WorkerPoolError> {
436    pool.enqueue(item)
437}
438
439/// Requests shutdown of a string worker pool.
440///
441/// # Parameters
442///
443/// - `pool`: Input consumed by [`crate::worker_pool::worker_pool_request_shutdown`].
444///
445/// # Returns
446///
447/// Returns the successful value produced by [`crate::worker_pool::worker_pool_request_shutdown`] or an error when the operation fails.
448///
449/// # Errors
450///
451/// Returns an error value when [`crate::worker_pool::worker_pool_request_shutdown`] cannot complete successfully.
452///
453/// # Examples
454///
455/// ```rust,ignore
456/// let _result = pharos_runtime_core::worker_pool::worker_pool_request_shutdown(todo!());
457/// ```
458///
459/// # Related
460///
461/// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
462pub fn worker_pool_request_shutdown(pool: &WorkerPool<String>) -> Result<(), WorkerPoolError> {
463    pool.request_shutdown()
464}
465
466/// Joins every worker after shutdown.
467///
468/// # Parameters
469///
470/// - `pool`: Input consumed by [`crate::worker_pool::worker_pool_join`].
471///
472/// # Returns
473///
474/// Returns the successful value produced by [`crate::worker_pool::worker_pool_join`] or an error when the operation fails.
475///
476/// # Errors
477///
478/// Returns an error value when [`crate::worker_pool::worker_pool_join`] cannot complete successfully.
479///
480/// # Examples
481///
482/// ```rust,ignore
483/// let _result = pharos_runtime_core::worker_pool::worker_pool_join(todo!());
484/// ```
485///
486/// # Related
487///
488/// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
489pub fn worker_pool_join(pool: &WorkerPool<String>) -> Result<(), WorkerPoolError> {
490    pool.join()
491}
492
493/// Destroys a string worker pool after joining its workers.
494///
495/// # Parameters
496///
497/// - `pool`: Input consumed by [`crate::worker_pool::worker_pool_destroy`].
498///
499/// # Returns
500///
501/// Returns the successful value produced by [`crate::worker_pool::worker_pool_destroy`] or an error when the operation fails.
502///
503/// # Errors
504///
505/// Returns an error value when [`crate::worker_pool::worker_pool_destroy`] cannot complete successfully.
506///
507/// # Examples
508///
509/// ```rust,ignore
510/// let _result = pharos_runtime_core::worker_pool::worker_pool_destroy(todo!());
511/// ```
512///
513/// # Related
514///
515/// - [`crate::worker_pool`] for adjacent helpers that share the same runtime surface.
516pub fn worker_pool_destroy(pool: WorkerPool<String>) -> Result<(), WorkerPoolError> {
517    pool.destroy()
518}
519
520#[cfg(test)]
521mod tests {
522    use super::{WorkerPool, WorkerPoolError};
523    use std::sync::atomic::{AtomicUsize, Ordering};
524    use std::sync::{Arc, Barrier};
525
526    /// Confirms a saturated pool rejects immediately and returns the item to the overload handler.
527    #[test]
528    fn rejects_work_when_the_bounded_queue_is_full() {
529        let entered = Arc::new(Barrier::new(2));
530        let release = Arc::new(Barrier::new(2));
531        let worker_entered = Arc::clone(&entered);
532        let worker_release = Arc::clone(&release);
533        let handled = Arc::new(AtomicUsize::new(0));
534        let worker_handled = Arc::clone(&handled);
535        let pool = WorkerPool::new(
536            1,
537            1,
538            Arc::new(move |_item: usize| {
539                if worker_handled.fetch_add(1, Ordering::SeqCst) == 0 {
540                    worker_entered.wait();
541                    worker_release.wait();
542                }
543            }),
544        )
545        .expect("create worker pool");
546        pool.enqueue(1).expect("enqueue active work");
547        entered.wait();
548        pool.enqueue(2).expect("fill queue");
549        let mut rejected = None;
550        let result = pool.enqueue_with_rejection(3, |item| rejected = Some(item));
551        assert_eq!(result, Err(WorkerPoolError::QueueFull));
552        assert_eq!(rejected, Some(3));
553        release.wait();
554        pool.destroy().expect("drain worker pool");
555    }
556
557    /// Confirms one slow task does not block an independent worker beyond the latency budget.
558    #[test]
559    fn serves_fast_work_while_another_worker_is_slow() {
560        let (completed, receiver) = std::sync::mpsc::channel();
561        let pool = WorkerPool::new(
562            2,
563            2,
564            Arc::new(move |slow: bool| {
565                if slow {
566                    std::thread::sleep(std::time::Duration::from_millis(200));
567                } else {
568                    let _ = completed.send(());
569                }
570            }),
571        )
572        .expect("create concurrent worker pool");
573        let started = std::time::Instant::now();
574        pool.enqueue(true).expect("enqueue slow work");
575        pool.enqueue(false).expect("enqueue fast work");
576        receiver
577            .recv_timeout(std::time::Duration::from_millis(100))
578            .expect("fast work latency budget");
579        assert!(started.elapsed() < std::time::Duration::from_millis(150));
580        pool.destroy().expect("drain concurrent worker pool");
581    }
582}
pharos_runtime_core/worker_pool.rs

pharos_runtime_core/
worker_pool.rs
