tokio/runtime/local_runtime/runtime.rs
1#![allow(irrefutable_let_patterns)]
2
3use crate::runtime::blocking::BlockingPool;
4use crate::runtime::scheduler::CurrentThread;
5use crate::runtime::{context, AutoBox, Builder, EnterGuard, Handle};
6use crate::task::JoinHandle;
7
8use crate::util::trace::SpawnMeta;
9use std::future::Future;
10use std::marker::PhantomData;
11use std::mem;
12use std::time::Duration;
13
14/// A local Tokio runtime.
15///
16/// This runtime is capable of driving tasks which are not `Send + Sync` without the use of a
17/// `LocalSet`, and thus supports `spawn_local` without the need for a `LocalSet` context.
18///
19/// This runtime cannot be moved between threads or driven from different threads.
20///
21/// This runtime is incompatible with `LocalSet`. You should not attempt to drive a `LocalSet` within a
22/// `LocalRuntime`.
23///
24/// Currently, this runtime supports one flavor, which is internally identical to `current_thread`,
25/// save for the aforementioned differences related to `spawn_local`.
26///
27/// For more general information on how to use runtimes, see the [module] docs.
28///
29/// [runtime]: crate::runtime::Runtime
30/// [module]: crate::runtime
31#[derive(Debug)]
32pub struct LocalRuntime {
33 /// Task scheduler
34 scheduler: LocalRuntimeScheduler,
35
36 /// Handle to runtime, also contains driver handles
37 handle: Handle,
38
39 /// Blocking pool handle, used to signal shutdown
40 blocking_pool: BlockingPool,
41
42 /// Marker used to make this !Send and !Sync.
43 _phantom: PhantomData<*mut u8>,
44}
45
46/// The runtime scheduler is always a `current_thread` scheduler right now.
47#[derive(Debug)]
48pub(crate) enum LocalRuntimeScheduler {
49 /// Execute all tasks on the current-thread.
50 CurrentThread(CurrentThread),
51}
52
53impl LocalRuntime {
54 pub(crate) fn from_parts(
55 scheduler: LocalRuntimeScheduler,
56 handle: Handle,
57 blocking_pool: BlockingPool,
58 ) -> LocalRuntime {
59 LocalRuntime {
60 scheduler,
61 handle,
62 blocking_pool,
63 _phantom: Default::default(),
64 }
65 }
66
67 /// Creates a new local runtime instance with default configuration values.
68 ///
69 /// This results in the scheduler, I/O driver, and time driver being
70 /// initialized.
71 ///
72 /// When a more complex configuration is necessary, the [runtime builder] may be used.
73 ///
74 /// See [module level][mod] documentation for more details.
75 ///
76 /// # Examples
77 ///
78 /// Creating a new `LocalRuntime` with default configuration values.
79 ///
80 /// ```
81 /// use tokio::runtime::LocalRuntime;
82 ///
83 /// let rt = LocalRuntime::new()
84 /// .unwrap();
85 ///
86 /// // Use the runtime...
87 /// ```
88 ///
89 /// # Errors
90 ///
91 /// Returns an error if the I/O driver or other OS resources required by the
92 /// runtime cannot be initialized.
93 ///
94 /// [mod]: crate::runtime
95 /// [runtime builder]: crate::runtime::Builder
96 pub fn new() -> std::io::Result<LocalRuntime> {
97 Builder::new_current_thread()
98 .enable_all()
99 .build_local(Default::default())
100 }
101
102 /// Returns a handle to the runtime's spawner.
103 ///
104 /// The returned handle can be used to spawn tasks that run on this runtime, and can
105 /// be cloned to allow moving the `Handle` to other threads.
106 ///
107 /// As the handle can be sent to other threads, it can only be used to spawn tasks that are `Send`.
108 ///
109 /// Calling [`Handle::block_on`] on a handle to a `LocalRuntime` is error-prone.
110 /// Refer to the documentation of [`Handle::block_on`] for more.
111 ///
112 /// # Examples
113 ///
114 /// ```
115 /// use tokio::runtime::LocalRuntime;
116 ///
117 /// let rt = LocalRuntime::new()
118 /// .unwrap();
119 ///
120 /// let handle = rt.handle();
121 ///
122 /// // Use the handle...
123 /// ```
124 pub fn handle(&self) -> &Handle {
125 &self.handle
126 }
127
128 /// Spawns a task on the runtime.
129 ///
130 /// This is analogous to the [`spawn`] method on the standard [`Runtime`], but works even if the task is not thread-safe.
131 ///
132 /// [`spawn`]: crate::runtime::Runtime::spawn
133 /// [`Runtime`]: crate::runtime::Runtime
134 ///
135 /// # Examples
136 ///
137 /// ```
138 /// use tokio::runtime::LocalRuntime;
139 ///
140 /// # fn dox() {
141 /// // Create the runtime
142 /// let rt = LocalRuntime::new().unwrap();
143 ///
144 /// // Spawn a future onto the runtime
145 /// rt.spawn_local(async {
146 /// println!("now running on a worker thread");
147 /// });
148 /// # }
149 /// ```
150 #[track_caller]
151 pub fn spawn_local<F>(&self, future: F) -> JoinHandle<F::Output>
152 where
153 F: Future + 'static,
154 F::Output: 'static,
155 {
156 let fut_size = std::mem::size_of::<F>();
157 let meta = SpawnMeta::new_unnamed(fut_size);
158
159 // safety: spawn_local can only be called from `LocalRuntime`, which this is
160 unsafe {
161 if AutoBox::<F>::SHOULD_BOX {
162 self.handle.spawn_local_named(Box::pin(future), meta)
163 } else {
164 self.handle.spawn_local_named(future, meta)
165 }
166 }
167 }
168
169 /// Runs the provided function on a thread from a dedicated blocking thread pool.
170 ///
171 /// This function _will_ be run on another thread.
172 ///
173 /// See the [documentation in the non-local runtime][Runtime] for more
174 /// information.
175 ///
176 /// [Runtime]: crate::runtime::Runtime::spawn_blocking
177 ///
178 /// # Examples
179 ///
180 /// ```
181 /// use tokio::runtime::LocalRuntime;
182 ///
183 /// # fn dox() {
184 /// // Create the runtime
185 /// let rt = LocalRuntime::new().unwrap();
186 ///
187 /// // Spawn a blocking function onto the runtime
188 /// rt.spawn_blocking(|| {
189 /// println!("now running on a worker thread");
190 /// });
191 /// # }
192 /// ```
193 #[track_caller]
194 pub fn spawn_blocking<F, R>(&self, func: F) -> JoinHandle<R>
195 where
196 F: FnOnce() -> R + Send + 'static,
197 R: Send + 'static,
198 {
199 self.handle.spawn_blocking(func)
200 }
201
202 /// Runs a future to completion on the Tokio runtime. This is the
203 /// runtime's entry point.
204 ///
205 /// See the documentation for [the equivalent method on Runtime][Runtime]
206 /// for more information.
207 ///
208 /// [Runtime]: crate::runtime::Runtime::block_on
209 ///
210 /// # Examples
211 ///
212 /// ```no_run
213 /// use tokio::runtime::LocalRuntime;
214 ///
215 /// // Create the runtime
216 /// let rt = LocalRuntime::new().unwrap();
217 ///
218 /// // Execute the future, blocking the current thread until completion
219 /// rt.block_on(async {
220 /// println!("hello");
221 /// });
222 /// ```
223 #[track_caller]
224 pub fn block_on<F: Future>(&self, future: F) -> F::Output {
225 let fut_size = mem::size_of::<F>();
226 let meta = SpawnMeta::new_unnamed(fut_size);
227
228 if AutoBox::<F>::SHOULD_BOX {
229 self.block_on_inner(Box::pin(future), meta)
230 } else {
231 self.block_on_inner(future, meta)
232 }
233 }
234
235 #[track_caller]
236 fn block_on_inner<F: Future>(&self, future: F, _meta: SpawnMeta<'_>) -> F::Output {
237 #[cfg(all(
238 tokio_unstable,
239 feature = "taskdump",
240 feature = "rt",
241 target_os = "linux",
242 any(
243 target_arch = "aarch64",
244 target_arch = "x86",
245 target_arch = "x86_64",
246 target_arch = "s390x"
247 )
248 ))]
249 let future = crate::runtime::task::trace::Trace::root(future);
250
251 #[cfg(all(tokio_unstable, feature = "tracing"))]
252 let future = crate::util::trace::task(
253 future,
254 "block_on",
255 _meta,
256 crate::runtime::task::Id::next().as_u64(),
257 );
258
259 let _enter = self.enter();
260
261 if let LocalRuntimeScheduler::CurrentThread(exec) = &self.scheduler {
262 exec.block_on(&self.handle.inner, future)
263 } else {
264 unreachable!("LocalRuntime only supports current_thread")
265 }
266 }
267
268 /// Enters the runtime context.
269 ///
270 /// This allows you to construct types that must have an executor
271 /// available on creation such as [`Sleep`] or [`TcpStream`]. It will
272 /// also allow you to call methods such as [`tokio::spawn`].
273 ///
274 /// If this is a handle to a [`LocalRuntime`], and this function is being invoked from the same
275 /// thread that the runtime was created on, you will also be able to call
276 /// [`tokio::task::spawn_local`].
277 ///
278 /// [`Sleep`]: struct@crate::time::Sleep
279 /// [`TcpStream`]: struct@crate::net::TcpStream
280 /// [`tokio::spawn`]: fn@crate::spawn
281 /// [`LocalRuntime`]: struct@crate::runtime::LocalRuntime
282 /// [`tokio::task::spawn_local`]: fn@crate::task::spawn_local
283 ///
284 /// # Example
285 ///
286 /// ```
287 /// use tokio::runtime::LocalRuntime;
288 /// use tokio::task::JoinHandle;
289 ///
290 /// fn function_that_spawns(msg: String) -> JoinHandle<()> {
291 /// // Had we not used `rt.enter` below, this would panic.
292 /// tokio::spawn(async move {
293 /// println!("{}", msg);
294 /// })
295 /// }
296 ///
297 /// fn main() {
298 /// let rt = LocalRuntime::new().unwrap();
299 ///
300 /// let s = "Hello World!".to_string();
301 ///
302 /// // By entering the context, we tie `tokio::spawn` to this executor.
303 /// let _guard = rt.enter();
304 /// let handle = function_that_spawns(s);
305 ///
306 /// // Wait for the task before we end the test.
307 /// rt.block_on(handle).unwrap();
308 /// }
309 /// ```
310 pub fn enter(&self) -> EnterGuard<'_> {
311 self.handle.enter()
312 }
313
314 /// Shuts down the runtime, waiting for at most `duration` for all spawned
315 /// work to stop.
316 ///
317 /// Note that `spawn_blocking` tasks, and only `spawn_blocking` tasks, can get left behind if
318 /// the timeout expires.
319 ///
320 /// See the [struct level documentation](LocalRuntime#shutdown) for more details.
321 ///
322 /// # Examples
323 ///
324 /// ```
325 /// # #[cfg(not(target_family = "wasm"))]
326 /// # {
327 /// use tokio::runtime::LocalRuntime;
328 /// use tokio::task;
329 ///
330 /// use std::thread;
331 /// use std::time::Duration;
332 ///
333 /// fn main() {
334 /// let runtime = LocalRuntime::new().unwrap();
335 ///
336 /// runtime.block_on(async move {
337 /// task::spawn_blocking(move || {
338 /// thread::sleep(Duration::from_secs(10_000));
339 /// });
340 /// });
341 ///
342 /// runtime.shutdown_timeout(Duration::from_millis(100));
343 /// }
344 /// # }
345 /// ```
346 pub fn shutdown_timeout(mut self, duration: Duration) {
347 // Wakeup and shutdown all the worker threads
348 self.handle.inner.shutdown();
349 self.blocking_pool.shutdown(Some(duration));
350 }
351
352 /// Shuts down the runtime, without waiting for any spawned work to stop.
353 ///
354 /// This can be useful if you want to drop a runtime from within another runtime.
355 /// Normally, dropping a runtime will block indefinitely for spawned blocking tasks
356 /// to complete, which would normally not be permitted within an asynchronous context.
357 /// By calling `shutdown_background()`, you can drop the runtime from such a context.
358 ///
359 /// Note however, that because we do not wait for any blocking tasks to complete, this
360 /// may result in a resource leak (in that any blocking tasks are still running until they
361 /// return. No other tasks will leak.
362 ///
363 /// See the [struct level documentation](LocalRuntime#shutdown) for more details.
364 ///
365 /// This function is equivalent to calling `shutdown_timeout(Duration::from_nanos(0))`.
366 ///
367 /// ```
368 /// use tokio::runtime::LocalRuntime;
369 ///
370 /// fn main() {
371 /// let runtime = LocalRuntime::new().unwrap();
372 ///
373 /// runtime.block_on(async move {
374 /// let inner_runtime = LocalRuntime::new().unwrap();
375 /// // ...
376 /// inner_runtime.shutdown_background();
377 /// });
378 /// }
379 /// ```
380 pub fn shutdown_background(self) {
381 self.shutdown_timeout(Duration::from_nanos(0));
382 }
383
384 /// Returns a view that lets you get information about how the runtime
385 /// is performing.
386 pub fn metrics(&self) -> crate::runtime::RuntimeMetrics {
387 self.handle.metrics()
388 }
389}
390
391impl Drop for LocalRuntime {
392 fn drop(&mut self) {
393 if let LocalRuntimeScheduler::CurrentThread(current_thread) = &mut self.scheduler {
394 // This ensures that tasks spawned on the current-thread
395 // runtime are dropped inside the runtime's context.
396 let _guard = context::try_set_current(&self.handle.inner);
397 current_thread.shutdown(&self.handle.inner);
398 } else {
399 unreachable!("LocalRuntime only supports current-thread")
400 }
401 }
402}
403
404impl std::panic::UnwindSafe for LocalRuntime {}
405
406impl std::panic::RefUnwindSafe for LocalRuntime {}