rt: add unstable option to disable the LIFO slot (#4936)

The multi-threaded scheduler includes a per-worker LIFO slot to
store the last scheduled task. This can improve certain usage patterns,
especially message passing between tasks. However, this LIFO slot is not
currently stealable.

Eventually, the LIFO slot **will** become stealable. However, as a
stop-gap, this unstable option lets users disable the LIFO task when
doing so improves their application's overall performance.

Refs: #4941

tokio/src/runtime/builder.rs

@@ -85,6 +85,11 @@ pub struct Builder {
     /// How many ticks before yielding to the driver for timer and I/O events?
     pub(super) event_interval: u32,
 
+    /// When true, the multi-threade scheduler LIFO slot should not be used.
+    ///
+    /// This option should only be exposed as unstable.
+    pub(super) disable_lifo_slot: bool,
+
     #[cfg(tokio_unstable)]
     pub(super) unhandled_panic: UnhandledPanic,
 }
@@ -252,6 +257,8 @@ impl Builder {
 
             #[cfg(tokio_unstable)]
             unhandled_panic: UnhandledPanic::Ignore,
+
+            disable_lifo_slot: false,
         }
     }
 
@@ -781,6 +788,47 @@ impl Builder {
             self.unhandled_panic = behavior;
             self
         }
+
+        /// Disables the LIFO task scheduler heuristic.
+        ///
+        /// The multi-threaded scheduler includes a heuristic for optimizing
+        /// message-passing patterns. This heuristic results in the **last**
+        /// scheduled task being polled first.
+        ///
+        /// To implement this heuristic, each worker thread has a slot which
+        /// holds the task that should be polled next. However, this slot cannot
+        /// be stolen by other worker threads, which can result in lower total
+        /// throughput when tasks tend to have longer poll times.
+        ///
+        /// This configuration option will disable this heuristic resulting in
+        /// all scheduled tasks being pushed into the worker-local queue, which
+        /// is stealable.
+        ///
+        /// Consider trying this option when the task "scheduled" time is high
+        /// but the runtime is underutilized. Use tokio-rs/tokio-metrics to
+        /// collect this data.
+        ///
+        /// # Unstable
+        ///
+        /// This configuration option is considered a workaround for the LIFO
+        /// slot not being stealable. When the slot becomes stealable, we will
+        /// revisit whther or not this option is necessary. See
+        /// tokio-rs/tokio#4941.
+        ///
+        /// # Examples
+        ///
+        /// ```
+        /// use tokio::runtime;
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .disable_lifo_slot()
+        ///     .build()
+        ///     .unwrap();
+        /// ```
+        pub fn disable_lifo_slot(&mut self) -> &mut Self {
+            self.disable_lifo_slot = true;
+            self
+        }
     }
 
     fn build_basic_runtime(&mut self) -> io::Result<Runtime> {
@@ -814,6 +862,7 @@ impl Builder {
                 event_interval: self.event_interval,
                 #[cfg(tokio_unstable)]
                 unhandled_panic: self.unhandled_panic.clone(),
+                disable_lifo_slot: self.disable_lifo_slot,
             },
         );
         let spawner = Spawner::Basic(scheduler.spawner().clone());
@@ -932,6 +981,7 @@ cfg_rt_multi_thread! {
                     event_interval: self.event_interval,
                     #[cfg(tokio_unstable)]
                     unhandled_panic: self.unhandled_panic.clone(),
+                    disable_lifo_slot: self.disable_lifo_slot,
                 },
             );
             let spawner = Spawner::ThreadPool(scheduler.spawner().clone());

tokio/src/runtime/config.rs

@@ -1,3 +1,4 @@
+#![cfg_attr(any(not(feature = "full"), tokio_wasm), allow(dead_code))]
 use crate::runtime::Callback;
 
 pub(crate) struct Config {
@@ -13,6 +14,15 @@ pub(crate) struct Config {
     /// Callback for a worker unparking itself
     pub(crate) after_unpark: Option<Callback>,
 
+    /// The multi-threaded scheduler includes a per-worker LIFO slot used to
+    /// store the last scheduled task. This can improve certain usage patterns,
+    /// especially message passing between tasks. However, this LIFO slot is not
+    /// currently stealable.
+    ///
+    /// Eventually, the LIFO slot **will** become stealable, however as a
+    /// stop-gap, this unstable option lets users disable the LIFO task.
+    pub(crate) disable_lifo_slot: bool,
+
     #[cfg(tokio_unstable)]
     /// How to respond to unhandled task panics.
     pub(crate) unhandled_panic: crate::runtime::UnhandledPanic,

tokio/src/runtime/thread_pool/worker.rs

@@ -758,7 +758,7 @@ impl Shared {
         // 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.
-        let should_notify = if is_yield {
+        let should_notify = if is_yield || self.config.disable_lifo_slot {
             core.run_queue
                 .push_back(task, &self.inject, &mut core.metrics);
             true

tokio/tests/rt_threaded.rs

@@ -542,3 +542,27 @@ async fn test_block_in_place4() {
 fn rt() -> runtime::Runtime {
     runtime::Runtime::new().unwrap()
 }
+
+#[cfg(tokio_unstable)]
+mod unstable {
+    use super::*;
+
+    #[test]
+    fn test_disable_lifo_slot() {
+        let rt = runtime::Builder::new_multi_thread()
+            .disable_lifo_slot()
+            .worker_threads(2)
+            .build()
+            .unwrap();
+
+        rt.block_on(async {
+            tokio::spawn(async {
+                // Spawn another task and block the thread until completion. If the LIFO slot
+                // is used then the test doesn't complete.
+                futures::executor::block_on(tokio::spawn(async {})).unwrap();
+            })
+            .await
+            .unwrap();
+        })
+    }
+}