karyon_core/async_runtime/
executor.rs

1use std::{future::Future, panic::catch_unwind, sync::Arc, thread};
2
3use once_cell::sync::OnceCell;
4
5#[cfg(feature = "smol")]
6pub use smol::Executor as SmolEx;
7
8#[cfg(feature = "tokio")]
9pub use tokio::runtime::Runtime;
10
11use super::Task;
12
13/// A handle to a multi-threaded async executor.
14///
15/// karyon does not run the executor for you. The caller owns the runtime and
16/// is responsible for driving it. On `tokio` this is just a `Runtime`. On
17/// `smol` you must spin up worker threads yourself, e.g. via
18/// [`easy_parallel`](https://docs.rs/easy-parallel):
19///
20/// ```ignore
21/// use std::sync::Arc;
22/// use async_channel::unbounded;
23/// use easy_parallel::Parallel;
24/// use smol::{future, Executor as SmolEx};
25///
26/// let ex = Arc::new(SmolEx::new());
27/// let (signal, shutdown) = unbounded::<()>();
28///
29/// let num_threads = std::thread::available_parallelism()
30///     .map(|n| n.get())
31///     .unwrap_or(1);
32///
33/// Parallel::new()
34///     .each(0..num_threads, |_| future::block_on(ex.run(shutdown.recv())))
35///     .finish(|| future::block_on(async {
36///         // your async main here
37///         drop(signal);
38///     }));
39/// ```
40///
41/// Pass `ex.clone().into()` to karyon APIs that take an [`Executor`].
42#[derive(Clone)]
43pub struct Executor {
44    #[cfg(feature = "smol")]
45    inner: Arc<SmolEx<'static>>,
46    #[cfg(feature = "tokio")]
47    inner: Arc<Runtime>,
48}
49
50impl Executor {
51    pub fn spawn<T: Send + 'static>(
52        &self,
53        future: impl Future<Output = T> + Send + 'static,
54    ) -> Task<T> {
55        self.inner.spawn(future).into()
56    }
57
58    #[cfg(feature = "tokio")]
59    pub fn handle(&self) -> &tokio::runtime::Handle {
60        self.inner.handle()
61    }
62}
63
64static GLOBAL_EXECUTOR: OnceCell<Executor> = OnceCell::new();
65
66/// Returns the process-wide global executor, driven on its own thread.
67/// Single-threaded on `smol`, multi-threaded on `tokio`.
68pub fn global_executor() -> Executor {
69    #[cfg(feature = "smol")]
70    fn init_executor() -> Executor {
71        let ex = smol::Executor::new();
72        thread::Builder::new()
73            .name("smol-executor".to_string())
74            .spawn(|| loop {
75                catch_unwind(|| {
76                    smol::block_on(global_executor().inner.run(std::future::pending::<()>()))
77                })
78                .ok();
79            })
80            .expect("cannot spawn executor thread");
81        // Prevent spawning another thread by running the process driver on this
82        // thread. see https://github.com/smol-rs/smol/blob/master/src/spawn.rs
83        ex.spawn(async_process::driver()).detach();
84        Executor {
85            inner: Arc::new(ex),
86        }
87    }
88
89    #[cfg(feature = "tokio")]
90    fn init_executor() -> Executor {
91        let ex = Arc::new(tokio::runtime::Runtime::new().expect("cannot build tokio runtime"));
92        thread::Builder::new()
93            .name("tokio-executor".to_string())
94            .spawn({
95                let ex = ex.clone();
96                move || {
97                    catch_unwind(|| ex.block_on(std::future::pending::<()>())).ok();
98                }
99            })
100            .expect("cannot spawn tokio runtime thread");
101        Executor { inner: ex }
102    }
103
104    GLOBAL_EXECUTOR.get_or_init(init_executor).clone()
105}
106
107#[cfg(feature = "smol")]
108impl From<Arc<smol::Executor<'static>>> for Executor {
109    fn from(ex: Arc<smol::Executor<'static>>) -> Executor {
110        Executor { inner: ex }
111    }
112}
113
114#[cfg(feature = "tokio")]
115impl From<Arc<tokio::runtime::Runtime>> for Executor {
116    fn from(rt: Arc<tokio::runtime::Runtime>) -> Executor {
117        Executor { inner: rt }
118    }
119}
120
121#[cfg(feature = "smol")]
122impl From<smol::Executor<'static>> for Executor {
123    fn from(ex: smol::Executor<'static>) -> Executor {
124        Executor {
125            inner: Arc::new(ex),
126        }
127    }
128}
129
130#[cfg(feature = "tokio")]
131impl From<tokio::runtime::Runtime> for Executor {
132    fn from(rt: tokio::runtime::Runtime) -> Executor {
133        Executor {
134            inner: Arc::new(rt),
135        }
136    }
137}
karyon_core/async_runtime/executor.rs

karyon_core/async_runtime/
executor.rs
