actori_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 = "actori_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("actori-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
51/// Execute blocking function on a thread pool, returns future that resolves
52/// to result of the function execution.
53pub fn run<F, I, E>(f: F) -> CpuFuture<I, E>
54where
55    F: FnOnce() -> Result<I, E> + Send + 'static,
56    I: Send + 'static,
57    E: Send + fmt::Debug + 'static,
58{
59    let (tx, rx) = oneshot::channel();
60    POOL.with(|pool| {
61        pool.execute(move || {
62            if !tx.is_canceled() {
63                let _ = tx.send(f());
64            }
65        })
66    });
67
68    CpuFuture { rx }
69}
70
71/// Blocking operation completion future. It resolves with results
72/// of blocking function execution.
73pub struct CpuFuture<I, E> {
74    rx: oneshot::Receiver<Result<I, E>>,
75}
76
77impl<I, E: fmt::Debug> Future for CpuFuture<I, E> {
78    type Output = Result<I, BlockingError<E>>;
79
80    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
81        let rx = Pin::new(&mut self.rx);
82        let res = match rx.poll(cx) {
83            Poll::Pending => return Poll::Pending,
84            Poll::Ready(res) => res
85                .map_err(|_| BlockingError::Canceled)
86                .and_then(|res| res.map_err(BlockingError::Error)),
87        };
88        Poll::Ready(res)
89    }
90}
actori_threadpool/lib.rs

actori_threadpool/
lib.rs
