tokio/runtime/builder.rs
1#![cfg_attr(loom, allow(unused_imports))]
2
3use crate::runtime::handle::Handle;
4use crate::runtime::{
5 blocking, driver, Callback, HistogramBuilder, Runtime, TaskCallback, TimerFlavor,
6};
7#[cfg(tokio_unstable)]
8use crate::runtime::{metrics::HistogramConfiguration, TaskMeta};
9
10use crate::runtime::{LocalOptions, LocalRuntime};
11use crate::util::rand::{RngSeed, RngSeedGenerator};
12
13use crate::runtime::blocking::BlockingPool;
14use crate::runtime::scheduler::CurrentThread;
15use std::fmt;
16use std::io;
17use std::thread::ThreadId;
18use std::time::Duration;
19
20/// Builds Tokio Runtime with custom configuration values.
21///
22/// Methods can be chained in order to set the configuration values. The
23/// Runtime is constructed by calling [`build`].
24///
25/// New instances of `Builder` are obtained via [`Builder::new_multi_thread`]
26/// or [`Builder::new_current_thread`].
27///
28/// See function level documentation for details on the various configuration
29/// settings.
30///
31/// [`build`]: method@Self::build
32/// [`Builder::new_multi_thread`]: method@Self::new_multi_thread
33/// [`Builder::new_current_thread`]: method@Self::new_current_thread
34///
35/// # Examples
36///
37/// ```
38/// # #[cfg(not(target_family = "wasm"))]
39/// # {
40/// use tokio::runtime::Builder;
41///
42/// fn main() {
43/// // build runtime
44/// let runtime = Builder::new_multi_thread()
45/// .worker_threads(4)
46/// .thread_name("my-custom-name")
47/// .thread_stack_size(3 * 1024 * 1024)
48/// .build()
49/// .unwrap();
50///
51/// // use runtime ...
52/// }
53/// # }
54/// ```
55pub struct Builder {
56 /// Runtime type
57 kind: Kind,
58
59 /// Name of the runtime.
60 name: Option<String>,
61
62 /// Whether or not to enable the I/O driver
63 enable_io: bool,
64 nevents: usize,
65
66 /// Whether or not to enable the time driver
67 enable_time: bool,
68
69 /// Whether or not the clock should start paused.
70 start_paused: bool,
71
72 /// The number of worker threads, used by Runtime.
73 ///
74 /// Only used when not using the current-thread executor.
75 worker_threads: Option<usize>,
76
77 /// Cap on thread usage.
78 max_blocking_threads: usize,
79
80 /// Name fn used for threads spawned by the runtime.
81 pub(super) thread_name: ThreadNameFn,
82
83 /// Stack size used for threads spawned by the runtime.
84 pub(super) thread_stack_size: Option<usize>,
85
86 /// Callback to run after each thread starts.
87 pub(super) after_start: Option<Callback>,
88
89 /// To run before each worker thread stops
90 pub(super) before_stop: Option<Callback>,
91
92 /// To run before each worker thread is parked.
93 pub(super) before_park: Option<Callback>,
94
95 /// To run after each thread is unparked.
96 pub(super) after_unpark: Option<Callback>,
97
98 /// To run before each task is spawned.
99 pub(super) before_spawn: Option<TaskCallback>,
100
101 /// To run before each poll
102 #[cfg(tokio_unstable)]
103 pub(super) before_poll: Option<TaskCallback>,
104
105 /// To run after each poll
106 #[cfg(tokio_unstable)]
107 pub(super) after_poll: Option<TaskCallback>,
108
109 /// To run after each task is terminated.
110 pub(super) after_termination: Option<TaskCallback>,
111
112 /// Customizable keep alive timeout for `BlockingPool`
113 pub(super) keep_alive: Option<Duration>,
114
115 /// How many ticks before pulling a task from the global/remote queue?
116 ///
117 /// When `None`, the value is unspecified and behavior details are left to
118 /// the scheduler. Each scheduler flavor could choose to either pick its own
119 /// default value or use some other strategy to decide when to poll from the
120 /// global queue. For example, the multi-threaded scheduler uses a
121 /// self-tuning strategy based on mean task poll times.
122 pub(super) global_queue_interval: Option<u32>,
123
124 /// How many ticks before yielding to the driver for timer and I/O events?
125 pub(super) event_interval: u32,
126
127 /// When true, the multi-threade scheduler LIFO slot should not be used.
128 ///
129 /// This option should only be exposed as unstable.
130 pub(super) disable_lifo_slot: bool,
131
132 /// Specify a random number generator seed to provide deterministic results
133 pub(super) seed_generator: RngSeedGenerator,
134
135 /// When true, enables task poll count histogram instrumentation.
136 pub(super) metrics_poll_count_histogram_enable: bool,
137
138 /// Configures the task poll count histogram
139 pub(super) metrics_poll_count_histogram: HistogramBuilder,
140
141 /// When true, enables task schedule latency instrumentation.
142 pub(super) metrics_schedule_latency_histogram_enabled: bool,
143
144 /// Configures the task schedule latency histogram.
145 pub(super) metrics_schedule_latency_histogram: HistogramBuilder,
146
147 #[cfg(tokio_unstable)]
148 pub(super) unhandled_panic: UnhandledPanic,
149
150 timer_flavor: TimerFlavor,
151
152 /// Whether or not to enable eager hand-off for the I/O and time drivers (in
153 /// `tokio_unstable`).
154 enable_eager_driver_handoff: bool,
155}
156
157cfg_unstable! {
158 /// How the runtime should respond to unhandled panics.
159 ///
160 /// Instances of `UnhandledPanic` are passed to `Builder::unhandled_panic`
161 /// to configure the runtime behavior when a spawned task panics.
162 ///
163 /// See [`Builder::unhandled_panic`] for more details.
164 #[derive(Debug, Clone)]
165 #[non_exhaustive]
166 pub enum UnhandledPanic {
167 /// The runtime should ignore panics on spawned tasks.
168 ///
169 /// The panic is forwarded to the task's [`JoinHandle`] and all spawned
170 /// tasks continue running normally.
171 ///
172 /// This is the default behavior.
173 ///
174 /// # Examples
175 ///
176 /// ```
177 /// # #[cfg(not(target_family = "wasm"))]
178 /// # {
179 /// use tokio::runtime::{self, UnhandledPanic};
180 ///
181 /// # pub fn main() {
182 /// let rt = runtime::Builder::new_current_thread()
183 /// .unhandled_panic(UnhandledPanic::Ignore)
184 /// .build()
185 /// .unwrap();
186 ///
187 /// let task1 = rt.spawn(async { panic!("boom"); });
188 /// let task2 = rt.spawn(async {
189 /// // This task completes normally
190 /// "done"
191 /// });
192 ///
193 /// rt.block_on(async {
194 /// // The panic on the first task is forwarded to the `JoinHandle`
195 /// assert!(task1.await.is_err());
196 ///
197 /// // The second task completes normally
198 /// assert!(task2.await.is_ok());
199 /// })
200 /// # }
201 /// # }
202 /// ```
203 ///
204 /// [`JoinHandle`]: struct@crate::task::JoinHandle
205 Ignore,
206
207 /// The runtime should immediately shutdown if a spawned task panics.
208 ///
209 /// The runtime will immediately shutdown even if the panicked task's
210 /// [`JoinHandle`] is still available. All further spawned tasks will be
211 /// immediately dropped and call to [`Runtime::block_on`] will panic.
212 ///
213 /// # Examples
214 ///
215 /// ```should_panic
216 /// use tokio::runtime::{self, UnhandledPanic};
217 ///
218 /// # pub fn main() {
219 /// let rt = runtime::Builder::new_current_thread()
220 /// .unhandled_panic(UnhandledPanic::ShutdownRuntime)
221 /// .build()
222 /// .unwrap();
223 ///
224 /// rt.spawn(async { panic!("boom"); });
225 /// rt.spawn(async {
226 /// // This task never completes.
227 /// });
228 ///
229 /// rt.block_on(async {
230 /// // Do some work
231 /// # loop { tokio::task::yield_now().await; }
232 /// })
233 /// # }
234 /// ```
235 ///
236 /// [`JoinHandle`]: struct@crate::task::JoinHandle
237 ShutdownRuntime,
238 }
239}
240
241pub(crate) type ThreadNameFn = std::sync::Arc<dyn Fn() -> String + Send + Sync + 'static>;
242
243#[derive(Clone, Copy)]
244pub(crate) enum Kind {
245 CurrentThread,
246 #[cfg(feature = "rt-multi-thread")]
247 MultiThread,
248}
249
250impl Builder {
251 /// Returns a new builder with the current thread scheduler selected.
252 ///
253 /// Configuration methods can be chained on the return value.
254 ///
255 /// To spawn non-`Send` tasks on the resulting runtime, combine it with a
256 /// [`LocalSet`], or call [`build_local`] to create a [`LocalRuntime`].
257 ///
258 /// [`LocalSet`]: crate::task::LocalSet
259 /// [`LocalRuntime`]: crate::runtime::LocalRuntime
260 /// [`build_local`]: crate::runtime::Builder::build_local
261 pub fn new_current_thread() -> Builder {
262 #[cfg(loom)]
263 const EVENT_INTERVAL: u32 = 4;
264 // The number `61` is fairly arbitrary. I believe this value was copied from golang.
265 #[cfg(not(loom))]
266 const EVENT_INTERVAL: u32 = 61;
267
268 Builder::new(Kind::CurrentThread, EVENT_INTERVAL)
269 }
270
271 /// Returns a new builder with the multi thread scheduler selected.
272 ///
273 /// Configuration methods can be chained on the return value.
274 #[cfg(feature = "rt-multi-thread")]
275 #[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))]
276 pub fn new_multi_thread() -> Builder {
277 // The number `61` is fairly arbitrary. I believe this value was copied from golang.
278 Builder::new(Kind::MultiThread, 61)
279 }
280
281 /// Returns a new runtime builder initialized with default configuration
282 /// values.
283 ///
284 /// Configuration methods can be chained on the return value.
285 pub(crate) fn new(kind: Kind, event_interval: u32) -> Builder {
286 Builder {
287 kind,
288
289 // Default runtime name
290 name: None,
291
292 // I/O defaults to "off"
293 enable_io: false,
294 nevents: 1024,
295
296 // Time defaults to "off"
297 enable_time: false,
298
299 // The clock starts not-paused
300 start_paused: false,
301
302 // Read from environment variable first in multi-threaded mode.
303 // Default to lazy auto-detection (one thread per CPU core)
304 worker_threads: None,
305
306 max_blocking_threads: 512,
307
308 // Default thread name
309 thread_name: std::sync::Arc::new(|| "tokio-rt-worker".into()),
310
311 // Do not set a stack size by default
312 thread_stack_size: None,
313
314 // No worker thread callbacks
315 after_start: None,
316 before_stop: None,
317 before_park: None,
318 after_unpark: None,
319
320 before_spawn: None,
321 after_termination: None,
322
323 #[cfg(tokio_unstable)]
324 before_poll: None,
325 #[cfg(tokio_unstable)]
326 after_poll: None,
327
328 keep_alive: None,
329
330 // Defaults for these values depend on the scheduler kind, so we get them
331 // as parameters.
332 global_queue_interval: None,
333 event_interval,
334
335 seed_generator: RngSeedGenerator::new(RngSeed::new()),
336
337 #[cfg(tokio_unstable)]
338 unhandled_panic: UnhandledPanic::Ignore,
339
340 metrics_poll_count_histogram_enable: false,
341
342 metrics_poll_count_histogram: HistogramBuilder::default(),
343
344 metrics_schedule_latency_histogram_enabled: false,
345
346 metrics_schedule_latency_histogram: HistogramBuilder::default(),
347
348 disable_lifo_slot: false,
349
350 timer_flavor: TimerFlavor::Traditional,
351
352 // Eager driver handoff is disabled by default.
353 enable_eager_driver_handoff: false,
354 }
355 }
356
357 /// Enables both I/O and time drivers.
358 ///
359 /// Doing this is a shorthand for calling `enable_io` and `enable_time`
360 /// individually. If additional components are added to Tokio in the future,
361 /// `enable_all` will include these future components.
362 ///
363 /// # Examples
364 ///
365 /// ```
366 /// # #[cfg(not(target_family = "wasm"))]
367 /// # {
368 /// use tokio::runtime;
369 ///
370 /// let rt = runtime::Builder::new_multi_thread()
371 /// .enable_all()
372 /// .build()
373 /// .unwrap();
374 /// # }
375 /// ```
376 pub fn enable_all(&mut self) -> &mut Self {
377 #[cfg(any(
378 feature = "net",
379 all(unix, feature = "process"),
380 all(unix, feature = "signal")
381 ))]
382 self.enable_io();
383
384 #[cfg(all(
385 tokio_unstable,
386 feature = "io-uring",
387 feature = "rt",
388 feature = "fs",
389 target_os = "linux",
390 ))]
391 self.enable_io_uring();
392
393 #[cfg(feature = "time")]
394 self.enable_time();
395
396 self
397 }
398
399 /// Enables the alternative timer implementation, which is disabled by default.
400 ///
401 /// The alternative timer implementation is an unstable feature that may
402 /// provide better performance on multi-threaded runtimes with a large number
403 /// of worker threads.
404 ///
405 /// This option only applies to multi-threaded runtimes. Attempting to use
406 /// this option with any other runtime type will have no effect.
407 ///
408 /// [Click here to share your experience with the alternative timer](https://github.com/tokio-rs/tokio/issues/7745)
409 ///
410 /// # Examples
411 ///
412 /// ```
413 /// # #[cfg(not(target_family = "wasm"))]
414 /// # {
415 /// use tokio::runtime;
416 ///
417 /// let rt = runtime::Builder::new_multi_thread()
418 /// .enable_alt_timer()
419 /// .build()
420 /// .unwrap();
421 /// # }
422 /// ```
423 #[cfg(all(tokio_unstable, feature = "time", feature = "rt-multi-thread"))]
424 #[cfg_attr(
425 docsrs,
426 doc(cfg(all(tokio_unstable, feature = "time", feature = "rt-multi-thread")))
427 )]
428 pub fn enable_alt_timer(&mut self) -> &mut Self {
429 self.enable_time();
430 self.timer_flavor = TimerFlavor::Alternative;
431 self
432 }
433
434 /// Enable eager hand-off of the I/O and time drivers for multi-threaded
435 /// runtimes, which is disabled by default.
436 ///
437 /// When this option is enabled, a worker thread which has parked on the I/O
438 /// or time driver will notify another worker thread once it is preparing to
439 /// begin polling a task from the run queue, so that the notified worker can
440 /// begin polling the I/O or time driver. This can reduce the latency with
441 /// which I/O and timer notifications are processed, especially when some
442 /// tasks have polls that take a long time to complete. In addition, it can
443 /// reduce the risk of a deadlock which may occur when a task blocks the
444 /// worker thread which is holding the I/O or time driver until some other
445 /// task, which is waiting for a notification from *that* driver, unblocks
446 /// it.
447 ///
448 /// This option is disabled by default, as enabling it may potentially
449 /// increase contention due to extra synchronization in cross-driver
450 /// wakeups.
451 ///
452 /// This option only applies to multi-threaded runtimes. Attempting to use
453 /// this option with any other runtime type will have no effect.
454 ///
455 /// **Note**: This is an [unstable API][unstable]. Eager driver hand-off is
456 /// an experimental feature whose behavior may be removed or changed in 1.x
457 /// releases. See [the documentation on unstable features][unstable] for
458 /// details.
459 ///
460 /// [unstable]: crate#unstable-features
461 #[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
462 #[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "rt-multi-thread"))))]
463 pub fn enable_eager_driver_handoff(&mut self) -> &mut Self {
464 self.enable_eager_driver_handoff = true;
465 self
466 }
467
468 /// Sets the number of worker threads the `Runtime` will use.
469 ///
470 /// This can be any number above 0 though it is advised to keep this value
471 /// on the smaller side.
472 ///
473 /// This will override the value read from environment variable `TOKIO_WORKER_THREADS`.
474 ///
475 /// # Default
476 ///
477 /// The default value is the number of cores available to the system.
478 ///
479 /// When using the `current_thread` runtime this method has no effect.
480 ///
481 /// # Examples
482 ///
483 /// ## Multi threaded runtime with 4 threads
484 ///
485 /// ```
486 /// # #[cfg(not(target_family = "wasm"))]
487 /// # {
488 /// use tokio::runtime;
489 ///
490 /// // This will spawn a work-stealing runtime with 4 worker threads.
491 /// let rt = runtime::Builder::new_multi_thread()
492 /// .worker_threads(4)
493 /// .build()
494 /// .unwrap();
495 ///
496 /// rt.spawn(async move {});
497 /// # }
498 /// ```
499 ///
500 /// ## Current thread runtime (will only run on the current thread via `Runtime::block_on`)
501 ///
502 /// ```
503 /// use tokio::runtime;
504 ///
505 /// // Create a runtime that _must_ be driven from a call
506 /// // to `Runtime::block_on`.
507 /// let rt = runtime::Builder::new_current_thread()
508 /// .build()
509 /// .unwrap();
510 ///
511 /// // This will run the runtime and future on the current thread
512 /// rt.block_on(async move {});
513 /// ```
514 ///
515 /// # Panics
516 ///
517 /// This will panic if `val` is not larger than `0`.
518 #[track_caller]
519 pub fn worker_threads(&mut self, val: usize) -> &mut Self {
520 assert!(val > 0, "Worker threads cannot be set to 0");
521 self.worker_threads = Some(val);
522 self
523 }
524
525 /// Specifies the limit for additional threads spawned by the Runtime.
526 ///
527 /// These threads are used for blocking operations like tasks spawned
528 /// through [`spawn_blocking`], this includes but is not limited to:
529 /// - [`fs`] operations
530 /// - dns resolution through [`ToSocketAddrs`]
531 /// - writing to [`Stdout`] or [`Stderr`]
532 /// - reading from [`Stdin`]
533 ///
534 /// Unlike the [`worker_threads`], they are not always active and will exit
535 /// if left idle for too long. You can change this timeout duration with [`thread_keep_alive`].
536 ///
537 /// It's recommended to not set this limit too low in order to avoid hanging on operations
538 /// requiring [`spawn_blocking`].
539 ///
540 /// The default value is 512.
541 ///
542 /// # Queue Behavior
543 ///
544 /// When a blocking task is submitted, it will be inserted into a queue. If available, one of
545 /// the idle threads will be notified to run the task. Otherwise, if the threshold set by this
546 /// method has not been reached, a new thread will be spawned. If no idle thread is available
547 /// and no more threads are allowed to be spawned, the task will remain in the queue until one
548 /// of the busy threads pick it up. Note that since the queue does not apply any backpressure,
549 /// it could potentially grow unbounded.
550 ///
551 /// # Panics
552 ///
553 /// This will panic if `val` is not larger than `0`.
554 ///
555 /// # Upgrading from 0.x
556 ///
557 /// In old versions `max_threads` limited both blocking and worker threads, but the
558 /// current `max_blocking_threads` does not include async worker threads in the count.
559 ///
560 /// [`spawn_blocking`]: fn@crate::task::spawn_blocking
561 /// [`fs`]: mod@crate::fs
562 /// [`ToSocketAddrs`]: trait@crate::net::ToSocketAddrs
563 /// [`Stdout`]: struct@crate::io::Stdout
564 /// [`Stdin`]: struct@crate::io::Stdin
565 /// [`Stderr`]: struct@crate::io::Stderr
566 /// [`worker_threads`]: Self::worker_threads
567 /// [`thread_keep_alive`]: Self::thread_keep_alive
568 #[track_caller]
569 #[cfg_attr(docsrs, doc(alias = "max_threads"))]
570 pub fn max_blocking_threads(&mut self, val: usize) -> &mut Self {
571 assert!(val > 0, "Max blocking threads cannot be set to 0");
572 self.max_blocking_threads = val;
573 self
574 }
575
576 /// Sets name of threads spawned by the `Runtime`'s thread pool.
577 ///
578 /// The default name is "tokio-rt-worker".
579 ///
580 /// # Examples
581 ///
582 /// ```
583 /// # #[cfg(not(target_family = "wasm"))]
584 /// # {
585 /// # use tokio::runtime;
586 ///
587 /// # pub fn main() {
588 /// let rt = runtime::Builder::new_multi_thread()
589 /// .thread_name("my-pool")
590 /// .build();
591 /// # }
592 /// # }
593 /// ```
594 pub fn thread_name(&mut self, val: impl Into<String>) -> &mut Self {
595 let val = val.into();
596 self.thread_name = std::sync::Arc::new(move || val.clone());
597 self
598 }
599
600 /// Sets the name of the runtime.
601 ///
602 /// # Examples
603 ///
604 /// ```
605 /// # #[cfg(not(target_family = "wasm"))]
606 /// # {
607 /// # use tokio::runtime;
608 ///
609 /// # pub fn main() {
610 /// let rt = runtime::Builder::new_multi_thread()
611 /// .name("my-runtime")
612 /// .build();
613 /// # }
614 /// # }
615 /// ```
616 /// # Panics
617 ///
618 /// This function will panic if an empty value is passed as an argument.
619 ///
620 #[track_caller]
621 pub fn name(&mut self, val: impl Into<String>) -> &mut Self {
622 let val = val.into();
623 assert!(!val.trim().is_empty(), "runtime name shouldn't be empty");
624 self.name = Some(val);
625 self
626 }
627
628 /// Sets a function used to generate the name of threads spawned by the `Runtime`'s thread pool.
629 ///
630 /// The default name fn is `|| "tokio-rt-worker".into()`.
631 ///
632 /// # Examples
633 ///
634 /// ```
635 /// # #[cfg(not(target_family = "wasm"))]
636 /// # {
637 /// # use tokio::runtime;
638 /// # use std::sync::atomic::{AtomicUsize, Ordering};
639 /// # pub fn main() {
640 /// let rt = runtime::Builder::new_multi_thread()
641 /// .thread_name_fn(|| {
642 /// static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
643 /// let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
644 /// format!("my-pool-{}", id)
645 /// })
646 /// .build();
647 /// # }
648 /// # }
649 /// ```
650 pub fn thread_name_fn<F>(&mut self, f: F) -> &mut Self
651 where
652 F: Fn() -> String + Send + Sync + 'static,
653 {
654 self.thread_name = std::sync::Arc::new(f);
655 self
656 }
657
658 /// Sets the stack size (in bytes) for worker threads.
659 ///
660 /// The actual stack size may be greater than this value if the platform
661 /// specifies minimal stack size.
662 ///
663 /// The default stack size for spawned threads is 2 MiB, though this
664 /// particular stack size is subject to change in the future.
665 ///
666 /// # Examples
667 ///
668 /// ```
669 /// # #[cfg(not(target_family = "wasm"))]
670 /// # {
671 /// # use tokio::runtime;
672 ///
673 /// # pub fn main() {
674 /// let rt = runtime::Builder::new_multi_thread()
675 /// .thread_stack_size(32 * 1024)
676 /// .build();
677 /// # }
678 /// # }
679 /// ```
680 pub fn thread_stack_size(&mut self, val: usize) -> &mut Self {
681 self.thread_stack_size = Some(val);
682 self
683 }
684
685 /// Executes function `f` after each thread is started but before it starts
686 /// doing work.
687 ///
688 /// This is intended for bookkeeping and monitoring use cases.
689 ///
690 /// # Examples
691 ///
692 /// ```
693 /// # #[cfg(not(target_family = "wasm"))]
694 /// # {
695 /// # use tokio::runtime;
696 /// # pub fn main() {
697 /// let runtime = runtime::Builder::new_multi_thread()
698 /// .on_thread_start(|| {
699 /// println!("thread started");
700 /// })
701 /// .build();
702 /// # }
703 /// # }
704 /// ```
705 #[cfg(not(loom))]
706 pub fn on_thread_start<F>(&mut self, f: F) -> &mut Self
707 where
708 F: Fn() + Send + Sync + 'static,
709 {
710 self.after_start = Some(std::sync::Arc::new(f));
711 self
712 }
713
714 /// Executes function `f` before each thread stops.
715 ///
716 /// This is intended for bookkeeping and monitoring use cases.
717 ///
718 /// # Examples
719 ///
720 /// ```
721 /// # #[cfg(not(target_family = "wasm"))]
722 /// {
723 /// # use tokio::runtime;
724 /// # pub fn main() {
725 /// let runtime = runtime::Builder::new_multi_thread()
726 /// .on_thread_stop(|| {
727 /// println!("thread stopping");
728 /// })
729 /// .build();
730 /// # }
731 /// # }
732 /// ```
733 #[cfg(not(loom))]
734 pub fn on_thread_stop<F>(&mut self, f: F) -> &mut Self
735 where
736 F: Fn() + Send + Sync + 'static,
737 {
738 self.before_stop = Some(std::sync::Arc::new(f));
739 self
740 }
741
742 /// Executes function `f` just before a thread is parked (goes idle).
743 /// `f` is called within the Tokio context, so functions like [`tokio::spawn`](crate::spawn)
744 /// can be called, and may result in this thread being unparked immediately.
745 ///
746 /// This can be used to start work only when the executor is idle, or for bookkeeping
747 /// and monitoring purposes.
748 ///
749 /// Note: There can only be one park callback for a runtime; calling this function
750 /// more than once replaces the last callback defined, rather than adding to it.
751 ///
752 /// # Examples
753 ///
754 /// ## Multithreaded executor
755 /// ```
756 /// # #[cfg(not(target_family = "wasm"))]
757 /// # {
758 /// # use std::sync::Arc;
759 /// # use std::sync::atomic::{AtomicBool, Ordering};
760 /// # use tokio::runtime;
761 /// # use tokio::sync::Barrier;
762 /// # pub fn main() {
763 /// let once = AtomicBool::new(true);
764 /// let barrier = Arc::new(Barrier::new(2));
765 ///
766 /// let runtime = runtime::Builder::new_multi_thread()
767 /// .worker_threads(1)
768 /// .on_thread_park({
769 /// let barrier = barrier.clone();
770 /// move || {
771 /// let barrier = barrier.clone();
772 /// if once.swap(false, Ordering::Relaxed) {
773 /// tokio::spawn(async move { barrier.wait().await; });
774 /// }
775 /// }
776 /// })
777 /// .build()
778 /// .unwrap();
779 ///
780 /// runtime.block_on(async {
781 /// barrier.wait().await;
782 /// })
783 /// # }
784 /// # }
785 /// ```
786 /// ## Current thread executor
787 /// ```
788 /// # use std::sync::Arc;
789 /// # use std::sync::atomic::{AtomicBool, Ordering};
790 /// # use tokio::runtime;
791 /// # use tokio::sync::Barrier;
792 /// # pub fn main() {
793 /// let once = AtomicBool::new(true);
794 /// let barrier = Arc::new(Barrier::new(2));
795 ///
796 /// let runtime = runtime::Builder::new_current_thread()
797 /// .on_thread_park({
798 /// let barrier = barrier.clone();
799 /// move || {
800 /// let barrier = barrier.clone();
801 /// if once.swap(false, Ordering::Relaxed) {
802 /// tokio::spawn(async move { barrier.wait().await; });
803 /// }
804 /// }
805 /// })
806 /// .build()
807 /// .unwrap();
808 ///
809 /// runtime.block_on(async {
810 /// barrier.wait().await;
811 /// })
812 /// # }
813 /// ```
814 #[cfg(not(loom))]
815 pub fn on_thread_park<F>(&mut self, f: F) -> &mut Self
816 where
817 F: Fn() + Send + Sync + 'static,
818 {
819 self.before_park = Some(std::sync::Arc::new(f));
820 self
821 }
822
823 /// Executes function `f` just after a thread unparks (starts executing tasks).
824 ///
825 /// This is intended for bookkeeping and monitoring use cases; note that work
826 /// in this callback will increase latencies when the application has allowed one or
827 /// more runtime threads to go idle.
828 ///
829 /// Note: There can only be one unpark callback for a runtime; calling this function
830 /// more than once replaces the last callback defined, rather than adding to it.
831 ///
832 /// # Examples
833 ///
834 /// ```
835 /// # #[cfg(not(target_family = "wasm"))]
836 /// # {
837 /// # use tokio::runtime;
838 /// # pub fn main() {
839 /// let runtime = runtime::Builder::new_multi_thread()
840 /// .on_thread_unpark(|| {
841 /// println!("thread unparking");
842 /// })
843 /// .build();
844 ///
845 /// runtime.unwrap().block_on(async {
846 /// tokio::task::yield_now().await;
847 /// println!("Hello from Tokio!");
848 /// })
849 /// # }
850 /// # }
851 /// ```
852 #[cfg(not(loom))]
853 pub fn on_thread_unpark<F>(&mut self, f: F) -> &mut Self
854 where
855 F: Fn() + Send + Sync + 'static,
856 {
857 self.after_unpark = Some(std::sync::Arc::new(f));
858 self
859 }
860
861 /// Executes function `f` just before a task is spawned.
862 ///
863 /// `f` is called within the Tokio context, so functions like
864 /// [`tokio::spawn`](crate::spawn) can be called, and may result in this callback being
865 /// invoked immediately.
866 ///
867 /// This can be used for bookkeeping or monitoring purposes.
868 ///
869 /// Note: There can only be one spawn callback for a runtime; calling this function more
870 /// than once replaces the last callback defined, rather than adding to it.
871 ///
872 /// This *does not* support [`LocalSet`](crate::task::LocalSet) at this time.
873 ///
874 /// **Note**: This is an [unstable API][unstable]. The public API of this type
875 /// may break in 1.x releases. See [the documentation on unstable
876 /// features][unstable] for details.
877 ///
878 /// [unstable]: crate#unstable-features
879 ///
880 /// # Examples
881 ///
882 /// ```
883 /// # use tokio::runtime;
884 /// # pub fn main() {
885 /// let runtime = runtime::Builder::new_current_thread()
886 /// .on_task_spawn(|_| {
887 /// println!("spawning task");
888 /// })
889 /// .build()
890 /// .unwrap();
891 ///
892 /// runtime.block_on(async {
893 /// tokio::task::spawn(std::future::ready(()));
894 ///
895 /// for _ in 0..64 {
896 /// tokio::task::yield_now().await;
897 /// }
898 /// })
899 /// # }
900 /// ```
901 #[cfg(all(not(loom), tokio_unstable))]
902 #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
903 pub fn on_task_spawn<F>(&mut self, f: F) -> &mut Self
904 where
905 F: Fn(&TaskMeta<'_>) + Send + Sync + 'static,
906 {
907 self.before_spawn = Some(std::sync::Arc::new(f));
908 self
909 }
910
911 /// Executes function `f` just before a task is polled
912 ///
913 /// `f` is called within the Tokio context, so functions like
914 /// [`tokio::spawn`](crate::spawn) can be called, and may result in this callback being
915 /// invoked immediately.
916 ///
917 /// **Note**: This is an [unstable API][unstable]. The public API of this type
918 /// may break in 1.x releases. See [the documentation on unstable
919 /// features][unstable] for details.
920 ///
921 /// [unstable]: crate#unstable-features
922 ///
923 /// # Examples
924 ///
925 /// ```
926 /// # #[cfg(not(target_family = "wasm"))]
927 /// # {
928 /// # use std::sync::{atomic::AtomicUsize, Arc};
929 /// # use tokio::task::yield_now;
930 /// # pub fn main() {
931 /// let poll_start_counter = Arc::new(AtomicUsize::new(0));
932 /// let poll_start = poll_start_counter.clone();
933 /// let rt = tokio::runtime::Builder::new_multi_thread()
934 /// .enable_all()
935 /// .on_before_task_poll(move |meta| {
936 /// println!("task {} is about to be polled", meta.id())
937 /// })
938 /// .build()
939 /// .unwrap();
940 /// let task = rt.spawn(async {
941 /// yield_now().await;
942 /// });
943 /// let _ = rt.block_on(task);
944 ///
945 /// # }
946 /// # }
947 /// ```
948 #[cfg(tokio_unstable)]
949 #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
950 pub fn on_before_task_poll<F>(&mut self, f: F) -> &mut Self
951 where
952 F: Fn(&TaskMeta<'_>) + Send + Sync + 'static,
953 {
954 self.before_poll = Some(std::sync::Arc::new(f));
955 self
956 }
957
958 /// Executes function `f` just after a task is polled
959 ///
960 /// `f` is called within the Tokio context, so functions like
961 /// [`tokio::spawn`](crate::spawn) can be called, and may result in this callback being
962 /// invoked immediately.
963 ///
964 /// **Note**: This is an [unstable API][unstable]. The public API of this type
965 /// may break in 1.x releases. See [the documentation on unstable
966 /// features][unstable] for details.
967 ///
968 /// [unstable]: crate#unstable-features
969 ///
970 /// # Examples
971 ///
972 /// ```
973 /// # #[cfg(not(target_family = "wasm"))]
974 /// # {
975 /// # use std::sync::{atomic::AtomicUsize, Arc};
976 /// # use tokio::task::yield_now;
977 /// # pub fn main() {
978 /// let poll_stop_counter = Arc::new(AtomicUsize::new(0));
979 /// let poll_stop = poll_stop_counter.clone();
980 /// let rt = tokio::runtime::Builder::new_multi_thread()
981 /// .enable_all()
982 /// .on_after_task_poll(move |meta| {
983 /// println!("task {} completed polling", meta.id());
984 /// })
985 /// .build()
986 /// .unwrap();
987 /// let task = rt.spawn(async {
988 /// yield_now().await;
989 /// });
990 /// let _ = rt.block_on(task);
991 ///
992 /// # }
993 /// # }
994 /// ```
995 #[cfg(tokio_unstable)]
996 #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
997 pub fn on_after_task_poll<F>(&mut self, f: F) -> &mut Self
998 where
999 F: Fn(&TaskMeta<'_>) + Send + Sync + 'static,
1000 {
1001 self.after_poll = Some(std::sync::Arc::new(f));
1002 self
1003 }
1004
1005 /// Executes function `f` just after a task is terminated.
1006 ///
1007 /// `f` is called within the Tokio context, so functions like
1008 /// [`tokio::spawn`](crate::spawn) can be called.
1009 ///
1010 /// This can be used for bookkeeping or monitoring purposes.
1011 ///
1012 /// Note: There can only be one task termination callback for a runtime; calling this
1013 /// function more than once replaces the last callback defined, rather than adding to it.
1014 ///
1015 /// This *does not* support [`LocalSet`](crate::task::LocalSet) at this time.
1016 ///
1017 /// **Note**: This is an [unstable API][unstable]. The public API of this type
1018 /// may break in 1.x releases. See [the documentation on unstable
1019 /// features][unstable] for details.
1020 ///
1021 /// [unstable]: crate#unstable-features
1022 ///
1023 /// # Examples
1024 ///
1025 /// ```
1026 /// # use tokio::runtime;
1027 /// # pub fn main() {
1028 /// let runtime = runtime::Builder::new_current_thread()
1029 /// .on_task_terminate(|_| {
1030 /// println!("killing task");
1031 /// })
1032 /// .build()
1033 /// .unwrap();
1034 ///
1035 /// runtime.block_on(async {
1036 /// tokio::task::spawn(std::future::ready(()));
1037 ///
1038 /// for _ in 0..64 {
1039 /// tokio::task::yield_now().await;
1040 /// }
1041 /// })
1042 /// # }
1043 /// ```
1044 #[cfg(all(not(loom), tokio_unstable))]
1045 #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
1046 pub fn on_task_terminate<F>(&mut self, f: F) -> &mut Self
1047 where
1048 F: Fn(&TaskMeta<'_>) + Send + Sync + 'static,
1049 {
1050 self.after_termination = Some(std::sync::Arc::new(f));
1051 self
1052 }
1053
1054 /// Creates the configured `Runtime`.
1055 ///
1056 /// The returned `Runtime` instance is ready to spawn tasks.
1057 ///
1058 /// # Examples
1059 ///
1060 /// ```
1061 /// # #[cfg(not(target_family = "wasm"))]
1062 /// # {
1063 /// use tokio::runtime::Builder;
1064 ///
1065 /// let rt = Builder::new_multi_thread().build().unwrap();
1066 ///
1067 /// rt.block_on(async {
1068 /// println!("Hello from the Tokio runtime");
1069 /// });
1070 /// # }
1071 /// ```
1072 pub fn build(&mut self) -> io::Result<Runtime> {
1073 match &self.kind {
1074 Kind::CurrentThread => self.build_current_thread_runtime(),
1075 #[cfg(feature = "rt-multi-thread")]
1076 Kind::MultiThread => self.build_threaded_runtime(),
1077 }
1078 }
1079
1080 /// Creates the configured [`LocalRuntime`].
1081 ///
1082 /// The returned [`LocalRuntime`] instance is ready to spawn tasks.
1083 ///
1084 /// # Panics
1085 ///
1086 /// This will panic if the runtime is configured with [`new_multi_thread()`].
1087 ///
1088 /// [`new_multi_thread()`]: Builder::new_multi_thread
1089 ///
1090 /// # Examples
1091 ///
1092 /// ```
1093 /// use tokio::runtime::{Builder, LocalOptions};
1094 ///
1095 /// let rt = Builder::new_current_thread()
1096 /// .build_local(LocalOptions::default())
1097 /// .unwrap();
1098 ///
1099 /// rt.spawn_local(async {
1100 /// println!("Hello from the Tokio runtime");
1101 /// });
1102 /// ```
1103 pub fn build_local(&mut self, _options: LocalOptions) -> io::Result<LocalRuntime> {
1104 match &self.kind {
1105 Kind::CurrentThread => self.build_current_thread_local_runtime(),
1106 #[cfg(feature = "rt-multi-thread")]
1107 Kind::MultiThread => panic!("multi_thread is not supported for LocalRuntime"),
1108 }
1109 }
1110
1111 fn get_cfg(&self) -> driver::Cfg {
1112 driver::Cfg {
1113 enable_pause_time: match self.kind {
1114 Kind::CurrentThread => true,
1115 #[cfg(feature = "rt-multi-thread")]
1116 Kind::MultiThread => false,
1117 },
1118 enable_io: self.enable_io,
1119 enable_time: self.enable_time,
1120 start_paused: self.start_paused,
1121 nevents: self.nevents,
1122 timer_flavor: self.timer_flavor,
1123 }
1124 }
1125
1126 /// Sets a custom timeout for a thread in the blocking pool.
1127 ///
1128 /// By default, the timeout for a thread is set to 10 seconds. This can
1129 /// be overridden using `.thread_keep_alive()`.
1130 ///
1131 /// # Example
1132 ///
1133 /// ```
1134 /// # #[cfg(not(target_family = "wasm"))]
1135 /// # {
1136 /// # use tokio::runtime;
1137 /// # use std::time::Duration;
1138 /// # pub fn main() {
1139 /// let rt = runtime::Builder::new_multi_thread()
1140 /// .thread_keep_alive(Duration::from_millis(100))
1141 /// .build();
1142 /// # }
1143 /// # }
1144 /// ```
1145 pub fn thread_keep_alive(&mut self, duration: Duration) -> &mut Self {
1146 self.keep_alive = Some(duration);
1147 self
1148 }
1149
1150 /// Sets the number of scheduler ticks after which the scheduler will poll the global
1151 /// task queue.
1152 ///
1153 /// A scheduler "tick" roughly corresponds to one `poll` invocation on a task.
1154 ///
1155 /// By default the global queue interval is 31 for the current-thread scheduler. Please see
1156 /// [the module documentation] for the default behavior of the multi-thread scheduler.
1157 ///
1158 /// Schedulers have a local queue of already-claimed tasks, and a global queue of incoming
1159 /// tasks. Setting the interval to a smaller value increases the fairness of the scheduler,
1160 /// at the cost of more synchronization overhead. That can be beneficial for prioritizing
1161 /// getting started on new work, especially if tasks frequently yield rather than complete
1162 /// or await on further I/O. Setting the interval to `1` will prioritize the global queue and
1163 /// tasks from the local queue will be executed only if the global queue is empty.
1164 /// Conversely, a higher value prioritizes existing work, and is a good choice when most
1165 /// tasks quickly complete polling.
1166 ///
1167 /// [the module documentation]: crate::runtime#multi-threaded-runtime-behavior-at-the-time-of-writing
1168 ///
1169 /// # Panics
1170 ///
1171 /// This function will panic if 0 is passed as an argument.
1172 ///
1173 /// # Examples
1174 ///
1175 /// ```
1176 /// # #[cfg(not(target_family = "wasm"))]
1177 /// # {
1178 /// # use tokio::runtime;
1179 /// # pub fn main() {
1180 /// let rt = runtime::Builder::new_multi_thread()
1181 /// .global_queue_interval(31)
1182 /// .build();
1183 /// # }
1184 /// # }
1185 /// ```
1186 #[track_caller]
1187 pub fn global_queue_interval(&mut self, val: u32) -> &mut Self {
1188 assert!(val > 0, "global_queue_interval must be greater than 0");
1189 self.global_queue_interval = Some(val);
1190 self
1191 }
1192
1193 /// Sets the number of scheduler ticks after which the scheduler will poll for
1194 /// external events (timers, I/O, and so on).
1195 ///
1196 /// A scheduler "tick" roughly corresponds to one `poll` invocation on a task.
1197 ///
1198 /// By default, the event interval is `61` for all scheduler types.
1199 ///
1200 /// Setting the event interval determines the effective "priority" of delivering
1201 /// these external events (which may wake up additional tasks), compared to
1202 /// executing tasks that are currently ready to run. A smaller value is useful
1203 /// when tasks frequently spend a long time in polling, or infrequently yield,
1204 /// which can result in overly long delays picking up I/O events. Conversely,
1205 /// picking up new events requires extra synchronization and syscall overhead,
1206 /// so if tasks generally complete their polling quickly, a higher event interval
1207 /// will minimize that overhead while still keeping the scheduler responsive to
1208 /// events.
1209 ///
1210 /// # Panics
1211 ///
1212 /// This function will panic if 0 is passed as an argument.
1213 ///
1214 /// # Examples
1215 ///
1216 /// ```
1217 /// # #[cfg(not(target_family = "wasm"))]
1218 /// # {
1219 /// # use tokio::runtime;
1220 /// # pub fn main() {
1221 /// let rt = runtime::Builder::new_multi_thread()
1222 /// .event_interval(31)
1223 /// .build();
1224 /// # }
1225 /// # }
1226 /// ```
1227 #[track_caller]
1228 pub fn event_interval(&mut self, val: u32) -> &mut Self {
1229 assert!(val > 0, "event_interval must be greater than 0");
1230 self.event_interval = val;
1231 self
1232 }
1233
1234 cfg_unstable! {
1235 /// Configure how the runtime responds to an unhandled panic on a
1236 /// spawned task.
1237 ///
1238 /// By default, an unhandled panic (i.e. a panic not caught by
1239 /// [`std::panic::catch_unwind`]) has no impact on the runtime's
1240 /// execution. The panic's error value is forwarded to the task's
1241 /// [`JoinHandle`] and all other spawned tasks continue running.
1242 ///
1243 /// The `unhandled_panic` option enables configuring this behavior.
1244 ///
1245 /// * `UnhandledPanic::Ignore` is the default behavior. Panics on
1246 /// spawned tasks have no impact on the runtime's execution.
1247 /// * `UnhandledPanic::ShutdownRuntime` will force the runtime to
1248 /// shutdown immediately when a spawned task panics even if that
1249 /// task's `JoinHandle` has not been dropped. All other spawned tasks
1250 /// will immediately terminate and further calls to
1251 /// [`Runtime::block_on`] will panic.
1252 ///
1253 /// # Panics
1254 /// This method panics if called with [`UnhandledPanic::ShutdownRuntime`]
1255 /// on a runtime other than the current thread runtime.
1256 ///
1257 /// # Unstable
1258 ///
1259 /// This option is currently unstable and its implementation is
1260 /// incomplete. The API may change or be removed in the future. See
1261 /// issue [tokio-rs/tokio#4516] for more details.
1262 ///
1263 /// # Examples
1264 ///
1265 /// The following demonstrates a runtime configured to shutdown on
1266 /// panic. The first spawned task panics and results in the runtime
1267 /// shutting down. The second spawned task never has a chance to
1268 /// execute. The call to `block_on` will panic due to the runtime being
1269 /// forcibly shutdown.
1270 ///
1271 /// ```should_panic
1272 /// use tokio::runtime::{self, UnhandledPanic};
1273 ///
1274 /// # pub fn main() {
1275 /// let rt = runtime::Builder::new_current_thread()
1276 /// .unhandled_panic(UnhandledPanic::ShutdownRuntime)
1277 /// .build()
1278 /// .unwrap();
1279 ///
1280 /// rt.spawn(async { panic!("boom"); });
1281 /// rt.spawn(async {
1282 /// // This task never completes.
1283 /// });
1284 ///
1285 /// rt.block_on(async {
1286 /// // Do some work
1287 /// # loop { tokio::task::yield_now().await; }
1288 /// })
1289 /// # }
1290 /// ```
1291 ///
1292 /// [`JoinHandle`]: struct@crate::task::JoinHandle
1293 /// [tokio-rs/tokio#4516]: https://github.com/tokio-rs/tokio/issues/4516
1294 pub fn unhandled_panic(&mut self, behavior: UnhandledPanic) -> &mut Self {
1295 if !matches!(self.kind, Kind::CurrentThread) && matches!(behavior, UnhandledPanic::ShutdownRuntime) {
1296 panic!("UnhandledPanic::ShutdownRuntime is only supported in current thread runtime");
1297 }
1298
1299 self.unhandled_panic = behavior;
1300 self
1301 }
1302
1303 /// Disables the LIFO task scheduler heuristic.
1304 ///
1305 /// The multi-threaded scheduler includes a heuristic for optimizing
1306 /// message-passing patterns. This heuristic results in the **last**
1307 /// scheduled task being polled first.
1308 ///
1309 /// To implement this heuristic, each worker thread has a slot which
1310 /// holds the task that should be polled next. However, this slot cannot
1311 /// be stolen by other worker threads, which can result in lower total
1312 /// throughput when tasks tend to have longer poll times.
1313 ///
1314 /// This configuration option will disable this heuristic resulting in
1315 /// all scheduled tasks being pushed into the worker-local queue, which
1316 /// is stealable.
1317 ///
1318 /// Consider trying this option when the task "scheduled" time is high
1319 /// but the runtime is underutilized. Use [tokio-rs/tokio-metrics] to
1320 /// collect this data.
1321 ///
1322 /// # Unstable
1323 ///
1324 /// This configuration option is considered a workaround for the LIFO
1325 /// slot not being stealable. When the slot becomes stealable, we will
1326 /// revisit whether or not this option is necessary. See
1327 /// issue [tokio-rs/tokio#4941].
1328 ///
1329 /// # Examples
1330 ///
1331 /// ```
1332 /// # #[cfg(not(target_family = "wasm"))]
1333 /// # {
1334 /// use tokio::runtime;
1335 ///
1336 /// let rt = runtime::Builder::new_multi_thread()
1337 /// .disable_lifo_slot()
1338 /// .build()
1339 /// .unwrap();
1340 /// # }
1341 /// ```
1342 ///
1343 /// [tokio-rs/tokio-metrics]: https://github.com/tokio-rs/tokio-metrics
1344 /// [tokio-rs/tokio#4941]: https://github.com/tokio-rs/tokio/issues/4941
1345 pub fn disable_lifo_slot(&mut self) -> &mut Self {
1346 self.disable_lifo_slot = true;
1347 self
1348 }
1349
1350 /// Specifies the random number generation seed to use within all
1351 /// threads associated with the runtime being built.
1352 ///
1353 /// This option is intended to make certain parts of the runtime
1354 /// deterministic (e.g. the [`tokio::select!`] macro). In the case of
1355 /// [`tokio::select!`] it will ensure that the order that branches are
1356 /// polled is deterministic.
1357 ///
1358 /// In addition to the code specifying `rng_seed` and interacting with
1359 /// the runtime, the internals of Tokio and the Rust compiler may affect
1360 /// the sequences of random numbers. In order to ensure repeatable
1361 /// results, the version of Tokio, the versions of all other
1362 /// dependencies that interact with Tokio, and the Rust compiler version
1363 /// should also all remain constant.
1364 ///
1365 /// # Examples
1366 ///
1367 /// ```
1368 /// # use tokio::runtime::{self, RngSeed};
1369 /// # pub fn main() {
1370 /// let seed = RngSeed::from_bytes(b"place your seed here");
1371 /// let rt = runtime::Builder::new_current_thread()
1372 /// .rng_seed(seed)
1373 /// .build();
1374 /// # }
1375 /// ```
1376 ///
1377 /// [`tokio::select!`]: crate::select
1378 pub fn rng_seed(&mut self, seed: RngSeed) -> &mut Self {
1379 self.seed_generator = RngSeedGenerator::new(seed);
1380 self
1381 }
1382 }
1383
1384 cfg_unstable_metrics! {
1385 /// Enables tracking the distribution of task poll times.
1386 ///
1387 /// Task poll times are not instrumented by default as doing so requires
1388 /// calling [`Instant::now()`] twice per task poll, which could add
1389 /// measurable overhead. Use the [`Handle::metrics()`] to access the
1390 /// metrics data.
1391 ///
1392 /// The histogram uses fixed bucket sizes. In other words, the histogram
1393 /// buckets are not dynamic based on input values. Use the
1394 /// `metrics_poll_time_histogram` builder methods to configure the
1395 /// histogram details.
1396 ///
1397 /// By default, a linear histogram with 10 buckets each 100 microseconds wide will be used.
1398 /// This has an extremely low memory footprint, but may not provide enough granularity. For
1399 /// better granularity with low memory usage, use [`metrics_poll_time_histogram_configuration()`]
1400 /// to select [`LogHistogram`] instead.
1401 ///
1402 /// # Examples
1403 ///
1404 /// ```
1405 /// # #[cfg(not(target_family = "wasm"))]
1406 /// # {
1407 /// use tokio::runtime;
1408 ///
1409 /// let rt = runtime::Builder::new_multi_thread()
1410 /// .enable_metrics_poll_time_histogram()
1411 /// .build()
1412 /// .unwrap();
1413 /// # // Test default values here
1414 /// # fn us(n: u64) -> std::time::Duration { std::time::Duration::from_micros(n) }
1415 /// # let m = rt.handle().metrics();
1416 /// # assert_eq!(m.poll_time_histogram_num_buckets(), 10);
1417 /// # assert_eq!(m.poll_time_histogram_bucket_range(0), us(0)..us(100));
1418 /// # assert_eq!(m.poll_time_histogram_bucket_range(1), us(100)..us(200));
1419 /// # }
1420 /// ```
1421 ///
1422 /// [`Handle::metrics()`]: crate::runtime::Handle::metrics
1423 /// [`Instant::now()`]: std::time::Instant::now
1424 /// [`LogHistogram`]: crate::runtime::LogHistogram
1425 /// [`metrics_poll_time_histogram_configuration()`]: Builder::metrics_poll_time_histogram_configuration
1426 pub fn enable_metrics_poll_time_histogram(&mut self) -> &mut Self {
1427 self.metrics_poll_count_histogram_enable = true;
1428 self
1429 }
1430
1431 /// Deprecated. Use [`enable_metrics_poll_time_histogram()`] instead.
1432 ///
1433 /// [`enable_metrics_poll_time_histogram()`]: Builder::enable_metrics_poll_time_histogram
1434 #[deprecated(note = "`poll_count_histogram` related methods have been renamed `poll_time_histogram` to better reflect their functionality.")]
1435 #[doc(hidden)]
1436 pub fn enable_metrics_poll_count_histogram(&mut self) -> &mut Self {
1437 self.enable_metrics_poll_time_histogram()
1438 }
1439
1440 /// Sets the histogram scale for tracking the distribution of task poll
1441 /// times.
1442 ///
1443 /// Tracking the distribution of task poll times can be done using a
1444 /// linear or log scale. When using linear scale, each histogram bucket
1445 /// will represent the same range of poll times. When using log scale,
1446 /// each histogram bucket will cover a range twice as big as the
1447 /// previous bucket.
1448 ///
1449 /// **Default:** linear scale.
1450 ///
1451 /// # Examples
1452 ///
1453 /// ```
1454 /// # #[cfg(not(target_family = "wasm"))]
1455 /// # {
1456 /// use tokio::runtime::{self, HistogramScale};
1457 ///
1458 /// # #[allow(deprecated)]
1459 /// let rt = runtime::Builder::new_multi_thread()
1460 /// .enable_metrics_poll_time_histogram()
1461 /// .metrics_poll_count_histogram_scale(HistogramScale::Log)
1462 /// .build()
1463 /// .unwrap();
1464 /// # }
1465 /// ```
1466 #[deprecated(note = "use `metrics_poll_time_histogram_configuration`")]
1467 pub fn metrics_poll_count_histogram_scale(&mut self, histogram_scale: crate::runtime::HistogramScale) -> &mut Self {
1468 self.metrics_poll_count_histogram.legacy_mut(|b|b.scale = histogram_scale);
1469 self
1470 }
1471
1472 /// Configure the histogram for tracking poll times
1473 ///
1474 /// By default, a linear histogram with 10 buckets each 100 microseconds wide will be used.
1475 /// This has an extremely low memory footprint, but may not provide enough granularity. For
1476 /// better granularity with low memory usage, use [`LogHistogram`] instead.
1477 ///
1478 /// # Examples
1479 /// Configure a [`LogHistogram`] with [default configuration]:
1480 /// ```
1481 /// # #[cfg(not(target_family = "wasm"))]
1482 /// # {
1483 /// use tokio::runtime;
1484 /// use tokio::runtime::{HistogramConfiguration, LogHistogram};
1485 ///
1486 /// let rt = runtime::Builder::new_multi_thread()
1487 /// .enable_metrics_poll_time_histogram()
1488 /// .metrics_poll_time_histogram_configuration(
1489 /// HistogramConfiguration::log(LogHistogram::default())
1490 /// )
1491 /// .build()
1492 /// .unwrap();
1493 /// # }
1494 /// ```
1495 ///
1496 /// Configure a linear histogram with 100 buckets, each 10μs wide
1497 /// ```
1498 /// # #[cfg(not(target_family = "wasm"))]
1499 /// # {
1500 /// use tokio::runtime;
1501 /// use std::time::Duration;
1502 /// use tokio::runtime::HistogramConfiguration;
1503 ///
1504 /// let rt = runtime::Builder::new_multi_thread()
1505 /// .enable_metrics_poll_time_histogram()
1506 /// .metrics_poll_time_histogram_configuration(
1507 /// HistogramConfiguration::linear(Duration::from_micros(10), 100)
1508 /// )
1509 /// .build()
1510 /// .unwrap();
1511 /// # }
1512 /// ```
1513 ///
1514 /// Configure a [`LogHistogram`] with the following settings:
1515 /// - Measure times from 100ns to 120s
1516 /// - Max error of 0.1
1517 /// - No more than 1024 buckets
1518 /// ```
1519 /// # #[cfg(not(target_family = "wasm"))]
1520 /// # {
1521 /// use std::time::Duration;
1522 /// use tokio::runtime;
1523 /// use tokio::runtime::{HistogramConfiguration, LogHistogram};
1524 ///
1525 /// let rt = runtime::Builder::new_multi_thread()
1526 /// .enable_metrics_poll_time_histogram()
1527 /// .metrics_poll_time_histogram_configuration(
1528 /// HistogramConfiguration::log(LogHistogram::builder()
1529 /// .max_value(Duration::from_secs(120))
1530 /// .min_value(Duration::from_nanos(100))
1531 /// .max_error(0.1)
1532 /// .max_buckets(1024)
1533 /// .expect("configuration uses 488 buckets")
1534 /// )
1535 /// )
1536 /// .build()
1537 /// .unwrap();
1538 /// # }
1539 /// ```
1540 ///
1541 /// When migrating from the legacy histogram ([`HistogramScale::Log`]) and wanting
1542 /// to match the previous behavior, use `precision_exact(0)`. This creates a histogram
1543 /// where each bucket is twice the size of the previous bucket.
1544 /// ```rust
1545 /// use std::time::Duration;
1546 /// use tokio::runtime::{HistogramConfiguration, LogHistogram};
1547 /// let rt = tokio::runtime::Builder::new_current_thread()
1548 /// .enable_all()
1549 /// .enable_metrics_poll_time_histogram()
1550 /// .metrics_poll_time_histogram_configuration(HistogramConfiguration::log(
1551 /// LogHistogram::builder()
1552 /// .min_value(Duration::from_micros(20))
1553 /// .max_value(Duration::from_millis(4))
1554 /// // Set `precision_exact` to `0` to match `HistogramScale::Log`
1555 /// .precision_exact(0)
1556 /// .max_buckets(10)
1557 /// .unwrap(),
1558 /// ))
1559 /// .build()
1560 /// .unwrap();
1561 /// ```
1562 ///
1563 /// [`LogHistogram`]: crate::runtime::LogHistogram
1564 /// [default configuration]: crate::runtime::LogHistogramBuilder
1565 /// [`HistogramScale::Log`]: crate::runtime::HistogramScale::Log
1566 pub fn metrics_poll_time_histogram_configuration(&mut self, configuration: HistogramConfiguration) -> &mut Self {
1567 self.metrics_poll_count_histogram.histogram_type = configuration.inner;
1568 self
1569 }
1570
1571 /// Sets the histogram resolution for tracking the distribution of task
1572 /// poll times.
1573 ///
1574 /// The resolution is the histogram's first bucket's range. When using a
1575 /// linear histogram scale, each bucket will cover the same range. When
1576 /// using a log scale, each bucket will cover a range twice as big as
1577 /// the previous bucket. In the log case, the resolution represents the
1578 /// smallest bucket range.
1579 ///
1580 /// Note that, when using log scale, the resolution is rounded up to the
1581 /// nearest power of 2 in nanoseconds.
1582 ///
1583 /// **Default:** 100 microseconds.
1584 ///
1585 /// # Examples
1586 ///
1587 /// ```
1588 /// # #[cfg(not(target_family = "wasm"))]
1589 /// # {
1590 /// use tokio::runtime;
1591 /// use std::time::Duration;
1592 ///
1593 /// # #[allow(deprecated)]
1594 /// let rt = runtime::Builder::new_multi_thread()
1595 /// .enable_metrics_poll_time_histogram()
1596 /// .metrics_poll_count_histogram_resolution(Duration::from_micros(100))
1597 /// .build()
1598 /// .unwrap();
1599 /// # }
1600 /// ```
1601 #[deprecated(note = "use `metrics_poll_time_histogram_configuration`")]
1602 pub fn metrics_poll_count_histogram_resolution(&mut self, resolution: Duration) -> &mut Self {
1603 assert!(resolution > Duration::from_secs(0));
1604 // Sanity check the argument and also make the cast below safe.
1605 assert!(resolution <= Duration::from_secs(1));
1606
1607 let resolution = resolution.as_nanos() as u64;
1608
1609 self.metrics_poll_count_histogram.legacy_mut(|b|b.resolution = resolution);
1610 self
1611 }
1612
1613 /// Sets the number of buckets for the histogram tracking the
1614 /// distribution of task poll times.
1615 ///
1616 /// The last bucket tracks all greater values that fall out of other
1617 /// ranges. So, configuring the histogram using a linear scale,
1618 /// resolution of 50ms, and 10 buckets, the 10th bucket will track task
1619 /// polls that take more than 450ms to complete.
1620 ///
1621 /// **Default:** 10
1622 ///
1623 /// # Examples
1624 ///
1625 /// ```
1626 /// # #[cfg(not(target_family = "wasm"))]
1627 /// # {
1628 /// use tokio::runtime;
1629 ///
1630 /// # #[allow(deprecated)]
1631 /// let rt = runtime::Builder::new_multi_thread()
1632 /// .enable_metrics_poll_time_histogram()
1633 /// .metrics_poll_count_histogram_buckets(15)
1634 /// .build()
1635 /// .unwrap();
1636 /// # }
1637 /// ```
1638 #[deprecated(note = "use `metrics_poll_time_histogram_configuration`")]
1639 pub fn metrics_poll_count_histogram_buckets(&mut self, buckets: usize) -> &mut Self {
1640 self.metrics_poll_count_histogram.legacy_mut(|b|b.num_buckets = buckets);
1641 self
1642 }
1643 }
1644
1645 fn build_current_thread_runtime(&mut self) -> io::Result<Runtime> {
1646 use crate::runtime::runtime::Scheduler;
1647
1648 let (scheduler, handle, blocking_pool) =
1649 self.build_current_thread_runtime_components(None)?;
1650
1651 Ok(Runtime::from_parts(
1652 Scheduler::CurrentThread(scheduler),
1653 handle,
1654 blocking_pool,
1655 ))
1656 }
1657
1658 fn build_current_thread_local_runtime(&mut self) -> io::Result<LocalRuntime> {
1659 use crate::runtime::local_runtime::LocalRuntimeScheduler;
1660
1661 let tid = std::thread::current().id();
1662
1663 let (scheduler, handle, blocking_pool) =
1664 self.build_current_thread_runtime_components(Some(tid))?;
1665
1666 Ok(LocalRuntime::from_parts(
1667 LocalRuntimeScheduler::CurrentThread(scheduler),
1668 handle,
1669 blocking_pool,
1670 ))
1671 }
1672
1673 fn build_current_thread_runtime_components(
1674 &mut self,
1675 local_tid: Option<ThreadId>,
1676 ) -> io::Result<(CurrentThread, Handle, BlockingPool)> {
1677 use crate::runtime::scheduler;
1678 use crate::runtime::Config;
1679
1680 let mut cfg = self.get_cfg();
1681 cfg.timer_flavor = TimerFlavor::Traditional;
1682 let (driver, driver_handle) = driver::Driver::new(cfg)?;
1683
1684 // Blocking pool
1685 let blocking_pool = blocking::create_blocking_pool(self, self.max_blocking_threads);
1686 let blocking_spawner = blocking_pool.spawner().clone();
1687
1688 // Generate a rng seed for this runtime.
1689 let seed_generator_1 = self.seed_generator.next_generator();
1690 let seed_generator_2 = self.seed_generator.next_generator();
1691
1692 // And now put a single-threaded scheduler on top of the timer. When
1693 // there are no futures ready to do something, it'll let the timer or
1694 // the reactor to generate some new stimuli for the futures to continue
1695 // in their life.
1696 let (scheduler, handle) = CurrentThread::new(
1697 driver,
1698 driver_handle,
1699 blocking_spawner,
1700 seed_generator_2,
1701 Config {
1702 before_park: self.before_park.clone(),
1703 after_unpark: self.after_unpark.clone(),
1704 before_spawn: self.before_spawn.clone(),
1705 #[cfg(tokio_unstable)]
1706 before_poll: self.before_poll.clone(),
1707 #[cfg(tokio_unstable)]
1708 after_poll: self.after_poll.clone(),
1709 after_termination: self.after_termination.clone(),
1710 global_queue_interval: self.global_queue_interval,
1711 event_interval: self.event_interval,
1712 #[cfg(tokio_unstable)]
1713 unhandled_panic: self.unhandled_panic.clone(),
1714 disable_lifo_slot: self.disable_lifo_slot,
1715 // This setting never makes sense for a current thread runtime,
1716 // as it only configures how the I/O driver is stolen across
1717 // workers.
1718 enable_eager_driver_handoff: false,
1719 seed_generator: seed_generator_1,
1720 metrics_poll_count_histogram: self.metrics_poll_count_histogram_builder(),
1721 metrics_schedule_latency_histogram: self
1722 .metrics_schedule_latency_histogram_builder(),
1723 },
1724 local_tid,
1725 self.name.clone(),
1726 );
1727
1728 let handle = Handle {
1729 inner: scheduler::Handle::CurrentThread(handle),
1730 };
1731
1732 Ok((scheduler, handle, blocking_pool))
1733 }
1734
1735 fn metrics_poll_count_histogram_builder(&self) -> Option<HistogramBuilder> {
1736 if self.metrics_poll_count_histogram_enable {
1737 Some(self.metrics_poll_count_histogram.clone())
1738 } else {
1739 None
1740 }
1741 }
1742
1743 fn metrics_schedule_latency_histogram_builder(&self) -> Option<HistogramBuilder> {
1744 if self.metrics_schedule_latency_histogram_enabled {
1745 Some(self.metrics_schedule_latency_histogram.clone())
1746 } else {
1747 None
1748 }
1749 }
1750}
1751
1752cfg_io_driver! {
1753 impl Builder {
1754 /// Enables the I/O driver.
1755 ///
1756 /// Doing this enables using net, process, signal, and some I/O types on
1757 /// the runtime.
1758 ///
1759 /// # Examples
1760 ///
1761 /// ```
1762 /// use tokio::runtime;
1763 ///
1764 /// let rt = runtime::Builder::new_multi_thread()
1765 /// .enable_io()
1766 /// .build()
1767 /// .unwrap();
1768 /// ```
1769 pub fn enable_io(&mut self) -> &mut Self {
1770 self.enable_io = true;
1771 self
1772 }
1773
1774 /// Enables the I/O driver and configures the max number of events to be
1775 /// processed per tick.
1776 ///
1777 /// # Examples
1778 ///
1779 /// ```
1780 /// use tokio::runtime;
1781 ///
1782 /// let rt = runtime::Builder::new_current_thread()
1783 /// .enable_io()
1784 /// .max_io_events_per_tick(1024)
1785 /// .build()
1786 /// .unwrap();
1787 /// ```
1788 pub fn max_io_events_per_tick(&mut self, capacity: usize) -> &mut Self {
1789 self.nevents = capacity;
1790 self
1791 }
1792 }
1793}
1794
1795cfg_time! {
1796 impl Builder {
1797 /// Enables the time driver.
1798 ///
1799 /// Doing this enables using `tokio::time` on the runtime.
1800 ///
1801 /// # Examples
1802 ///
1803 /// ```
1804 /// # #[cfg(not(target_family = "wasm"))]
1805 /// # {
1806 /// use tokio::runtime;
1807 ///
1808 /// let rt = runtime::Builder::new_multi_thread()
1809 /// .enable_time()
1810 /// .build()
1811 /// .unwrap();
1812 /// # }
1813 /// ```
1814 pub fn enable_time(&mut self) -> &mut Self {
1815 self.enable_time = true;
1816 self
1817 }
1818 }
1819}
1820
1821cfg_io_uring! {
1822 impl Builder {
1823 /// Enables the tokio's io_uring driver.
1824 ///
1825 /// Doing this enables using io_uring operations on the runtime.
1826 ///
1827 /// # Examples
1828 ///
1829 /// ```
1830 /// use tokio::runtime;
1831 ///
1832 /// let rt = runtime::Builder::new_multi_thread()
1833 /// .enable_io_uring()
1834 /// .build()
1835 /// .unwrap();
1836 /// ```
1837 #[cfg_attr(docsrs, doc(cfg(feature = "io-uring")))]
1838 pub fn enable_io_uring(&mut self) -> &mut Self {
1839 // Currently, the uring flag is equivalent to `enable_io`.
1840 self.enable_io = true;
1841 self
1842 }
1843 }
1844}
1845
1846cfg_test_util! {
1847 impl Builder {
1848 /// Controls if the runtime's clock starts paused or advancing.
1849 ///
1850 /// Pausing time requires the current-thread runtime; construction of
1851 /// the runtime will panic otherwise.
1852 ///
1853 /// # Examples
1854 ///
1855 /// ```
1856 /// use tokio::runtime;
1857 ///
1858 /// let rt = runtime::Builder::new_current_thread()
1859 /// .enable_time()
1860 /// .start_paused(true)
1861 /// .build()
1862 /// .unwrap();
1863 /// ```
1864 pub fn start_paused(&mut self, start_paused: bool) -> &mut Self {
1865 self.start_paused = start_paused;
1866 self
1867 }
1868 }
1869}
1870
1871cfg_schedule_latency! {
1872 impl Builder {
1873 /// Enables tracking the distribution of task schedule latencies. Task
1874 /// schedule latency is the time between when a task is scheduled for
1875 /// execution and when it is polled.
1876 ///
1877 /// **This feature is only supported on 64-bit targets.**
1878 ///
1879 /// Task schedule latencies are not instrumented by default as doing
1880 /// so requires calling [`Instant::now()`] when a task is scheduled
1881 /// and when it is polled, which could add measurable overhead. Use
1882 /// the [`Handle::metrics()`] to access the metrics data.
1883 ///
1884 /// By default, a linear histogram with 10 buckets each 100 microseconds wide will be used.
1885 /// This has an extremely low memory footprint, but may not provide enough granularity. For
1886 /// better granularity with low memory usage, use [`metrics_schedule_latency_histogram_configuration()`]
1887 /// to select [`LogHistogram`] instead.
1888 ///
1889 /// # Examples
1890 ///
1891 /// ```
1892 /// # #[cfg(not(target_family = "wasm"))]
1893 /// # {
1894 /// use tokio::runtime;
1895 ///
1896 /// let rt = runtime::Builder::new_multi_thread()
1897 /// .enable_metrics_schedule_latency_histogram()
1898 /// .build()
1899 /// .unwrap();
1900 /// # // Test default values here
1901 /// # fn us(n: u64) -> std::time::Duration { std::time::Duration::from_micros(n) }
1902 /// # let m = rt.handle().metrics();
1903 /// # assert_eq!(m.schedule_latency_histogram_num_buckets(), 10);
1904 /// # assert_eq!(m.schedule_latency_histogram_bucket_range(0), us(0)..us(100));
1905 /// # assert_eq!(m.schedule_latency_histogram_bucket_range(1), us(100)..us(200));
1906 /// # }
1907 /// ```
1908 ///
1909 /// [`Handle::metrics()`]: crate::runtime::Handle::metrics
1910 /// [`Instant::now()`]: std::time::Instant::now
1911 /// [`LogHistogram`]: crate::runtime::LogHistogram
1912 /// [`metrics_schedule_latency_histogram_configuration()`]: Builder::metrics_schedule_latency_histogram_configuration
1913 pub fn enable_metrics_schedule_latency_histogram(&mut self) -> &mut Self {
1914 self.metrics_schedule_latency_histogram_enabled = true;
1915 self
1916 }
1917
1918 /// Configure the histogram for tracking task schedule latencies.
1919 ///
1920 /// Tracking of task schedule latencies must be enabled with
1921 /// [`enable_metrics_schedule_latency_histogram()`] for this function
1922 /// to have any effect.
1923 ///
1924 /// By default, a linear histogram with 10 buckets each 100 microseconds wide will be used.
1925 /// This has an extremely low memory footprint, but may not provide enough granularity. For
1926 /// better granularity with low memory usage, use [`LogHistogram`] instead.
1927 ///
1928 /// # Examples
1929 /// Configure a [`LogHistogram`] with [default configuration]:
1930 /// ```
1931 /// # #[cfg(not(target_family = "wasm"))]
1932 /// # {
1933 /// use tokio::runtime;
1934 /// use tokio::runtime::{HistogramConfiguration, LogHistogram};
1935 ///
1936 /// let rt = runtime::Builder::new_multi_thread()
1937 /// .enable_metrics_schedule_latency_histogram()
1938 /// .metrics_schedule_latency_histogram_configuration(
1939 /// HistogramConfiguration::log(LogHistogram::default())
1940 /// )
1941 /// .build()
1942 /// .unwrap();
1943 /// # }
1944 /// ```
1945 ///
1946 /// Configure a linear histogram with 100 buckets, each 10μs wide
1947 /// ```
1948 /// # #[cfg(not(target_family = "wasm"))]
1949 /// # {
1950 /// use tokio::runtime;
1951 /// use std::time::Duration;
1952 /// use tokio::runtime::HistogramConfiguration;
1953 ///
1954 /// let rt = runtime::Builder::new_multi_thread()
1955 /// .enable_metrics_schedule_latency_histogram()
1956 /// .metrics_schedule_latency_histogram_configuration(
1957 /// HistogramConfiguration::linear(Duration::from_micros(10), 100)
1958 /// )
1959 /// .build()
1960 /// .unwrap();
1961 /// # }
1962 /// ```
1963 ///
1964 /// Configure a [`LogHistogram`] with the following settings:
1965 /// - Measure times from 100ns to 120s
1966 /// - Max error of 0.1
1967 /// - No more than 1024 buckets
1968 /// ```
1969 /// # #[cfg(not(target_family = "wasm"))]
1970 /// # {
1971 /// use std::time::Duration;
1972 /// use tokio::runtime;
1973 /// use tokio::runtime::{HistogramConfiguration, LogHistogram};
1974 ///
1975 /// let rt = runtime::Builder::new_multi_thread()
1976 /// .enable_metrics_schedule_latency_histogram()
1977 /// .metrics_schedule_latency_histogram_configuration(
1978 /// HistogramConfiguration::log(LogHistogram::builder()
1979 /// .max_value(Duration::from_secs(120))
1980 /// .min_value(Duration::from_nanos(100))
1981 /// .max_error(0.1)
1982 /// .max_buckets(1024)
1983 /// .expect("configuration uses 488 buckets")
1984 /// )
1985 /// )
1986 /// .build()
1987 /// .unwrap();
1988 /// # }
1989 /// ```
1990 ///
1991 /// [`LogHistogram`]: crate::runtime::LogHistogram
1992 /// [`enable_metrics_schedule_latency_histogram()`]: Builder::enable_metrics_schedule_latency_histogram
1993 pub fn metrics_schedule_latency_histogram_configuration(&mut self, configuration: HistogramConfiguration) -> &mut Self {
1994 self.metrics_schedule_latency_histogram.histogram_type = configuration.inner;
1995 self
1996 }
1997 }
1998}
1999
2000cfg_rt_multi_thread! {
2001 impl Builder {
2002 fn build_threaded_runtime(&mut self) -> io::Result<Runtime> {
2003 use crate::loom::sys::num_cpus;
2004 use crate::runtime::{Config, runtime::Scheduler};
2005 use crate::runtime::scheduler::{self, MultiThread};
2006
2007 let worker_threads = self.worker_threads.unwrap_or_else(num_cpus);
2008
2009 let (driver, driver_handle) = driver::Driver::new(self.get_cfg())?;
2010
2011 // Create the blocking pool
2012 let blocking_pool =
2013 blocking::create_blocking_pool(self, self.max_blocking_threads + worker_threads);
2014 let blocking_spawner = blocking_pool.spawner().clone();
2015
2016 // Generate a rng seed for this runtime.
2017 let seed_generator_1 = self.seed_generator.next_generator();
2018 let seed_generator_2 = self.seed_generator.next_generator();
2019
2020 let (scheduler, handle, launch) = MultiThread::new(
2021 worker_threads,
2022 driver,
2023 driver_handle,
2024 blocking_spawner,
2025 seed_generator_2,
2026 Config {
2027 before_park: self.before_park.clone(),
2028 after_unpark: self.after_unpark.clone(),
2029 before_spawn: self.before_spawn.clone(),
2030 #[cfg(tokio_unstable)]
2031 before_poll: self.before_poll.clone(),
2032 #[cfg(tokio_unstable)]
2033 after_poll: self.after_poll.clone(),
2034 after_termination: self.after_termination.clone(),
2035 global_queue_interval: self.global_queue_interval,
2036 event_interval: self.event_interval,
2037 #[cfg(tokio_unstable)]
2038 unhandled_panic: self.unhandled_panic.clone(),
2039 disable_lifo_slot: self.disable_lifo_slot,
2040 enable_eager_driver_handoff: self.enable_eager_driver_handoff,
2041 seed_generator: seed_generator_1,
2042 metrics_poll_count_histogram: self.metrics_poll_count_histogram_builder(),
2043 metrics_schedule_latency_histogram: self.metrics_schedule_latency_histogram_builder(),
2044 },
2045 self.timer_flavor,
2046 self.name.clone(),
2047 );
2048
2049 let handle = Handle { inner: scheduler::Handle::MultiThread(handle) };
2050
2051 // Spawn the thread pool workers
2052 let _enter = handle.enter();
2053 launch.launch();
2054
2055 Ok(Runtime::from_parts(Scheduler::MultiThread(scheduler), handle, blocking_pool))
2056 }
2057 }
2058}
2059
2060impl fmt::Debug for Builder {
2061 fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
2062 let mut debug = fmt.debug_struct("Builder");
2063
2064 if let Some(name) = &self.name {
2065 debug.field("name", name);
2066 }
2067
2068 debug
2069 .field("worker_threads", &self.worker_threads)
2070 .field("max_blocking_threads", &self.max_blocking_threads)
2071 .field(
2072 "thread_name",
2073 &"<dyn Fn() -> String + Send + Sync + 'static>",
2074 )
2075 .field("thread_stack_size", &self.thread_stack_size)
2076 .field("after_start", &self.after_start.as_ref().map(|_| "..."))
2077 .field("before_stop", &self.before_stop.as_ref().map(|_| "..."))
2078 .field("before_park", &self.before_park.as_ref().map(|_| "..."))
2079 .field("after_unpark", &self.after_unpark.as_ref().map(|_| "..."))
2080 .field(
2081 "enable_eager_driver_handoff",
2082 &self.enable_eager_driver_handoff,
2083 );
2084
2085 if self.name.is_none() {
2086 debug.finish_non_exhaustive()
2087 } else {
2088 debug.finish()
2089 }
2090 }
2091}