diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs index 6032f2c60..6532a60d3 100644 --- a/tokio/src/runtime/builder.rs +++ b/tokio/src/runtime/builder.rs @@ -143,9 +143,8 @@ pub struct Builder { timer_flavor: TimerFlavor, - /// Whether or not to enable eager hand-off for the I/O and time drivers (in - /// `tokio_unstable`). - enable_eager_driver_handoff: bool, + #[cfg(feature = "rt-multi-thread")] + worker_thread_unparking_mode: UnparkingMode, } cfg_unstable! { @@ -232,6 +231,84 @@ cfg_unstable! { } } +/// Configures the multi-threaded runtime's policy for unparking worker +/// threads. +/// +/// Instances of `UnparkingMode` are passed to +/// [`Builder::worker_thread_unparking_mode`] to configure the runtime +/// behavior for when worker threads are unparked. +/// +/// See the individual variants of this enum for more details on each +/// unparking mode's behavior. +#[cfg(feature = "rt-multi-thread")] +#[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))] +#[cfg_attr(not(tokio_unstable), allow(unreachable_pub))] +#[derive(Debug, Default, Clone)] +#[non_exhaustive] +pub enum UnparkingMode { + /// Traditional pre-Tokio 1.51.0 unparking behavior. + /// + /// This mode attempts to minimize unnecessary cross-thread wakeups. When + /// the traditional unparking behavior is selected, a parked worker thread + /// will be woken when A task running on a worker thread notifies another + /// task, and it is pushed to that worker thread's local queue (see + /// [here][rt-behavior] for details). If the [LIFO slot optimization][lifo] + /// is enabled, pushing a task to an *empty* LIFO slot will not notify + /// another worker thread. If the LIFO slot already contains a task when + /// another task is woken, the previous LIFO task is pushed to the worker's + /// local queue, which will unpark a parked worker thread. + /// + /// This unparking mode reduces the overhead of cross-thread notifications, + /// which generally results in better throughput for applications under + /// heavy load. However, it is more susceptible to latency bubbles when + /// tasks may become CPU-bound for long periods of time, and can deadlock + /// if a task blocks the thread indefinitely. + /// + /// [rt-behavior]: crate::runtime#multi-threaded-runtime-behavior-at-the-time-of-writing + /// [lifo]: crate::runtime::Builder::disable_lifo_slot + #[default] + Traditional, + /// Experimental unparking behavior that reduces the risk of deadlocks and + /// latency bubbles, at the expense of increased cross-thread notification + /// overhead. + /// + /// In contrast to the [`Traditional`](Self::Traditional) unparking mode, + /// this mode unparks worker threads more aggressively. This prevents tasks + /// which block their worker thread or execute large amounts of CPU-bound + /// code without yielding from starving other tasks from running, as + /// described in issues like [#4941] and [#6315]. This can increase overhead + /// due to waking parked workers in more frequently, and may not be + /// necessary in applications which never expect tasks to go CPU-bound for + /// long periods of time without yielding. + /// + /// In the `Cautious` unparking mode, a parked worker thread is unparked + /// whenever a task is notified by another task. Unlike the `Traditional` + /// unparking mode, this occurs regardless of whether or not the notified + /// task is placed in its worker thread's [LIFO slot]. This ensures that + /// there will always be another worker thread available to steal the + /// notified task, should the currently executing task block the worker for + /// a long period of time. This prevents the issue described in [#4941]. + /// + /// In addition, this mode will also unpark a parked worker thread whenever + /// a worker thread which had previously parked on the I/O or timer driver + /// transitions to begin polling tasks. This ensures that another worker is + /// always available to process I/O events immediately, so that a + /// long-running task on the worker which was previously holding the I/O or + /// time driver does not preventing I/O or timer notifications from being + /// processed in a timely manner. + /// + /// Applications which are not under constant load at all times may benefit + /// from this unparking mode, especially if they anticipate that some tasks + /// may perform large amounts of CPU-bound work without yielding. + /// + /// [#4941]: https://github.com/tokio-rs/tokio/issues/4941 + /// [#6315]: https://github.com/tokio-rs/tokio/issues/6315 + /// [LIFO slot]: crate::runtime::Builder::disable_lifo_slot + #[cfg(tokio_unstable)] + #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))] + Cautious, +} + pub(crate) type ThreadNameFn = std::sync::Arc String + Send + Sync + 'static>; #[derive(Clone, Copy)] @@ -339,8 +416,8 @@ impl Builder { timer_flavor: TimerFlavor::Traditional, - // Eager driver handoff is disabled by default. - enable_eager_driver_handoff: false, + #[cfg(feature = "rt-multi-thread")] + worker_thread_unparking_mode: UnparkingMode::default(), } } @@ -450,8 +527,50 @@ impl Builder { /// [unstable]: crate#unstable-features #[cfg(all(tokio_unstable, feature = "rt-multi-thread"))] #[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "rt-multi-thread"))))] + #[deprecated( + since = "1.53.0", + note = "use `worker_thread_unparking_mode(UnparkingMode::Cautious)` instead" + )] pub fn enable_eager_driver_handoff(&mut self) -> &mut Self { - self.enable_eager_driver_handoff = true; + self.worker_thread_unparking_mode(UnparkingMode::Cautious) + } + + /// Selects the [unparking behavior](UnparkingMode) used by the + /// multi-threaded runtime. This configuration determines when the runtime + /// will attempt to unpark an idle worker thread. + /// + /// This option only applies to multi-threaded runtimes. Attempting to use + /// this option with any other runtime type will have no effect. + /// + /// By default, the multi-threaded runtime will use the [`Traditional`] + /// unparking mode. This mode attempts to avoid the overhead of unnecessary + /// cross-thread notifications, which biases the runtime for improved + /// throughput in applications under high levels of load. + /// + /// Alternatively, this method may be used to select the [`Cautious`] + /// unparking mode, which unparks worker threads more aggressively. This + /// mode prevents potential deadlocks and/or latency bubbles which may occur + /// when tasks perform very large amounts of CPU-bound work without + /// yielding, or block the worker thread for other reasons. However, this + /// behavior also results in increased runtime overhead, and may not be + /// necessary in applications which are always under high load and in which + /// tasks always yield within a short period of time. + /// + /// See the documentation for the [`UnparkingMode`] type and its variants + /// for more details on the behaviors of these modes. + /// + /// **Note**: This is an [unstable API][unstable]. Eager driver hand-off is + /// an experimental feature whose behavior may be removed or changed in 1.x + /// releases. See [the documentation on unstable features][unstable] for + /// details. + /// + /// [unstable]: crate#unstable-features + /// [`Traditional`]: UnparkingMode::Traditional + /// [`Cautious`]: UnparkingMode::Cautious + #[cfg(all(tokio_unstable, feature = "rt-multi-thread"))] + #[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "rt-multi-thread"))))] + pub fn worker_thread_unparking_mode(&mut self, mode: UnparkingMode) -> &mut Self { + self.worker_thread_unparking_mode = mode; self } @@ -1708,6 +1827,9 @@ impl Builder { enable_eager_driver_handoff: false, seed_generator: seed_generator_1, metrics_poll_count_histogram: self.metrics_poll_count_histogram_builder(), + // This setting never makes sense for the current thread + // runtime, as it has no notion of "waking a parked worker". + wake_on_lifo_push: false, }, local_tid, self.name.clone(), @@ -1888,7 +2010,8 @@ cfg_rt_multi_thread! { #[cfg(tokio_unstable)] unhandled_panic: self.unhandled_panic.clone(), disable_lifo_slot: self.disable_lifo_slot, - enable_eager_driver_handoff: self.enable_eager_driver_handoff, + enable_eager_driver_handoff: self.worker_thread_unparking_mode.enable_eager_driver_handoff(), + wake_on_lifo_push: self.worker_thread_unparking_mode.wake_on_lifo_push(), seed_generator: seed_generator_1, metrics_poll_count_histogram: self.metrics_poll_count_histogram_builder(), }, @@ -1926,11 +2049,12 @@ impl fmt::Debug for Builder { .field("after_start", &self.after_start.as_ref().map(|_| "...")) .field("before_stop", &self.before_stop.as_ref().map(|_| "...")) .field("before_park", &self.before_park.as_ref().map(|_| "...")) - .field("after_unpark", &self.after_unpark.as_ref().map(|_| "...")) - .field( - "enable_eager_driver_handoff", - &self.enable_eager_driver_handoff, - ); + .field("after_unpark", &self.after_unpark.as_ref().map(|_| "...")); + #[cfg(feature = "rt-multi-thread")] + debug.field( + "worker_thread_unparking_mode", + &self.worker_thread_unparking_mode, + ); if self.name.is_none() { debug.finish_non_exhaustive() @@ -1939,3 +2063,22 @@ impl fmt::Debug for Builder { } } } + +#[cfg(feature = "rt-multi-thread")] +impl UnparkingMode { + fn wake_on_lifo_push(&self) -> bool { + match self { + Self::Traditional => false, + #[cfg(tokio_unstable)] + Self::Cautious => true, + } + } + + fn enable_eager_driver_handoff(&self) -> bool { + match self { + Self::Traditional => false, + #[cfg(tokio_unstable)] + Self::Cautious => true, + } + } +} diff --git a/tokio/src/runtime/config.rs b/tokio/src/runtime/config.rs index ad25eb326..0ff11df66 100644 --- a/tokio/src/runtime/config.rs +++ b/tokio/src/runtime/config.rs @@ -58,4 +58,10 @@ pub(crate) struct Config { /// from polling the I/O driver to polling its own tasks (requires /// `tokio_unstable`). pub(crate) enable_eager_driver_handoff: bool, + + /// If `true`, a parked worker is woken whenever a task is pushed to a + /// worker's LIFO slot, to ensure that the LIFO task is always + /// stealable.`Otherwise, pushing a task to the LIFO slot does not wake a + /// parked worker. + pub(crate) wake_on_lifo_push: bool, } diff --git a/tokio/src/runtime/mod.rs b/tokio/src/runtime/mod.rs index b4449c477..094364e49 100644 --- a/tokio/src/runtime/mod.rs +++ b/tokio/src/runtime/mod.rs @@ -575,6 +575,10 @@ cfg_rt! { pub use self::builder::UnhandledPanic; pub use crate::util::rand::RngSeed; + #[cfg(feature = "rt-multi-thread")] + #[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))] + pub use self::builder::UnparkingMode; + /// Returns the index of the current worker thread, if called from a /// runtime worker thread. /// diff --git a/tokio/src/runtime/scheduler/multi_thread/worker.rs b/tokio/src/runtime/scheduler/multi_thread/worker.rs index ab3b38c3f..c682b5206 100644 --- a/tokio/src/runtime/scheduler/multi_thread/worker.rs +++ b/tokio/src/runtime/scheduler/multi_thread/worker.rs @@ -1349,9 +1349,12 @@ impl Handle { // task must always be pushed to the back of the queue, enabling other // tasks to be executed. If **not** a yield, then there is more // flexibility and the task may go to the front of the queue. - if is_yield || !core.lifo_enabled { + let should_notify = if is_yield || !core.lifo_enabled { core.run_queue .push_back_or_overflow(task, self, &mut core.stats); + // Always notify a worker, as we have pushed to the end of the + // queue. + true } else { // Push to the LIFO slot if let Some(prev) = core.run_queue.push_lifo(task) { @@ -1359,13 +1362,18 @@ impl Handle { // to be pushed to the back of the run queue. core.run_queue .push_back_or_overflow(prev, self, &mut core.stats); + // Again, we have pushed the previous LIFO task to the end of + // the queue, so we should always notify a parked worker. + true + } else { + self.shared.config.wake_on_lifo_push } }; // Only notify if not currently parked. If `park` is `None`, then the // scheduling is from a resource driver. As notifications often come in // batches, the notification is delayed until the park is complete. - if core.park.is_some() { + if should_notify && core.park.is_some() { self.notify_parked_local(); } } diff --git a/tokio/tests/rt_threaded.rs b/tokio/tests/rt_threaded.rs index ad00abaaa..96ae86ad6 100644 --- a/tokio/tests/rt_threaded.rs +++ b/tokio/tests/rt_threaded.rs @@ -694,10 +694,12 @@ fn mutex_in_block_in_place() { // Tests that when a task is notified by another task and is placed in the LIFO // slot, and then the notifying task blocks the runtime, the notified task will -// be stolen by another worker thread. +// be stolen by another worker thread, if the `Cautious` unparking mode is +// selected. // // Integration test for: https://github.com/tokio-rs/tokio/issues/4941 #[test] +#[cfg(tokio_unstable)] fn lifo_stealable() { use std::time::Duration; @@ -726,6 +728,7 @@ fn lifo_stealable() { // there's still at least one worker free to steal the blocked task. .worker_threads(4) .enable_time() + .worker_thread_unparking_mode(runtime::UnparkingMode::Cautious) .build() .unwrap(); diff --git a/tokio/tests/rt_unstable_eager_driver_handoff.rs b/tokio/tests/rt_unstable_eager_driver_handoff.rs index d41b14cac..aa5771afe 100644 --- a/tokio/tests/rt_unstable_eager_driver_handoff.rs +++ b/tokio/tests/rt_unstable_eager_driver_handoff.rs @@ -8,17 +8,17 @@ use std::sync::mpsc::RecvTimeoutError; use std::time::Duration; -/// Test that, without `enable_eager_driver_handoff`, we can reliably reproduce +/// Test that, without the `Cautious` unparking mode, we can reliably reproduce /// a deadlock when a task blocks indefinitely. If this test fails, it means -/// that the test `eager_driver_handoff_fixes_deadlock` is not actually testing -/// a condition that can deadlock the runtime. +/// that the test `cautious_unparking_fixes_deadlock` is not actually testing a +/// condition that can deadlock the runtime. #[test] fn deadlocks_consistently() { let rt = rt_builder().build().unwrap(); assert_eq!( do_test(rt), Err(RecvTimeoutError::Timeout), - "runtime did not deadlock! the `eager_driver_handoff_fixes_deadlock` \ + "runtime did not deadlock! the `cautious_unparking_fixes_deadlock` \ test may no longer reproduce the bug it is intended to test a fix \ for!", ); @@ -26,12 +26,15 @@ fn deadlocks_consistently() { /// This is the one that actually tests whether eager driver handoff works as /// expected: it runs the same reproducer as `deadlocks_consistently` a single -/// timebut with `enable_eager_driver_handoff` enabled. If this test fails, it +/// time, but with the `Cautious` unparking mode selected. If this test fails, it /// means that the eager driver handoff fix is not working as expected. #[test] #[cfg(tokio_unstable)] -fn eager_driver_handoff_fixes_deadlock() { - let rt = rt_builder().enable_eager_driver_handoff().build().unwrap(); +fn cautious_unparking_fixes_deadlock() { + let rt = rt_builder() + .worker_thread_unparking_mode(tokio::runtime::UnparkingMode::Cautious) + .build() + .unwrap(); assert_eq!( do_test(rt), Ok(()), @@ -140,7 +143,7 @@ fn do_test(rt: tokio::runtime::Runtime) -> Result<(), RecvTimeoutError> { /// Base runtime builder for both tests in this module: two worker threads, time /// driver enabled. This returns a `Builder`, rather than a `Runtime`, so that -/// the `eager_driver_handoff_fixes_deadlock` test can configure the runtime to +/// the `cautious_unparking_fixes_deadlock` test can configure the runtime to /// enable eager driver handoff before building it. fn rt_builder() -> tokio::runtime::Builder { let mut builder = tokio::runtime::Builder::new_multi_thread();