rt: add unstable option to disable the LIFO slot

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();
+        })
+    }
+}