rt: add unstable option to disable the LIFO slot

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.

tokio/src/runtime/builder.rs

@@ -87,6 +87,11 @@ pub struct Builder {
 
     #[cfg(tokio_unstable)]
     pub(super) unhandled_panic: UnhandledPanic,
+
+    /// 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_unstable! {
@@ -252,6 +257,8 @@ impl Builder {
 
             #[cfg(tokio_unstable)]
             unhandled_panic: UnhandledPanic::Ignore,
+
+            disable_lifo_slot: false,
         }
     }
 
@@ -781,6 +788,37 @@ 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.
+        ///
+        /// Eventually, the LIFO slot will become stealable and this option will
+        /// probably go away.
+        ///
+        /// ```
+        /// use tokio::runtime;
+        ///
+        /// let rt = runtime::Builder::new_multi_threaded()
+        ///     .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 +852,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 +971,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

@@ -16,4 +16,13 @@ pub(crate) struct Config {
     #[cfg(tokio_unstable)]
     /// How to respond to unhandled task panics.
     pub(crate) unhandled_panic: crate::runtime::UnhandledPanic,
+
+    /// 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,
 }

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