1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

use super::{Executor, Idle, Inner, Opaque, Spawner};
use crate::{ParallelSend, Runtime};
use std::future::Future;

/// The trait to spawn execution of pending jobs on async runtime
#[cfg_attr(feature = "doc-cfg", doc(cfg(feature = "futures")))]
pub trait ExecutorSpawner: Sized {
    /// The type of join handle which returns `()`
    type JoinHandle;

    /// Spawn pending jobs using async runtime spawn function
    fn spawn_executor(self, task: Executor) -> Self::JoinHandle;
}

macro_rules! async_rt_impl {
    ($($(#[$meta:meta])* $type:ident { $join_handle:ty, $spawn_local:path, $spawn:path })*) => {
        $(
            $(#[$meta])*
            pub struct $type;

            $(#[$meta])*
            impl ExecutorSpawner for $type {
                type JoinHandle = $join_handle;

                fn spawn_executor(
                    self,
                    task: Executor,
                ) -> Self::JoinHandle {
                    #[cfg(not(feature = "parallel"))]
                    use $spawn_local as spawn_parallel;
                    #[cfg(feature = "parallel")]
                    use $spawn as spawn_parallel;

                    spawn_parallel(task)
                }
            }
        )*
    };
}

async_rt_impl! {
    /// The [`tokio`] async runtime for spawning executors.
    #[cfg(feature = "tokio")]
    #[cfg_attr(feature = "doc-cfg", doc(cfg(feature = "tokio")))]
    Tokio { tokio::task::JoinHandle<()>, tokio::task::spawn_local, tokio::task::spawn }

    /// The [`async_std`] runtime for spawning executors.
    #[cfg(feature = "async-std")]
    #[cfg_attr(feature = "doc-cfg", doc(cfg(feature = "async-std")))]
    AsyncStd { async_std::task::JoinHandle<()>, async_std::task::spawn_local, async_std::task::spawn }

    /// The [`smol`] async runtime for spawning executors.
    #[cfg(all(feature = "smol", feature = "parallel"))]
    #[cfg_attr(feature = "doc-cfg", doc(cfg(all(feature = "smol", feature = "parallel"))))]
    Smol { smol::Task<()>, smol::spawn_local, smol::spawn }
}

#[cfg(all(feature = "smol", feature = "parallel"))]
use smol::Executor as SmolExecutor;
#[cfg(all(feature = "smol", not(feature = "parallel")))]
use smol::LocalExecutor as SmolExecutor;

#[cfg(feature = "smol")]
impl<'a> ExecutorSpawner for &SmolExecutor<'a> {
    type JoinHandle = smol::Task<()>;

    fn spawn_executor(self, task: Executor) -> Self::JoinHandle {
        self.spawn(task)
    }
}

impl Inner {
    pub fn has_spawner(&self) -> bool {
        unsafe { self.get_opaque() }.spawner.is_some()
    }
}

impl Opaque {
    pub fn get_spawner(&self) -> &Spawner {
        self.spawner
            .as_ref()
            .expect("Async executor is not initialized for the Runtime. Possibly missing call `Runtime::run_executor()` or `Runtime::spawn_executor()`")
    }
}

impl Runtime {
    fn get_spawner(&self) -> &Spawner {
        let inner = self.inner.lock();
        let opaque = unsafe { &*(inner.get_opaque() as *const Opaque) };
        opaque.get_spawner()
    }

    /// Await until all pending jobs and spawned futures will be done
    #[cfg_attr(feature = "doc-cfg", doc(cfg(feature = "futures")))]
    #[inline(always)]
    pub fn idle(&self) -> Idle {
        self.get_spawner().idle()
    }

    /// Run pending jobs and futures executor
    #[cfg_attr(feature = "doc-cfg", doc(cfg(feature = "futures")))]
    #[inline(always)]
    pub fn run_executor(&self) -> Executor {
        let mut inner = self.inner.lock();
        let opaque = unsafe { &mut *(inner.get_opaque_mut() as *mut Opaque) };
        assert!(
            opaque.spawner.is_none(),
            "Async executor already initialized for the Runtime."
        );
        let (executor, spawner) = Executor::new();
        opaque.spawner = Some(spawner);
        executor
    }

    /// Spawn pending jobs and futures executor
    #[cfg_attr(feature = "doc-cfg", doc(cfg(feature = "futures")))]
    #[inline(always)]
    pub fn spawn_executor<A: ExecutorSpawner>(&self, spawner: A) -> A::JoinHandle {
        spawner.spawn_executor(self.run_executor())
    }

    pub(crate) fn spawn_pending_jobs(&self) {
        let runtime = self.clone();
        self.spawn(async move { runtime.execute_pending_jobs().await });
    }

    async fn execute_pending_jobs(&self) {
        loop {
            match self.execute_pending_job() {
                // No tasks in queue
                Ok(false) => break,
                // Task was executed successfully
                Ok(true) => (),
                // Task was failed with exception
                Err(error) => {
                    eprintln!("Error when pending job executing: {}", error);
                }
            }
            futures_lite::future::yield_now().await;
        }
    }

    /// Spawn future using runtime
    #[cfg_attr(feature = "doc-cfg", doc(cfg(feature = "futures")))]
    pub fn spawn<F, T>(&self, future: F)
    where
        F: Future<Output = T> + ParallelSend + 'static,
        T: ParallelSend + 'static,
    {
        self.get_spawner().spawn(future);
    }
}