wip

tokio/src/runtime/defer.rs

@@ -25,10 +25,14 @@ impl Defer {
         self.deferred.is_empty()
     }
 
-    pub(crate) fn wake(&mut self) {
+    pub(crate) fn wake(&mut self) -> usize {
+        let ret = self.deferred.len();
+
         for waker in self.deferred.drain(..) {
             waker.wake();
         }
+
+        ret
     }
 
     #[cfg(tokio_taskdump)]

tokio/src/runtime/scheduler/multi_thread/counters.rs

@@ -0,0 +1,69 @@
+#[cfg(tokio_internal_mt_counters)]
+mod imp {
+    use std::sync::atomic::AtomicUsize;
+    use std::sync::atomic::Ordering::Relaxed;
+
+    static NUM_MAINTENANCE: AtomicUsize = AtomicUsize::new(0);
+    static NUM_NOTIFY_LOCAL: AtomicUsize = AtomicUsize::new(0);
+    static NUM_UNPARKS_LOCAL: AtomicUsize = AtomicUsize::new(0);
+    static NUM_NEED_SEARCHERS: AtomicUsize = AtomicUsize::new(0);
+    static NUM_WAKE_DEFERS: AtomicUsize = AtomicUsize::new(0);
+    static NUM_WAKE_DEFERS_MULT: AtomicUsize = AtomicUsize::new(0);
+
+    impl Drop for super::Counters {
+        fn drop(&mut self) {
+            let notifies_local = NUM_NOTIFY_LOCAL.load(Relaxed);
+            let unparks_local = NUM_UNPARKS_LOCAL.load(Relaxed);
+            let maintenance = NUM_MAINTENANCE.load(Relaxed);
+            let need_searchers = NUM_NEED_SEARCHERS.load(Relaxed);
+            let defers = NUM_WAKE_DEFERS.load(Relaxed);
+            let defers_mult = NUM_WAKE_DEFERS_MULT.load(Relaxed);
+
+            println!("---");
+            println!("notifies (local): {}", notifies_local);
+            println!(" unparks (local): {}", unparks_local);
+            println!("     maintenance: {}", maintenance);
+            println!("  need_searchers: {}", need_searchers);
+            println!("   waking defers: {}", defers);
+            println!("          (mult): {}", defers_mult);
+        }
+    }
+
+    pub(crate) fn inc_num_inc_notify_local() {
+        NUM_NOTIFY_LOCAL.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_unparks_local() {
+        NUM_UNPARKS_LOCAL.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_maintenance() {
+        NUM_MAINTENANCE.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_need_searchers() {
+        NUM_NEED_SEARCHERS.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_defers(batch: usize) {
+        NUM_WAKE_DEFERS.fetch_add(1, Relaxed);
+
+        if batch > 1 {
+            NUM_WAKE_DEFERS_MULT.fetch_add(1, Relaxed);
+        }
+    }
+}
+
+#[cfg(not(tokio_internal_mt_counters))]
+mod imp {
+    pub(crate) fn inc_num_inc_notify_local() {}
+    pub(crate) fn inc_num_unparks_local() {}
+    pub(crate) fn inc_num_maintenance() {}
+    pub(crate) fn inc_num_need_searchers() {}
+    pub(crate) fn inc_num_defers(_batch: usize) {}
+}
+
+#[derive(Debug)]
+pub(crate) struct Counters;
+
+pub(super) use imp::*;

tokio/src/runtime/scheduler/multi_thread/idle.rs

@@ -1,17 +1,19 @@
 //! Coordinates idling workers
 
-use crate::loom::sync::atomic::AtomicUsize;
+use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
 use crate::loom::sync::Mutex;
 
-use std::fmt;
-use std::sync::atomic::Ordering::{self, SeqCst};
+use std::sync::atomic::Ordering::{AcqRel, Acquire, Release};
 
 pub(super) struct Idle {
-    /// Tracks both the number of searching workers and the number of unparked
-    /// workers.
-    ///
-    /// Used as a fast-path to avoid acquiring the lock when needed.
-    state: AtomicUsize,
+    /// Number of searching workers
+    num_searching: AtomicUsize,
+
+    /// Number of sleeping workers
+    num_sleeping: AtomicUsize,
+
+    /// Used to catch false-negatives when notifying workers.
+    needs_searching: AtomicBool,
 
     /// Sleeping workers
     sleepers: Mutex<Vec<usize>>,
@@ -20,19 +22,15 @@ pub(super) struct Idle {
     num_workers: usize,
 }
 
-const UNPARK_SHIFT: usize = 16;
-const UNPARK_MASK: usize = !SEARCH_MASK;
-const SEARCH_MASK: usize = (1 << UNPARK_SHIFT) - 1;
-
 #[derive(Copy, Clone)]
 struct State(usize);
 
 impl Idle {
     pub(super) fn new(num_workers: usize) -> Idle {
-        let init = State::new(num_workers);
-
         Idle {
-            state: AtomicUsize::new(init.into()),
+            num_searching: AtomicUsize::new(0),
+            num_sleeping: AtomicUsize::new(0),
+            needs_searching: AtomicBool::new(false),
             sleepers: Mutex::new(Vec::with_capacity(num_workers)),
             num_workers,
         }
@@ -40,73 +38,124 @@ impl Idle {
 
     /// If there are no workers actively searching, returns the index of a
     /// worker currently sleeping.
-    pub(super) fn worker_to_notify(&self) -> Option<usize> {
-        // If at least one worker is spinning, work being notified will
-        // eventually be found. A searching thread will find **some** work and
-        // notify another worker, eventually leading to our work being found.
+    pub(super) fn worker_to_notify_local(&self) -> Option<usize> {
+        // Because this is only called from a worker thread, we can be more
+        // relaxed here. If we get false-negatives, the caller will eventually
+        // process the work.
         //
-        // For this to happen, this load must happen before the thread
-        // transitioning `num_searching` to zero. Acquire / Release does not
-        // provide sufficient guarantees, so this load is done with `SeqCst` and
-        // will pair with the `fetch_sub(1)` when transitioning out of
-        // searching.
-        if !self.notify_should_wakeup() {
+        // Before attempting the CAS (which is expensive), do a load which is cheap.
+        if self.num_searching.load(Acquire) != 0 {
             return None;
         }
 
-        // Acquire the lock
-        let mut sleepers = self.sleepers.lock();
-
-        // Check again, now that the lock is acquired
-        if !self.notify_should_wakeup() {
+        if self
+            .num_searching
+            .compare_exchange(0, 1, Acquire, Acquire)
+            .is_err()
+        {
             return None;
         }
 
-        // A worker should be woken up, atomically increment the number of
-        // searching workers as well as the number of unparked workers.
-        State::unpark_one(&self.state, 1);
+        // We will notify a worker.
+        let mut sleepers = self.sleepers.lock();
 
-        // Get the worker to unpark
-        let ret = sleepers.pop();
-        debug_assert!(ret.is_some());
+        if let Some(ret) = sleepers.pop() {
+            self.num_sleeping
+                .store(self.num_sleeping.load(Acquire) - 1, Release);
+            return Some(ret);
+        }
 
-        ret
+        self.num_searching.fetch_sub(1, Release);
+        None
     }
 
-    /// Returns `true` if the worker needs to do a final check for submitted
-    /// work.
-    pub(super) fn transition_worker_to_parked(&self, worker: usize, is_searching: bool) -> bool {
+    pub(super) fn worker_to_notify_remote(&self) -> Option<usize> {
+        // Because this function is called from *outside* the runtime, we need
+        // to be more aggressive with our synchronization. We need to create a
+        // barrier between pushing a task into the queue (done right before
+        // calling this method) and ensuring there is at least one spinning
+        // thread. A load is not sufficient, we must also write to create the
+        // release relationship.
+        if self.num_searching.fetch_add(0, AcqRel) != 0 {
+            return None;
+        }
+
+        // We just created the release ordering and also noticed there are no
+        // searchers. Try incrementing the number of searches.
+        if self
+            .num_searching
+            .compare_exchange(0, 1, AcqRel, Acquire)
+            .is_err()
+        {
+            // A worker started searching, because of the ordering set by
+            // `fetch_add` we don't need to do anything else
+            return None;
+        }
+
+        // We will notify a worker.
+        let mut sleepers = self.sleepers.lock();
+
+        if let Some(ret) = sleepers.pop() {
+            self.num_sleeping
+                .store(self.num_sleeping.load(Acquire) - 1, Release);
+            return Some(ret);
+        }
+
+        self.num_searching.fetch_sub(1, Release);
+
+        super::counters::inc_num_need_searchers();
+
+        // We failed to find a worker to wake, we need to make sure we don't lose this wake.
+        self.needs_searching.store(true, Release);
+        None
+    }
+
+    /// Returns `false` if the worker must transition back to searching
+    pub(super) fn transition_worker_to_parked(&self, worker: usize) -> bool {
         // Acquire the lock
         let mut sleepers = self.sleepers.lock();
 
-        // Decrement the number of unparked threads
-        let ret = State::dec_num_unparked(&self.state, is_searching);
-
-        // Track the sleeping worker
-        sleepers.push(worker);
-
-        ret
-    }
-
-    pub(super) fn transition_worker_to_searching(&self) -> bool {
-        let state = State::load(&self.state, SeqCst);
-        if 2 * state.num_searching() >= self.num_workers {
+        if self.needs_searching.load(Acquire) {
             return false;
         }
 
-        // It is possible for this routine to allow more than 50% of the workers
-        // to search. That is OK. Limiting searchers is only an optimization to
-        // prevent too much contention.
-        State::inc_num_searching(&self.state, SeqCst);
+        // Track the sleeping worker
+        sleepers.push(worker);
+        self.num_sleeping
+            .store(self.num_sleeping.load(Acquire) + 1, Release);
+
         true
     }
 
+    /// Returns `true` if the worker has become searching
+    pub(super) fn try_transition_worker_to_searching(&self) -> bool {
+        let num_searching = self.num_searching.load(Acquire);
+        let num_sleeping = self.num_sleeping.load(Acquire);
+
+        if 2 * num_searching >= self.num_workers - num_sleeping {
+            return false;
+        }
+
+        self.transition_worker_to_searching();
+        true
+    }
+
+    pub(super) fn transition_worker_to_searching(&self) {
+        // Because we are about to become a searching worker, we can
+        // optimistically clear the need searching flag.
+        self.needs_searching.store(false, Release);
+
+        self.num_searching.fetch_add(1, AcqRel);
+    }
+
     /// A lightweight transition from searching -> running.
     ///
     /// Returns `true` if this is the final searching worker. The caller
     /// **must** notify a new worker.
     pub(super) fn transition_worker_from_searching(&self) -> bool {
-        State::dec_num_searching(&self.state)
+        let prev = self.num_searching.fetch_sub(1, AcqRel);
+        debug_assert!(prev > 0);
+        prev == 1
     }
 
     /// Unpark a specific worker. This happens if tasks are submitted from
@@ -118,11 +167,10 @@ impl Idle {
 
         for index in 0..sleepers.len() {
             if sleepers[index] == worker_id {
+                self.num_sleeping
+                    .store(self.num_sleeping.load(Acquire) - 1, Release);
                 sleepers.swap_remove(index);
 
-                // Update the state accordingly while the lock is held.
-                State::unpark_one(&self.state, 0);
-
                 return true;
             }
         }
@@ -135,92 +183,4 @@ impl Idle {
         let sleepers = self.sleepers.lock();
         sleepers.contains(&worker_id)
     }
-
-    fn notify_should_wakeup(&self) -> bool {
-        let state = State(self.state.fetch_add(0, SeqCst));
-        state.num_searching() == 0 && state.num_unparked() < self.num_workers
-    }
-}
-
-impl State {
-    fn new(num_workers: usize) -> State {
-        // All workers start in the unparked state
-        let ret = State(num_workers << UNPARK_SHIFT);
-        debug_assert_eq!(num_workers, ret.num_unparked());
-        debug_assert_eq!(0, ret.num_searching());
-        ret
-    }
-
-    fn load(cell: &AtomicUsize, ordering: Ordering) -> State {
-        State(cell.load(ordering))
-    }
-
-    fn unpark_one(cell: &AtomicUsize, num_searching: usize) {
-        cell.fetch_add(num_searching | (1 << UNPARK_SHIFT), SeqCst);
-    }
-
-    fn inc_num_searching(cell: &AtomicUsize, ordering: Ordering) {
-        cell.fetch_add(1, ordering);
-    }
-
-    /// Returns `true` if this is the final searching worker
-    fn dec_num_searching(cell: &AtomicUsize) -> bool {
-        let state = State(cell.fetch_sub(1, SeqCst));
-        state.num_searching() == 1
-    }
-
-    /// Track a sleeping worker
-    ///
-    /// Returns `true` if this is the final searching worker.
-    fn dec_num_unparked(cell: &AtomicUsize, is_searching: bool) -> bool {
-        let mut dec = 1 << UNPARK_SHIFT;
-
-        if is_searching {
-            dec += 1;
-        }
-
-        let prev = State(cell.fetch_sub(dec, SeqCst));
-        is_searching && prev.num_searching() == 1
-    }
-
-    /// Number of workers currently searching
-    fn num_searching(self) -> usize {
-        self.0 & SEARCH_MASK
-    }
-
-    /// Number of workers currently unparked
-    fn num_unparked(self) -> usize {
-        (self.0 & UNPARK_MASK) >> UNPARK_SHIFT
-    }
-}
-
-impl From<usize> for State {
-    fn from(src: usize) -> State {
-        State(src)
-    }
-}
-
-impl From<State> for usize {
-    fn from(src: State) -> usize {
-        src.0
-    }
-}
-
-impl fmt::Debug for State {
-    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
-        fmt.debug_struct("worker::State")
-            .field("num_unparked", &self.num_unparked())
-            .field("num_searching", &self.num_searching())
-            .finish()
-    }
-}
-
-#[test]
-fn test_state() {
-    assert_eq!(0, UNPARK_MASK & SEARCH_MASK);
-    assert_eq!(0, !(UNPARK_MASK | SEARCH_MASK));
-
-    let state = State::new(10);
-    assert_eq!(10, state.num_unparked());
-    assert_eq!(0, state.num_searching());
 }

tokio/src/runtime/scheduler/multi_thread/mod.rs

@@ -1,4 +1,5 @@
 //! Multi-threaded runtime
+mod counters;
 
 mod handle;
 pub(crate) use handle::Handle;

tokio/src/runtime/scheduler/multi_thread/worker.rs

@@ -84,16 +84,12 @@ pub(super) struct Worker {
 
 /// Core data
 struct Core {
+    /// Index holding this worker's remote state
+    index: usize,
+
     /// Used to schedule bookkeeping tasks every so often.
     tick: u32,
 
-    /// When a task is scheduled from a worker, it is stored in this slot. The
-    /// worker will check this slot for a task **before** checking the run
-    /// queue. This effectively results in the **last** scheduled task to be run
-    /// next (LIFO). This is an optimization for message passing patterns and
-    /// helps to reduce latency.
-    lifo_slot: Option<Notified>,
-
     /// The worker-local run queue.
     run_queue: queue::Local<Arc<Handle>>,
 
@@ -148,10 +144,20 @@ pub(super) struct Shared {
     pub(super) scheduler_metrics: SchedulerMetrics,
 
     pub(super) worker_metrics: Box<[WorkerMetrics]>,
+
+    /// Internal-only performance counters
+    _counters: super::counters::Counters,
 }
 
 /// Used to communicate with a worker from other threads.
 struct Remote {
+    /// When a task is scheduled from a worker, it is stored in this slot. The
+    /// worker will check this slot for a task **before** checking the run
+    /// queue. This effectively results in the **last** scheduled task to be run
+    /// next (LIFO). This is an optimization for message passing patterns and
+    /// helps to reduce latency.
+    lifo_slot: Lifo,
+
     /// Steals tasks from this worker.
     steal: queue::Steal<Arc<Handle>>,
 
@@ -182,6 +188,8 @@ type Task = task::Task<Arc<Handle>>;
 /// A notified task handle
 type Notified = task::Notified<Arc<Handle>>;
 
+type Lifo = task::AtomicCell<Arc<Handle>>;
+
 // Tracks thread-local state
 scoped_thread_local!(static CURRENT: Context);
 
@@ -198,7 +206,7 @@ pub(super) fn create(
     let mut worker_metrics = Vec::with_capacity(size);
 
     // Create the local queues
-    for _ in 0..size {
+    for i in 0..size {
         let (steal, run_queue) = queue::local();
 
         let park = park.clone();
@@ -206,8 +214,8 @@ pub(super) fn create(
         let metrics = WorkerMetrics::from_config(&config);
 
         cores.push(Box::new(Core {
+            index: i,
             tick: 0,
-            lifo_slot: None,
             run_queue,
             is_searching: false,
             is_shutdown: false,
@@ -216,7 +224,11 @@ pub(super) fn create(
             rand: FastRand::new(config.seed_generator.next_seed()),
         }));
 
-        remotes.push(Remote { steal, unpark });
+        remotes.push(Remote {
+            lifo_slot: task::AtomicCell::new(),
+            steal,
+            unpark,
+        });
         worker_metrics.push(metrics);
     }
 
@@ -230,6 +242,7 @@ pub(super) fn create(
             config,
             scheduler_metrics: SchedulerMetrics::new(),
             worker_metrics: worker_metrics.into_boxed_slice(),
+            _counters: super::counters::Counters,
         },
         driver: driver_handle,
         blocking_spawner,
@@ -428,18 +441,21 @@ impl Context {
                 continue;
             }
 
-            // There is no more **local** work to process, try to steal work
-            // from other workers.
-            if let Some(task) = core.steal_work(&self.worker) {
-                core = self.run_task(task, core)?;
-            } else {
-                // Wait for work
-                core = if did_defer_tasks() {
-                    self.park_timeout(core, Some(Duration::from_millis(0)))
-                } else {
-                    self.park(core)
-                };
+            if core.transition_to_searching(&self.worker) {
+                // There is no more **local** work to process, try to steal work
+                // from other workers.
+                if let Some(task) = core.steal_work(&self.worker) {
+                    core = self.run_task(task, core)?;
+                    continue;
+                }
             }
+
+            // Wait for work
+            core = if did_defer_tasks() {
+                self.park_timeout(core, Some(Duration::from_millis(0)))
+            } else {
+                self.park(core)
+            };
         }
 
         core.pre_shutdown(&self.worker);
@@ -482,7 +498,7 @@ impl Context {
                 core.metrics.end_poll();
 
                 // Check for a task in the LIFO slot
-                let task = match core.lifo_slot.take() {
+                let task = match self.worker.lifo_slot().take_local() {
                     Some(task) => task,
                     None => return Ok(core),
                 };
@@ -512,6 +528,7 @@ impl Context {
 
     fn maintenance(&self, mut core: Box<Core>) -> Box<Core> {
         if core.tick % self.worker.handle.shared.config.event_interval == 0 {
+            super::counters::inc_num_maintenance();
             // Call `park` with a 0 timeout. This enables the I/O driver, timer, ...
             // to run without actually putting the thread to sleep.
             core = self.park_timeout(core, Some(Duration::from_millis(0)));
@@ -535,23 +552,29 @@ impl Context {
     /// Also, we rely on the workstealing algorithm to spread the tasks amongst workers
     /// after all the IOs get dispatched
     fn park(&self, mut core: Box<Core>) -> Box<Core> {
+        // First, try to transition to the parked state. If this doesn't
+        // succeed, then we abort the parking process. This transition can fail
+        // if we detect that there *may* have been a lost wakeup. See the
+        // transition fn in `idle.rs` for more details.
+        if !core.transition_to_parked(&self.worker) {
+            return core;
+        }
+
         if let Some(f) = &self.worker.handle.shared.config.before_park {
             f();
         }
 
-        if core.transition_to_parked(&self.worker) {
-            while !core.is_shutdown {
-                core.metrics.about_to_park();
-                core = self.park_timeout(core, None);
-                core.metrics.returned_from_park();
-
-                // Run regularly scheduled maintenance
-                core.maintenance(&self.worker);
-
-                if core.transition_from_parked(&self.worker) {
-                    break;
-                }
+        while !core.is_shutdown {
+            if core.transition_from_parked_if_notified(&self.worker) {
+                break;
             }
+
+            core.metrics.about_to_park();
+            core = self.park_timeout(core, None);
+            core.metrics.returned_from_park();
+
+            // Run regularly scheduled maintenance
+            core.maintenance(&self.worker);
         }
 
         if let Some(f) = &self.worker.handle.shared.config.after_unpark {
@@ -585,7 +608,8 @@ impl Context {
         // If there are tasks available to steal, but this worker is not
         // looking for tasks to steal, notify another worker.
         if !core.is_searching && core.run_queue.is_stealable() {
-            self.worker.handle.notify_parked();
+            // TODO: is this correct?
+            self.worker.handle.notify_parked_local();
         }
 
         core
@@ -601,14 +625,19 @@ impl Core {
     /// Return the next notified task available to this worker.
     fn next_task(&mut self, worker: &Worker) -> Option<Notified> {
         if self.tick % worker.handle.shared.config.global_queue_interval == 0 {
-            worker.inject().pop().or_else(|| self.next_local_task())
+            worker
+                .next_injected_task()
+                .or_else(|| self.next_local_task(&worker.handle))
         } else {
-            self.next_local_task().or_else(|| worker.inject().pop())
+            self.next_local_task(&worker.handle)
+                .or_else(|| worker.next_injected_task())
         }
     }
 
-    fn next_local_task(&mut self) -> Option<Notified> {
-        self.lifo_slot.take().or_else(|| self.run_queue.pop())
+    fn next_local_task(&mut self, handle: &Handle) -> Option<Notified> {
+        self.lifo_slot(handle)
+            .take_local()
+            .or_else(|| self.run_queue.pop())
     }
 
     /// Function responsible for stealing tasks from another worker
@@ -617,86 +646,135 @@ impl Core {
     /// a new worker will actually try to steal. The idea is to make sure not all
     /// workers will be trying to steal at the same time.
     fn steal_work(&mut self, worker: &Worker) -> Option<Notified> {
-        if !self.transition_to_searching(worker) {
-            return None;
-        }
+        #[cfg(not(loom))]
+        const ATTEMPTS: usize = 2;
 
+        #[cfg(loom)]
+        const ATTEMPTS: usize = 1;
+
+        // Number of remotes
         let num = worker.handle.shared.remotes.len();
-        // Start from a random worker
-        let start = self.rand.fastrand_n(num as u32) as usize;
 
-        for i in 0..num {
-            let i = (start + i) % num;
+        let mut backoff = 1;
 
-            // Don't steal from ourself! We know we don't have work.
-            if i == worker.index {
-                continue;
+        // Run this a few times
+        for i in 0..ATTEMPTS {
+            // Only try stealing the LIFO slot
+            let steal_lifo = i == ATTEMPTS - 1;
+
+            // Start from a random worker
+            let start = self.rand.fastrand_n(num as u32) as usize;
+
+            for i in 0..num {
+                let i = (start + i) % num;
+
+                // Don't steal from ourself! We know we don't have work.
+                if i == worker.index {
+                    continue;
+                }
+
+                let target = &worker.handle.shared.remotes[i];
+                if let Some(task) = target
+                    .steal
+                    .steal_into(&mut self.run_queue, &mut self.metrics)
+                {
+                    return Some(task);
+                }
+
+                // if steal_lifo {
+                //     // Try stealing from the LIFO slot
+                //     if let Some(task) = target.lifo_slot.take_remote() {
+                //         self.metrics.incr_steal_count(1);
+                //         self.metrics.incr_steal_operations();
+                //         return Some(task);
+                //     }
+                // }
             }
 
-            let target = &worker.handle.shared.remotes[i];
-            if let Some(task) = target
-                .steal
-                .steal_into(&mut self.run_queue, &mut self.metrics)
-            {
+            std::thread::sleep(std::time::Duration::from_micros(backoff));
+            backoff *= 2;
+
+            // Fallback on checking the global queue
+            if let Some(task) = worker.handle.shared.inject.pop() {
                 return Some(task);
             }
         }
 
-        // Fallback on checking the global queue
-        worker.handle.shared.inject.pop()
+        None
     }
 
     fn transition_to_searching(&mut self, worker: &Worker) -> bool {
         if !self.is_searching {
-            self.is_searching = worker.handle.shared.idle.transition_worker_to_searching();
+            if did_defer_tasks() {
+                return false;
+            }
+
+            self.is_searching = worker
+                .handle
+                .shared
+                .idle
+                .try_transition_worker_to_searching();
         }
 
         self.is_searching
     }
 
+    /// Called right before running a task. In this case, if this is the last
+    /// searching worker, we need to wake up another worker as there might be
+    /// other queued tasks.
     fn transition_from_searching(&mut self, worker: &Worker) {
         if !self.is_searching {
             return;
         }
 
         self.is_searching = false;
-        worker.handle.transition_worker_from_searching();
+
+        if worker.handle.shared.idle.transition_worker_from_searching() {
+            // We are the final searching worker, so notify a new one
+            worker.handle.notify_parked_local();
+        }
     }
 
     /// Prepares the worker state for parking.
     ///
     /// Returns true if the transition happened, false if there is work to do first.
     fn transition_to_parked(&mut self, worker: &Worker) -> bool {
-        // Workers should not park if they have work to do
-        if self.lifo_slot.is_some() || self.run_queue.has_tasks() {
-            return false;
+        if self.is_searching {
+            self.is_searching = false;
+
+            if worker.handle.shared.idle.transition_worker_from_searching()
+                && worker.handle.work_to_steal()
+            {
+                // Transition *back* to searching
+                self.is_searching = true;
+                worker.handle.shared.idle.transition_worker_to_searching();
+
+                return false;
+            }
         }
 
-        // When the final worker transitions **out** of searching to parked, it
-        // must check all the queues one last time in case work materialized
-        // between the last work scan and transitioning out of searching.
-        let is_last_searcher = worker
+        // When the final worker transitions **out** of searching to parked, the
+        // transition might fail. In this case, it becomes a searching worker.
+        let res = worker
             .handle
             .shared
             .idle
-            .transition_worker_to_parked(worker.index, self.is_searching);
+            .transition_worker_to_parked(worker.index);
 
-        // The worker is no longer searching. Setting this is the local cache
-        // only.
-        self.is_searching = false;
-
-        if is_last_searcher {
-            worker.handle.notify_if_work_pending();
+        if res {
+            true
+        } else {
+            self.is_searching = true;
+            worker.handle.shared.idle.transition_worker_to_searching();
+            false
         }
-
-        true
     }
 
     /// Returns `true` if the transition happened.
-    fn transition_from_parked(&mut self, worker: &Worker) -> bool {
+    fn transition_from_parked_if_notified(&mut self, worker: &Worker) -> bool {
         // If a task is in the lifo slot, then we must unpark regardless of
         // being notified
-        if self.lifo_slot.is_some() {
+        if worker.lifo_slot().is_some() || self.run_queue.has_tasks() {
             // When a worker wakes, it should only transition to the "searching"
             // state when the wake originates from another worker *or* a new task
             // is pushed. We do *not* want the worker to transition to "searching"
@@ -741,17 +819,30 @@ impl Core {
         let mut park = self.park.take().expect("park missing");
 
         // Drain the queue
-        while self.next_local_task().is_some() {}
+        while self.next_local_task(handle).is_some() {}
 
         park.shutdown(&handle.driver);
     }
+
+    fn lifo_slot<'a>(&self, handle: &'a Handle) -> &'a Lifo {
+        &handle.shared.remotes[self.index].lifo_slot
+    }
 }
 
 impl Worker {
+    fn next_injected_task(&self) -> Option<Notified> {
+        self.inject().pop()
+    }
+
     /// Returns a reference to the scheduler's injection queue.
     fn inject(&self) -> &Inject<Arc<Handle>> {
         &self.handle.shared.inject
     }
+
+    /// Returns a reference to the worker's lifo slot
+    fn lifo_slot(&self) -> &Lifo {
+        &self.handle.shared.remotes[self.index].lifo_slot
+    }
 }
 
 // TODO: Move `Handle` impls into handle.rs
@@ -786,7 +877,8 @@ impl Handle {
             // Otherwise, use the inject queue.
             self.shared.inject.push(task);
             self.shared.scheduler_metrics.inc_remote_schedule_count();
-            self.notify_parked();
+
+            self.notify_parked_remote();
         })
     }
 
@@ -797,30 +889,24 @@ 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.
-        let should_notify = if is_yield || self.shared.config.disable_lifo_slot {
+        if is_yield || self.shared.config.disable_lifo_slot {
             core.run_queue
                 .push_back(task, &self.shared.inject, &mut core.metrics);
-            true
         } else {
             // Push to the LIFO slot
-            let prev = core.lifo_slot.take();
-            let ret = prev.is_some();
+            let prev = core.lifo_slot(self).swap_local(task);
 
             if let Some(prev) = prev {
                 core.run_queue
                     .push_back(prev, &self.shared.inject, &mut core.metrics);
             }
-
-            core.lifo_slot = Some(task);
-
-            ret
         };
 
         // 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 should_notify && core.park.is_some() {
-            self.notify_parked();
+        if core.park.is_some() {
+            self.notify_parked_local();
         }
     }
 
@@ -830,8 +916,22 @@ impl Handle {
         }
     }
 
-    fn notify_parked(&self) {
-        if let Some(index) = self.shared.idle.worker_to_notify() {
+    /// A weaker notification. It is possible for there to be false-negatives (a
+    /// worker should be woken, but isn't). This is very rare and considered OK
+    /// as it is called from a worker thread which will, eventually, process the
+    /// work.
+    fn notify_parked_local(&self) {
+        super::counters::inc_num_inc_notify_local();
+
+        if let Some(index) = self.shared.idle.worker_to_notify_local() {
+            super::counters::inc_num_unparks_local();
+            self.shared.remotes[index].unpark.unpark(&self.driver);
+        }
+    }
+
+    /// A stronger notify that ensures a worker will always wake up if needed.
+    fn notify_parked_remote(&self) {
+        if let Some(index) = self.shared.idle.worker_to_notify_remote() {
             self.shared.remotes[index].unpark.unpark(&self.driver);
         }
     }
@@ -842,25 +942,14 @@ impl Handle {
         }
     }
 
-    fn notify_if_work_pending(&self) {
+    fn work_to_steal(&self) -> bool {
         for remote in &self.shared.remotes[..] {
             if !remote.steal.is_empty() {
-                self.notify_parked();
-                return;
+                return true;
             }
         }
 
-        if !self.shared.inject.is_empty() {
-            self.notify_parked();
-        }
-    }
-
-    fn transition_worker_from_searching(&self) {
-        if self.shared.idle.transition_worker_from_searching() {
-            // We are the final searching worker. Because work was found, we
-            // need to notify another worker.
-            self.notify_parked();
-        }
+        !self.shared.inject.is_empty()
     }
 
     /// Signals that a worker has observed the shutdown signal and has replaced
@@ -898,8 +987,10 @@ fn did_defer_tasks() -> bool {
     context::with_defer(|deferred| !deferred.is_empty()).unwrap()
 }
 
+/// Returns the number of deferred tasks that were woken
 fn wake_deferred_tasks() {
-    context::with_defer(|deferred| deferred.wake());
+    let n = context::with_defer(|deferred| deferred.wake()).unwrap_or(0);
+    super::counters::inc_num_defers(n);
 }
 
 cfg_metrics! {

tokio/src/runtime/task/atomic_cell.rs

@@ -0,0 +1,92 @@
+use crate::loom::sync::atomic::AtomicPtr;
+use crate::loom::sync::atomic::Ordering::{AcqRel, Acquire, Release};
+use crate::runtime::task::{Header, Notified};
+
+use std::marker::PhantomData;
+use std::ptr::{self, NonNull};
+
+pub(crate) struct AtomicCell<S> {
+    task: AtomicPtr<Header>,
+    _p: PhantomData<S>,
+}
+
+impl<S> AtomicCell<S> {
+    pub(crate) fn new() -> AtomicCell<S> {
+        AtomicCell {
+            task: AtomicPtr::default(),
+            _p: PhantomData,
+        }
+    }
+
+    /// Should be called from a local context
+    pub(crate) fn is_some(&self) -> bool {
+        !self.task.load(Acquire).is_null()
+    }
+
+    pub(crate) fn take_local(&self) -> Option<Notified<S>> {
+        let ptr = self.task.load(Acquire);
+
+        if ptr.is_null() {
+            return None;
+        }
+
+        if self
+            .task
+            .compare_exchange(ptr, ptr::null_mut(), AcqRel, Acquire)
+            .is_err()
+        {
+            return None;
+        }
+
+        NonNull::new(ptr).map(|ptr| unsafe { Notified::from_raw(ptr) })
+    }
+
+    pub(crate) fn swap_local(&self, task: Notified<S>) -> Option<Notified<S>> {
+        let next = task.into_raw().as_ptr();
+        let prev = self.task.load(Acquire);
+
+        if prev.is_null() {
+            // Since this method is only called from the only thread that can
+            // set the value to !null, it is safe to use a store here.
+            self.task.store(next, Release);
+            return None;
+        }
+
+        if self
+            .task
+            .compare_exchange(prev, next, AcqRel, Acquire)
+            .is_ok()
+        {
+            // Safety: we already checked !null above
+            let prev = unsafe { Notified::from_raw(NonNull::new_unchecked(prev)) };
+            return Some(prev);
+        }
+
+        // The compare-exchanged failed, but there is no need to try again since
+        // this is the only thread that could set the cell to !null.
+        self.task.store(next, Release);
+        None
+    }
+
+    pub(crate) fn take_remote(&self) -> Option<Notified<S>> {
+        let task = self.task.load(Acquire);
+
+        if task.is_null() {
+            return None;
+        }
+
+        std::thread::sleep(std::time::Duration::from_micros(3));
+
+        // Try to take it once
+        if self
+            .task
+            .compare_exchange(task, ptr::null_mut(), Acquire, Acquire)
+            .is_ok()
+        {
+            // safety: we checked for null above
+            return Some(unsafe { Notified::from_raw(NonNull::new_unchecked(task)) });
+        }
+
+        None
+    }
+}

tokio/src/runtime/task/mod.rs

@@ -183,6 +183,9 @@ mod id;
 pub use id::{id, try_id, Id};
 
 cfg_rt_multi_thread! {
+    mod atomic_cell;
+    pub(super) use atomic_cell::AtomicCell;
+
     mod inject;
     pub(super) use self::inject::Inject;
 }