actix_threadpool/
lib.rs

1//! Thread pool for blocking operations
2
3use std::fmt;
4use std::future::Future;
5use std::pin::Pin;
6use std::task::{Context, Poll};
7
8use derive_more::Display;
9use futures_channel::oneshot;
10use parking_lot::Mutex;
11use threadpool::ThreadPool;
12
13/// Env variable for default cpu pool size.
14const ENV_CPU_POOL_VAR: &str = "ACTIX_THREADPOOL";
15
16lazy_static::lazy_static! {
17    pub(crate) static ref DEFAULT_POOL: Mutex<ThreadPool> = {
18        let num = std::env::var(ENV_CPU_POOL_VAR)
19            .map_err(|_| ())
20            .and_then(|val| {
21                val.parse().map_err(|_| log::warn!(
22                    "Can not parse {} value, using default",
23                    ENV_CPU_POOL_VAR,
24                ))
25            })
26            .unwrap_or_else(|_| num_cpus::get() * 5);
27        Mutex::new(
28            threadpool::Builder::new()
29                .thread_name("actix-web".to_owned())
30                .num_threads(num)
31                .build(),
32        )
33    };
34}
35
36thread_local! {
37    static POOL: ThreadPool = {
38        DEFAULT_POOL.lock().clone()
39    };
40}
41
42/// Blocking operation execution error
43#[derive(Debug, Display)]
44pub enum BlockingError<E: fmt::Debug> {
45    #[display(fmt = "{:?}", _0)]
46    Error(E),
47    #[display(fmt = "Thread pool is gone")]
48    Canceled,
49}
50
51impl<E: fmt::Debug> std::error::Error for BlockingError<E> {}
52
53/// Execute blocking function on a thread pool, returns future that resolves
54/// to result of the function execution.
55pub fn run<F, I, E>(f: F) -> CpuFuture<I, E>
56where
57    F: FnOnce() -> Result<I, E> + Send + 'static,
58    I: Send + 'static,
59    E: Send + fmt::Debug + 'static,
60{
61    let (tx, rx) = oneshot::channel();
62    POOL.with(|pool| {
63        pool.execute(move || {
64            if !tx.is_canceled() {
65                let _ = tx.send(f());
66            }
67        })
68    });
69
70    CpuFuture { rx }
71}
72
73/// Blocking operation completion future. It resolves with results
74/// of blocking function execution.
75pub struct CpuFuture<I, E> {
76    rx: oneshot::Receiver<Result<I, E>>,
77}
78
79impl<I, E: fmt::Debug> Future for CpuFuture<I, E> {
80    type Output = Result<I, BlockingError<E>>;
81
82    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
83        let rx = Pin::new(&mut self.rx);
84        let res = match rx.poll(cx) {
85            Poll::Pending => return Poll::Pending,
86            Poll::Ready(res) => res
87                .map_err(|_| BlockingError::Canceled)
88                .and_then(|res| res.map_err(BlockingError::Error)),
89        };
90        Poll::Ready(res)
91    }
92}
actix_threadpool/lib.rs

actix_threadpool/
lib.rs
