syncweb_core/daemon/
pool.rs

1use std::fmt;
2
3use crate::error::{Result, SyncwebError};
4
5/// A fixed-size Rayon pool owned by a daemon for CPU-bound archive work.
6pub struct ManagedPool {
7    pool: rayon::ThreadPool,
8    name: String,
9    thread_count: usize,
10}
11
12impl ManagedPool {
13    /// Create a named pool with a fixed number of worker threads.
14    ///
15    /// A thread count of zero uses the host's available parallelism.
16    ///
17    /// # Errors
18    ///
19    /// Returns an error when Rayon cannot create the pool.
20    pub fn new(name: impl Into<String>, thread_count: usize) -> Result<Self> {
21        let pool_name = name.into();
22        let normalized_threads = if thread_count == 0 {
23            std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get)
24        } else {
25            thread_count
26        };
27        let thread_name = pool_name.clone();
28        let pool = rayon::ThreadPoolBuilder::new()
29            .num_threads(normalized_threads)
30            .thread_name(move |index| format!("{thread_name}-{index}"))
31            .build()
32            .map_err(|error| SyncwebError::operation("failed to create managed thread pool", error))?;
33        Ok(Self {
34            pool,
35            name: pool_name,
36            thread_count: normalized_threads,
37        })
38    }
39
40    /// Run a synchronous operation on this pool and return its result.
41    pub fn install<F, R>(&self, operation: F) -> R
42    where
43        F: FnOnce() -> R + Send,
44        R: Send,
45    {
46        self.pool.install(operation)
47    }
48
49    /// Queue a task in FIFO order on this pool.
50    pub fn spawn_fifo<F>(&self, operation: F)
51    where
52        F: FnOnce() + Send + 'static,
53    {
54        self.pool.spawn_fifo(operation);
55    }
56
57    /// Return the fixed number of worker threads in this pool.
58    #[must_use]
59    pub const fn thread_count(&self) -> usize {
60        self.thread_count
61    }
62}
63
64impl fmt::Debug for ManagedPool {
65    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
66        formatter
67            .debug_struct("ManagedPool")
68            .field("name", &self.name)
69            .field("thread_count", &self.thread_count)
70            .finish_non_exhaustive()
71    }
72}
73
74#[cfg(test)]
75mod tests {
76    use std::{sync::mpsc, thread, time::Duration};
77
78    use super::ManagedPool;
79
80    #[test]
81    fn custom_thread_count_is_fixed() -> crate::error::Result<()> {
82        let pool = ManagedPool::new("test-archive", 2)?;
83        if pool.thread_count() != 2 {
84            return Err(crate::error::SyncwebError::InvalidConfig(
85                "managed pool did not retain its configured thread count".to_owned(),
86            ));
87        }
88        Ok(())
89    }
90
91    #[test]
92    fn install_runs_on_named_pool_thread() -> crate::error::Result<()> {
93        let pool = ManagedPool::new("test-archive", 1)?;
94        let worker_name_result = pool.install(|| thread::current().name().map(str::to_owned));
95        let worker_name = worker_name_result
96            .ok_or_else(|| crate::error::SyncwebError::InvalidConfig("managed pool thread has no name".to_owned()))?;
97        if !worker_name.starts_with("test-archive-") {
98            return Err(crate::error::SyncwebError::InvalidConfig(
99                "managed pool worker name has an unexpected prefix".to_owned(),
100            ));
101        }
102        Ok(())
103    }
104
105    #[test]
106    fn spawn_fifo_executes() -> crate::error::Result<()> {
107        let pool = ManagedPool::new("test-archive", 1)?;
108        let (sender, receiver) = mpsc::channel();
109        pool.spawn_fifo(move || {
110            let _ = sender.send(());
111        });
112        receiver
113            .recv_timeout(Duration::from_secs(1))
114            .map_err(|error| crate::error::SyncwebError::operation("managed pool task did not run", error))?;
115        Ok(())
116    }
117}
syncweb_core/daemon/pool.rs

syncweb_core/daemon/
pool.rs
