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 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204
use super::{Executor, Enter, SpawnError}; use futures::Future; use std::cell::Cell; /// Executes futures on the default executor for the current execution context. /// /// `DefaultExecutor` implements `Executor` and can be used to spawn futures /// without referencing a specific executor. /// /// When an executor starts, it sets the `DefaultExecutor` handle to point to an /// executor (usually itself) that is used to spawn new tasks. /// /// The current `DefaultExecutor` reference is tracked using a thread-local /// variable and is set using `tokio_executor::with_default` #[derive(Debug, Clone)] pub struct DefaultExecutor { _dummy: (), } impl DefaultExecutor { /// Returns a handle to the default executor for the current context. /// /// Futures may be spawned onto the default executor using this handle. /// /// The returned handle will reference whichever executor is configured as /// the default **at the time `spawn` is called**. This enables /// `DefaultExecutor::current()` to be called before an execution context is /// setup, then passed **into** an execution context before it is used. /// /// This is also true for sending the handle across threads, so calling /// `DefaultExecutor::current()` on thread A and then sending the result to /// thread B will _not_ reference the default executor that was set on thread A. pub fn current() -> DefaultExecutor { DefaultExecutor { _dummy: (), } } #[inline] fn with_current<F: FnOnce(&mut Executor) -> R, R>(f: F) -> Option<R> { EXECUTOR.with(|current_executor| { match current_executor.replace(State::Active) { State::Ready(executor_ptr) => { let executor = unsafe { &mut *executor_ptr }; let result = f(executor); current_executor.set(State::Ready(executor_ptr)); Some(result) }, State::Empty | State::Active => None, } }) } } #[derive(Clone, Copy)] enum State { // default executor not defined Empty, // default executor is defined and ready to be used Ready(*mut Executor), // default executor is currently active (used to detect recursive calls) Active } /// Thread-local tracking the current executor thread_local!(static EXECUTOR: Cell<State> = Cell::new(State::Empty)); // ===== impl DefaultExecutor ===== impl super::Executor for DefaultExecutor { fn spawn(&mut self, future: Box<Future<Item = (), Error = ()> + Send>) -> Result<(), SpawnError> { DefaultExecutor::with_current(|executor| executor.spawn(future)) .unwrap_or_else(|| Err(SpawnError::shutdown())) } fn status(&self) -> Result<(), SpawnError> { DefaultExecutor::with_current(|executor| executor.status()) .unwrap_or_else(|| Err(SpawnError::shutdown())) } } // ===== global spawn fns ===== /// Submits a future for execution on the default executor -- usually a /// threadpool. /// /// Futures are lazy constructs. When they are defined, no work happens. In /// order for the logic defined by the future to be run, the future must be /// spawned on an executor. This function is the easiest way to do so. /// /// This function must be called from an execution context, i.e. from a future /// that has been already spawned onto an executor. /// /// Once spawned, the future will execute. The details of how that happens is /// left up to the executor instance. If the executor is a thread pool, the /// future will be pushed onto a queue that a worker thread polls from. If the /// executor is a "current thread" executor, the future might be polled /// immediately from within the call to `spawn` or it might be pushed onto an /// internal queue. /// /// # Panics /// /// This function will panic if the default executor is not set or if spawning /// onto the default executor returns an error. To avoid the panic, use the /// `DefaultExecutor` handle directly. /// /// # Examples /// /// ```rust /// # extern crate futures; /// # extern crate tokio_executor; /// # use tokio_executor::spawn; /// # pub fn dox() { /// use futures::future::lazy; /// /// spawn(lazy(|| { /// println!("running on the default executor"); /// Ok(()) /// })); /// # } /// # pub fn main() {} /// ``` pub fn spawn<T>(future: T) where T: Future<Item = (), Error = ()> + Send + 'static, { DefaultExecutor::current().spawn(Box::new(future)) .unwrap() } /// Set the default executor for the duration of the closure /// /// # Panics /// /// This function panics if there already is a default executor set. pub fn with_default<T, F, R>(executor: &mut T, enter: &mut Enter, f: F) -> R where T: Executor, F: FnOnce(&mut Enter) -> R { EXECUTOR.with(|cell| { match cell.get() { State::Ready(_) | State::Active => panic!("default executor already set for execution context"), _ => {} } // Ensure that the executor is removed from the thread-local context // when leaving the scope. This handles cases that involve panicking. struct Reset<'a>(&'a Cell<State>); impl<'a> Drop for Reset<'a> { fn drop(&mut self) { self.0.set(State::Empty); } } let _reset = Reset(cell); // While scary, this is safe. The function takes a // `&mut Executor`, which guarantees that the reference lives for the // duration of `with_default`. // // Because we are always clearing the TLS value at the end of the // function, we can cast the reference to 'static which thread-local // cells require. let executor = unsafe { hide_lt(executor as &mut _ as *mut _) }; cell.set(State::Ready(executor)); f(enter) }) } unsafe fn hide_lt<'a>(p: *mut (Executor + 'a)) -> *mut (Executor + 'static) { use std::mem; mem::transmute(p) } #[cfg(test)] mod tests { use super::{Executor, DefaultExecutor, with_default}; #[test] fn default_executor_is_send_and_sync() { fn assert_send_sync<T: Send + Sync>() {} assert_send_sync::<DefaultExecutor>(); } #[test] fn nested_default_executor_status() { let mut enter = super::super::enter().unwrap(); let mut executor = DefaultExecutor::current(); let result = with_default(&mut executor, &mut enter, |_| { DefaultExecutor::current().status() }); assert!(result.err().unwrap().is_shutdown()) } }