Merge remote-tracking branch 'origin/master' into rt-threaded-rework

tokio/src/loom/std/unsafe_cell.rs

@@ -6,10 +6,12 @@ impl<T> UnsafeCell<T> {
         UnsafeCell(std::cell::UnsafeCell::new(data))
     }
 
+    #[inline(always)]
     pub(crate) fn with<R>(&self, f: impl FnOnce(*const T) -> R) -> R {
         f(self.0.get())
     }
 
+    #[inline(always)]
     pub(crate) fn with_mut<R>(&self, f: impl FnOnce(*mut T) -> R) -> R {
         f(self.0.get())
     }

tokio/src/runtime/builder.rs

@@ -1172,7 +1172,7 @@ cfg_rt_multi_thread! {
         fn build_threaded_runtime(&mut self) -> io::Result<Runtime> {
             use crate::loom::sys::num_cpus;
             use crate::runtime::{Config, runtime::Scheduler};
-            use crate::runtime::scheduler::{self, MultiThread};
+            use crate::runtime::scheduler::MultiThread;
 
             let core_threads = self.worker_threads.unwrap_or_else(num_cpus);
 
@@ -1187,7 +1187,7 @@ cfg_rt_multi_thread! {
             let seed_generator_1 = self.seed_generator.next_generator();
             let seed_generator_2 = self.seed_generator.next_generator();
 
-            let (scheduler, handle, launch) = MultiThread::new(
+            let (scheduler, handle) = MultiThread::new(
                 core_threads,
                 driver,
                 driver_handle,
@@ -1206,12 +1206,6 @@ cfg_rt_multi_thread! {
                 },
             );
 
-            let handle = Handle { inner: scheduler::Handle::MultiThread(handle) };
-
-            // Spawn the thread pool workers
-            let _enter = handle.enter();
-            launch.launch();
-
             Ok(Runtime::from_parts(Scheduler::MultiThread(scheduler), handle, blocking_pool))
         }
     }

tokio/src/runtime/mod.rs

@@ -186,6 +186,7 @@ pub(crate) mod coop;
 pub(crate) mod park;
 
 mod driver;
+use driver::Driver;
 
 pub(crate) mod scheduler;
 

tokio/src/runtime/scheduler/lock.rs

@@ -1,3 +1,5 @@
+use crate::loom::sync::{Mutex, MutexGuard};
+
 /// A lock (mutex) yielding generic data.
 pub(crate) trait Lock<T> {
     type Handle: AsMut<T>;

tokio/src/runtime/scheduler/mod.rs

@@ -123,6 +123,15 @@ cfg_rt! {
                 _ => panic!("not a CurrentThread handle"),
             }
         }
+
+        cfg_rt_multi_thread! {
+            pub(crate) fn expect_multi_thread(&self) -> &Arc<multi_thread::Handle> {
+                match self {
+                    Handle::MultiThread(handle) => handle,
+                    _ => panic!("not a `MultiThread` handle"),
+                }
+            }
+        }
     }
 
     cfg_metrics! {

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

@@ -5,24 +5,63 @@ mod imp {
 
     static NUM_MAINTENANCE: AtomicUsize = AtomicUsize::new(0);
     static NUM_NOTIFY_LOCAL: AtomicUsize = AtomicUsize::new(0);
+    static NUM_NOTIFY_REMOTE: AtomicUsize = AtomicUsize::new(0);
     static NUM_UNPARKS_LOCAL: AtomicUsize = AtomicUsize::new(0);
+    static NUM_UNPARKS_REMOTE: AtomicUsize = AtomicUsize::new(0);
     static NUM_LIFO_SCHEDULES: AtomicUsize = AtomicUsize::new(0);
     static NUM_LIFO_CAPPED: AtomicUsize = AtomicUsize::new(0);
+    static NUM_STEALS: AtomicUsize = AtomicUsize::new(0);
+    static NUM_OVERFLOW: AtomicUsize = AtomicUsize::new(0);
+    static NUM_PARK: AtomicUsize = AtomicUsize::new(0);
+    static NUM_POLLS: AtomicUsize = AtomicUsize::new(0);
+    static NUM_LIFO_POLLS: AtomicUsize = AtomicUsize::new(0);
+    static NUM_REMOTE_BATCH: AtomicUsize = AtomicUsize::new(0);
+    static NUM_GLOBAL_QUEUE_INTERVAL: AtomicUsize = AtomicUsize::new(0);
+    static NUM_NO_AVAIL_CORE: AtomicUsize = AtomicUsize::new(0);
+    static NUM_RELAY_SEARCH: AtomicUsize = AtomicUsize::new(0);
+    static NUM_SPIN_STALL: AtomicUsize = AtomicUsize::new(0);
+    static NUM_NO_LOCAL_WORK: AtomicUsize = AtomicUsize::new(0);
 
     impl Drop for super::Counters {
         fn drop(&mut self) {
             let notifies_local = NUM_NOTIFY_LOCAL.load(Relaxed);
+            let notifies_remote = NUM_NOTIFY_REMOTE.load(Relaxed);
             let unparks_local = NUM_UNPARKS_LOCAL.load(Relaxed);
+            let unparks_remote = NUM_UNPARKS_REMOTE.load(Relaxed);
             let maintenance = NUM_MAINTENANCE.load(Relaxed);
             let lifo_scheds = NUM_LIFO_SCHEDULES.load(Relaxed);
             let lifo_capped = NUM_LIFO_CAPPED.load(Relaxed);
+            let num_steals = NUM_STEALS.load(Relaxed);
+            let num_overflow = NUM_OVERFLOW.load(Relaxed);
+            let num_park = NUM_PARK.load(Relaxed);
+            let num_polls = NUM_POLLS.load(Relaxed);
+            let num_lifo_polls = NUM_LIFO_POLLS.load(Relaxed);
+            let num_remote_batch = NUM_REMOTE_BATCH.load(Relaxed);
+            let num_global_queue_interval = NUM_GLOBAL_QUEUE_INTERVAL.load(Relaxed);
+            let num_no_avail_core = NUM_NO_AVAIL_CORE.load(Relaxed);
+            let num_relay_search = NUM_RELAY_SEARCH.load(Relaxed);
+            let num_spin_stall = NUM_SPIN_STALL.load(Relaxed);
+            let num_no_local_work = NUM_NO_LOCAL_WORK.load(Relaxed);
 
             println!("---");
-            println!("notifies (local): {}", notifies_local);
-            println!(" unparks (local): {}", unparks_local);
-            println!("     maintenance: {}", maintenance);
-            println!("  LIFO schedules: {}", lifo_scheds);
-            println!("     LIFO capped: {}", lifo_capped);
+            println!("notifies (remote): {}", notifies_remote);
+            println!(" notifies (local): {}", notifies_local);
+            println!("  unparks (local): {}", unparks_local);
+            println!(" unparks (remote): {}", unparks_remote);
+            println!("  notify, no core: {}", num_no_avail_core);
+            println!("      maintenance: {}", maintenance);
+            println!("   LIFO schedules: {}", lifo_scheds);
+            println!("      LIFO capped: {}", lifo_capped);
+            println!("           steals: {}", num_steals);
+            println!("  queue overflows: {}", num_overflow);
+            println!("            parks: {}", num_park);
+            println!("            polls: {}", num_polls);
+            println!("     polls (LIFO): {}", num_lifo_polls);
+            println!("remote task batch: {}", num_remote_batch);
+            println!("global Q interval: {}", num_global_queue_interval);
+            println!("     relay search: {}", num_relay_search);
+            println!("       spin stall: {}", num_spin_stall);
+            println!("    no local work: {}", num_no_local_work);
         }
     }
 
@@ -30,10 +69,18 @@ mod imp {
         NUM_NOTIFY_LOCAL.fetch_add(1, Relaxed);
     }
 
+    pub(crate) fn inc_num_notify_remote() {
+        NUM_NOTIFY_REMOTE.fetch_add(1, Relaxed);
+    }
+
     pub(crate) fn inc_num_unparks_local() {
         NUM_UNPARKS_LOCAL.fetch_add(1, Relaxed);
     }
 
+    pub(crate) fn inc_num_unparks_remote() {
+        NUM_UNPARKS_REMOTE.fetch_add(1, Relaxed);
+    }
+
     pub(crate) fn inc_num_maintenance() {
         NUM_MAINTENANCE.fetch_add(1, Relaxed);
     }
@@ -45,15 +92,72 @@ mod imp {
     pub(crate) fn inc_lifo_capped() {
         NUM_LIFO_CAPPED.fetch_add(1, Relaxed);
     }
+
+    pub(crate) fn inc_num_steals() {
+        NUM_STEALS.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_overflows() {
+        NUM_OVERFLOW.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_parks() {
+        NUM_PARK.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_polls() {
+        NUM_POLLS.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_lifo_polls() {
+        NUM_LIFO_POLLS.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_remote_batch() {
+        NUM_REMOTE_BATCH.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_global_queue_interval() {
+        NUM_GLOBAL_QUEUE_INTERVAL.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_notify_no_core() {
+        NUM_NO_AVAIL_CORE.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_relay_search() {
+        NUM_RELAY_SEARCH.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_spin_stall() {
+        NUM_SPIN_STALL.fetch_add(1, Relaxed);
+    }
+
+    pub(crate) fn inc_num_no_local_work() {
+        NUM_NO_LOCAL_WORK.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_notify_remote() {}
     pub(crate) fn inc_num_unparks_local() {}
+    pub(crate) fn inc_num_unparks_remote() {}
     pub(crate) fn inc_num_maintenance() {}
     pub(crate) fn inc_lifo_schedules() {}
     pub(crate) fn inc_lifo_capped() {}
+    pub(crate) fn inc_num_steals() {}
+    pub(crate) fn inc_num_overflows() {}
+    pub(crate) fn inc_num_parks() {}
+    pub(crate) fn inc_num_polls() {}
+    pub(crate) fn inc_num_lifo_polls() {}
+    pub(crate) fn inc_num_remote_batch() {}
+    pub(crate) fn inc_global_queue_interval() {}
+    pub(crate) fn inc_notify_no_core() {}
+    pub(crate) fn inc_num_relay_search() {}
+    pub(crate) fn inc_num_spin_stall() {}
+    pub(crate) fn inc_num_no_local_work() {}
 }
 
 #[derive(Debug)]

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

@@ -43,7 +43,8 @@ impl Handle {
     }
 
     pub(crate) fn shutdown(&self) {
-        self.close();
+        self.shared.close();
+        self.driver.unpark();
     }
 
     pub(super) fn bind_new_task<T>(me: &Arc<Self>, future: T, id: task::Id) -> JoinHandle<T::Output>
@@ -54,7 +55,7 @@ impl Handle {
         let (handle, notified) = me.shared.owned.bind(future, me.clone(), id);
 
         if let Some(notified) = notified {
-            me.schedule_task(notified, false);
+            me.shared.schedule_task(notified, false);
         }
 
         handle

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

@@ -1,240 +1,434 @@
 //! Coordinates idling workers
 
-use crate::loom::sync::atomic::AtomicUsize;
-use crate::runtime::scheduler::multi_thread::Shared;
+use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
+use crate::loom::sync::MutexGuard;
+use crate::runtime::scheduler::multi_thread::{worker, Core, Shared};
 
-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 cores
+    num_searching: AtomicUsize,
 
-    /// Total number of workers.
-    num_workers: usize,
+    /// Number of idle cores
+    num_idle: AtomicUsize,
+
+    /// Map of idle cores
+    idle_map: IdleMap,
+
+    /// Used to catch false-negatives when waking workers
+    needs_searching: AtomicBool,
+
+    /// Total number of cores
+    num_cores: usize,
+}
+
+pub(super) struct IdleMap {
+    chunks: Vec<AtomicUsize>,
+}
+
+pub(super) struct Snapshot {
+    chunks: Vec<usize>,
 }
 
 /// Data synchronized by the scheduler mutex
 pub(super) struct Synced {
-    /// Sleeping workers
+    /// Worker IDs that are currently sleeping
     sleepers: Vec<usize>,
+
+    /// Cores available for workers
+    available_cores: Vec<Box<Core>>,
 }
 
-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, Synced) {
-        let init = State::new(num_workers);
-
+    pub(super) fn new(cores: Vec<Box<Core>>, num_workers: usize) -> (Idle, Synced) {
         let idle = Idle {
-            state: AtomicUsize::new(init.into()),
-            num_workers,
+            num_searching: AtomicUsize::new(0),
+            num_idle: AtomicUsize::new(cores.len()),
+            idle_map: IdleMap::new(&cores),
+            needs_searching: AtomicBool::new(false),
+            num_cores: cores.len(),
         };
 
         let synced = Synced {
             sleepers: Vec::with_capacity(num_workers),
+            available_cores: cores,
         };
 
         (idle, synced)
     }
 
-    /// If there are no workers actively searching, returns the index of a
-    /// worker currently sleeping.
-    pub(super) fn worker_to_notify(&self, shared: &Shared) -> 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.
-        //
-        // 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() {
-            return None;
+    pub(super) fn num_idle(&self, synced: &Synced) -> usize {
+        debug_assert_eq!(synced.available_cores.len(), self.num_idle.load(Acquire));
+        synced.available_cores.len()
+    }
+
+    pub(super) fn num_searching(&self) -> usize {
+        self.num_searching.load(Acquire)
+    }
+
+    pub(super) fn is_idle(&self, index: usize) -> bool {
+        self.idle_map.get(index)
+    }
+
+    pub(super) fn snapshot(&self, snapshot: &mut Snapshot) {
+        snapshot.update(&self.idle_map)
+    }
+
+    /// Try to acquire an available core
+    pub(super) fn try_acquire_available_core(&self, synced: &mut Synced) -> Option<Box<Core>> {
+        let ret = synced.available_cores.pop();
+
+        if let Some(core) = &ret {
+            // Decrement the number of idle cores
+            let num_idle = self.num_idle.load(Acquire) - 1;
+            debug_assert_eq!(num_idle, synced.available_cores.len());
+            self.num_idle.store(num_idle, Release);
+
+            self.idle_map.unset(core.index);
+            debug_assert!(self.idle_map.matches(&synced.available_cores));
         }
 
-        // Acquire the lock
-        let mut lock = shared.synced.lock();
-
-        // Check again, now that the lock is acquired
-        if !self.notify_should_wakeup() {
-            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);
-
-        // Get the worker to unpark
-        let ret = lock.idle.sleepers.pop();
-        debug_assert!(ret.is_some());
-
         ret
     }
 
-    /// Returns `true` if the worker needs to do a final check for submitted
-    /// work.
-    pub(super) fn transition_worker_to_parked(
+    /// We need at least one searching worker
+    pub(super) fn notify_local(&self, shared: &Shared) {
+        if self.num_searching.load(Acquire) != 0 {
+            // There already is a searching worker. Note, that this could be a
+            // false positive. However, because this method is called **from** a
+            // worker, we know that there is at least one worker currently
+            // awake, so the scheduler won't deadlock.
+            return;
+        }
+
+        if self.num_idle.load(Acquire) == 0 {
+            self.needs_searching.store(true, Release);
+            return;
+        }
+
+        // There aren't any searching workers. Try to initialize one
+        if self
+            .num_searching
+            .compare_exchange(0, 1, AcqRel, Acquire)
+            .is_err()
+        {
+            // Failing the compare_exchange means another thread concurrently
+            // launched a searching worker.
+            return;
+        }
+
+        super::counters::inc_num_unparks_local();
+
+        // Acquire the lock
+        let synced = shared.synced.lock();
+        self.notify_synced(synced, shared);
+    }
+
+    /// Notifies a single worker
+    pub(super) fn notify_remote(&self, synced: MutexGuard<'_, worker::Synced>, shared: &Shared) {
+        if synced.idle.sleepers.is_empty() {
+            self.needs_searching.store(true, Release);
+            return;
+        }
+
+        // We need to establish a stronger barrier than with `notify_local`
+        if self
+            .num_searching
+            .compare_exchange(0, 1, AcqRel, Acquire)
+            .is_err()
+        {
+            return;
+        }
+
+        self.notify_synced(synced, shared);
+    }
+
+    /// Notify a worker while synced
+    fn notify_synced(&self, mut synced: MutexGuard<'_, worker::Synced>, shared: &Shared) {
+        // Find a sleeping worker
+        if let Some(worker) = synced.idle.sleepers.pop() {
+            // Find an available core
+            if let Some(mut core) = synced.idle.available_cores.pop() {
+                debug_assert!(!core.is_searching);
+                core.is_searching = true;
+
+                self.idle_map.unset(core.index);
+                debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
+
+                // Assign the core to the worker
+                synced.assigned_cores[worker] = Some(core);
+
+                let num_idle = synced.idle.available_cores.len();
+                debug_assert_eq!(num_idle, self.num_idle.load(Acquire) - 1);
+
+                // Update the number of sleeping workers
+                self.num_idle.store(num_idle, Release);
+
+                // Drop the lock before notifying the condvar.
+                drop(synced);
+
+                super::counters::inc_num_unparks_remote();
+
+                // Notify the worker
+                shared.condvars[worker].notify_one();
+                return;
+            } else {
+                synced.idle.sleepers.push(worker);
+            }
+        }
+
+        super::counters::inc_notify_no_core();
+
+        // Set the `needs_searching` flag, this happens *while* the lock is held.
+        self.needs_searching.store(true, Release);
+        self.num_searching.fetch_sub(1, Release);
+
+        // Explicit mutex guard drop to show that holding the guard to this
+        // point is significant. `needs_searching` and `num_searching` must be
+        // updated in the critical section.
+        drop(synced);
+    }
+
+    pub(super) fn notify_mult(
         &self,
-        shared: &Shared,
-        worker: usize,
-        is_searching: bool,
-    ) -> bool {
-        // Acquire the lock
-        let mut lock = shared.synced.lock();
+        synced: &mut worker::Synced,
+        workers: &mut Vec<usize>,
+        num: usize,
+    ) {
+        debug_assert!(workers.is_empty());
 
-        // Decrement the number of unparked threads
-        let ret = State::dec_num_unparked(&self.state, is_searching);
+        for _ in 0..num {
+            if let Some(worker) = synced.idle.sleepers.pop() {
+                if let Some(core) = synced.idle.available_cores.pop() {
+                    debug_assert!(!core.is_searching);
 
-        // Track the sleeping worker
-        lock.idle.sleepers.push(worker);
+                    self.idle_map.unset(core.index);
 
-        ret
-    }
+                    synced.assigned_cores[worker] = Some(core);
 
-    pub(super) fn transition_worker_to_searching(&self) -> bool {
-        let state = State::load(&self.state, SeqCst);
-        if 2 * state.num_searching() >= self.num_workers {
-            return false;
+                    workers.push(worker);
+
+                    continue;
+                } else {
+                    synced.idle.sleepers.push(worker);
+                }
+            }
+
+            break;
         }
 
-        // 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);
-        true
+        if !workers.is_empty() {
+            debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
+            let num_idle = synced.idle.available_cores.len();
+            self.num_idle.store(num_idle, Release);
+        } else {
+            debug_assert_eq!(
+                synced.idle.available_cores.len(),
+                self.num_idle.load(Acquire)
+            );
+            self.needs_searching.store(true, Release);
+        }
+    }
+
+    pub(super) fn shutdown(&self, synced: &mut worker::Synced, shared: &Shared) {
+        // Wake every sleeping worker and assign a core to it. There may not be
+        // enough sleeping workers for all cores, but other workers will
+        // eventually find the cores and shut them down.
+        while !synced.idle.sleepers.is_empty() && !synced.idle.available_cores.is_empty() {
+            let worker = synced.idle.sleepers.pop().unwrap();
+            let core = synced.idle.available_cores.pop().unwrap();
+
+            self.idle_map.unset(core.index);
+
+            synced.assigned_cores[worker] = Some(core);
+            shared.condvars[worker].notify_one();
+
+            self.num_idle
+                .store(synced.idle.available_cores.len(), Release);
+        }
+
+        debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
+
+        // Wake up any other workers
+        while let Some(index) = synced.idle.sleepers.pop() {
+            shared.condvars[index].notify_one();
+        }
+    }
+
+    /// The worker releases the given core, making it available to other workers
+    /// that are waiting.
+    pub(super) fn release_core(&self, synced: &mut worker::Synced, core: Box<Core>) {
+        // The core should not be searching at this point
+        debug_assert!(!core.is_searching);
+
+        // Check that this isn't the final worker to go idle *and*
+        // `needs_searching` is set.
+        debug_assert!(!self.needs_searching.load(Acquire) || num_active_workers(&synced.idle) > 1);
+
+        let num_idle = synced.idle.available_cores.len();
+        debug_assert_eq!(num_idle, self.num_idle.load(Acquire));
+
+        self.idle_map.set(core.index);
+
+        // Store the core in the list of available cores
+        synced.idle.available_cores.push(core);
+
+        debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
+
+        // Update `num_idle`
+        self.num_idle.store(num_idle + 1, Release);
+    }
+
+    pub(super) fn transition_worker_to_parked(&self, synced: &mut worker::Synced, index: usize) {
+        // Store the worker index in the list of sleepers
+        synced.idle.sleepers.push(index);
+
+        // The worker's assigned core slot should be empty
+        debug_assert!(synced.assigned_cores[index].is_none());
+    }
+
+    pub(super) fn try_transition_worker_to_searching(&self, core: &mut Core) {
+        debug_assert!(!core.is_searching);
+
+        let num_searching = self.num_searching.load(Acquire);
+        let num_idle = self.num_idle.load(Acquire);
+
+        if 2 * num_searching >= self.num_cores - num_idle {
+            return;
+        }
+
+        self.transition_worker_to_searching(core);
+    }
+
+    /// Needs to happen while synchronized in order to avoid races
+    pub(super) fn transition_worker_to_searching_if_needed(
+        &self,
+        _synced: &mut Synced,
+        core: &mut Core,
+    ) -> bool {
+        if self.needs_searching.load(Acquire) {
+            // Needs to be called while holding the lock
+            self.transition_worker_to_searching(core);
+            true
+        } else {
+            false
+        }
+    }
+
+    fn transition_worker_to_searching(&self, core: &mut Core) {
+        core.is_searching = true;
+        self.num_searching.fetch_add(1, AcqRel);
+        self.needs_searching.store(false, Release);
     }
 
     /// 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)
-    }
+    pub(super) fn transition_worker_from_searching(&self, core: &mut Core) -> bool {
+        debug_assert!(core.is_searching);
+        core.is_searching = false;
 
-    /// Unpark a specific worker. This happens if tasks are submitted from
-    /// within the worker's park routine.
-    ///
-    /// Returns `true` if the worker was parked before calling the method.
-    pub(super) fn unpark_worker_by_id(&self, shared: &Shared, worker_id: usize) -> bool {
-        let mut lock = shared.synced.lock();
-        let sleepers = &mut lock.idle.sleepers;
+        let prev = self.num_searching.fetch_sub(1, AcqRel);
+        debug_assert!(prev > 0);
 
-        for index in 0..sleepers.len() {
-            if sleepers[index] == worker_id {
-                sleepers.swap_remove(index);
-
-                // Update the state accordingly while the lock is held.
-                State::unpark_one(&self.state, 0);
-
-                return true;
-            }
-        }
-
-        false
-    }
-
-    /// Returns `true` if `worker_id` is contained in the sleep set.
-    pub(super) fn is_parked(&self, shared: &Shared, worker_id: usize) -> bool {
-        let lock = shared.synced.lock();
-        lock.idle.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
+        prev == 1
     }
 }
 
-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());
+const BITS: usize = usize::BITS as usize;
+const BIT_MASK: usize = (usize::BITS - 1) as usize;
+
+impl IdleMap {
+    fn new(cores: &[Box<Core>]) -> IdleMap {
+        let ret = IdleMap::new_n(num_chunks(cores.len()));
+        ret.set_all(cores);
+
         ret
     }
 
-    fn load(cell: &AtomicUsize, ordering: Ordering) -> State {
-        State(cell.load(ordering))
+    fn new_n(n: usize) -> IdleMap {
+        let chunks = (0..n).map(|_| AtomicUsize::new(0)).collect();
+        IdleMap { chunks }
     }
 
-    fn unpark_one(cell: &AtomicUsize, num_searching: usize) {
-        cell.fetch_add(num_searching | (1 << UNPARK_SHIFT), SeqCst);
+    fn get(&self, index: usize) -> bool {
+        let (chunk, mask) = index_to_mask(index);
+        self.chunks[chunk].load(Acquire) & mask == mask
     }
 
-    fn inc_num_searching(cell: &AtomicUsize, ordering: Ordering) {
-        cell.fetch_add(1, ordering);
+    fn set(&self, index: usize) {
+        let (chunk, mask) = index_to_mask(index);
+        let prev = self.chunks[chunk].load(Acquire);
+        let next = prev | mask;
+        self.chunks[chunk].store(next, Release);
     }
 
-    /// 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
+    fn set_all(&self, cores: &[Box<Core>]) {
+        for core in cores {
+            self.set(core.index);
+        }
     }
 
-    /// 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;
+    fn unset(&self, index: usize) {
+        let (chunk, mask) = index_to_mask(index);
+        let prev = self.chunks[chunk].load(Acquire);
+        let next = prev & !mask;
+        self.chunks[chunk].store(next, Release);
+    }
 
-        if is_searching {
-            dec += 1;
+    fn matches(&self, idle_cores: &[Box<Core>]) -> bool {
+        let expect = IdleMap::new_n(self.chunks.len());
+        expect.set_all(idle_cores);
+
+        for (i, chunk) in expect.chunks.iter().enumerate() {
+            if chunk.load(Acquire) != self.chunks[i].load(Acquire) {
+                return false;
+            }
         }
 
-        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
+        true
     }
 }
 
-impl From<usize> for State {
-    fn from(src: usize) -> State {
-        State(src)
+impl Snapshot {
+    pub(crate) fn new(idle: &Idle) -> Snapshot {
+        let chunks = vec![0; idle.idle_map.chunks.len()];
+        let mut ret = Snapshot { chunks };
+        ret.update(&idle.idle_map);
+        ret
+    }
+
+    fn update(&mut self, idle_map: &IdleMap) {
+        for i in 0..self.chunks.len() {
+            self.chunks[i] = idle_map.chunks[i].load(Acquire);
+        }
+    }
+
+    pub(super) fn is_idle(&self, index: usize) -> bool {
+        let (chunk, mask) = index_to_mask(index);
+        debug_assert!(
+            chunk < self.chunks.len(),
+            "index={}; chunks={}",
+            index,
+            self.chunks.len()
+        );
+        self.chunks[chunk] & mask == mask
     }
 }
 
-impl From<State> for usize {
-    fn from(src: State) -> usize {
-        src.0
-    }
+fn num_chunks(max_cores: usize) -> usize {
+    (max_cores / BITS) + 1
 }
 
-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()
-    }
+fn index_to_mask(index: usize) -> (usize, usize) {
+    let mask = 1 << (index & BIT_MASK);
+    let chunk = index / BITS;
+
+    (chunk, mask)
 }
 
-#[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());
+fn num_active_workers(synced: &Synced) -> usize {
+    synced.available_cores.capacity() - synced.available_cores.len()
 }

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

@@ -15,13 +15,11 @@ use self::idle::Idle;
 mod stats;
 pub(crate) use stats::Stats;
 
-mod park;
-pub(crate) use park::{Parker, Unparker};
-
 pub(crate) mod queue;
 
 mod worker;
-pub(crate) use worker::{Context, Launch, Shared};
+use worker::Core;
+pub(crate) use worker::{Context, Shared};
 
 cfg_taskdump! {
     mod trace;
@@ -37,9 +35,8 @@ cfg_not_taskdump! {
 
 pub(crate) use worker::block_in_place;
 
-use crate::loom::sync::Arc;
 use crate::runtime::{
-    blocking,
+    self, blocking,
     driver::{self, Driver},
     scheduler, Config,
 };
@@ -61,18 +58,17 @@ impl MultiThread {
         blocking_spawner: blocking::Spawner,
         seed_generator: RngSeedGenerator,
         config: Config,
-    ) -> (MultiThread, Arc<Handle>, Launch) {
-        let parker = Parker::new(driver);
-        let (handle, launch) = worker::create(
+    ) -> (MultiThread, runtime::Handle) {
+        let handle = worker::create(
             size,
-            parker,
+            driver,
             driver_handle,
             blocking_spawner,
             seed_generator,
             config,
         );
 
-        (MultiThread, handle, launch)
+        (MultiThread, handle)
     }
 
     /// Blocks the current thread waiting for the future to complete.

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

@@ -33,6 +33,7 @@ pub(crate) struct Local<T: 'static> {
 /// Consumer handle. May be used from many threads.
 pub(crate) struct Steal<T: 'static>(Arc<Inner<T>>);
 
+#[repr(align(128))]
 pub(crate) struct Inner<T: 'static> {
     /// Concurrently updated by many threads.
     ///
@@ -119,6 +120,11 @@ impl<T> Local<T> {
         LOCAL_QUEUE_CAPACITY
     }
 
+    /// Returns `true` if there are no entries in the queue
+    pub(crate) fn is_empty(&self) -> bool {
+        self.inner.is_empty()
+    }
+
     /// Returns false if there are any entries in the queue
     ///
     /// Separate to is_stealable so that refactors of is_stealable to "protect"
@@ -199,11 +205,13 @@ impl<T> Local<T> {
                 // There is capacity for the task
                 break tail;
             } else if steal != real {
+                super::counters::inc_num_overflows();
                 // Concurrently stealing, this will free up capacity, so only
                 // push the task onto the inject queue
                 overflow.push(task);
                 return;
             } else {
+                super::counters::inc_num_overflows();
                 // Push the current task and half of the queue into the
                 // inject queue.
                 match self.push_overflow(task, real, tail, overflow, stats) {
@@ -420,6 +428,8 @@ impl<T> Steal<T> {
             return None;
         }
 
+        super::counters::inc_num_steals();
+
         dst_stats.incr_steal_count(n as u16);
         dst_stats.incr_steal_operations();
 

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

@@ -10,6 +10,16 @@ pub(crate) struct Stats {
     /// user.
     batch: MetricsBatch,
 
+    /// Exponentially-weighted moving average of time spent polling scheduled a
+    /// task.
+    ///
+    /// Tracked in nanoseconds, stored as a f64 since that is what we use with
+    /// the EWMA calculations
+    task_poll_time_ewma: f64,
+}
+
+/// Transient state
+pub(crate) struct Ephemeral {
     /// Instant at which work last resumed (continued after park).
     ///
     /// This duplicates the value stored in `MetricsBatch`. We will unify
@@ -19,12 +29,20 @@ pub(crate) struct Stats {
     /// Number of tasks polled in the batch of scheduled tasks
     tasks_polled_in_batch: usize,
 
-    /// Exponentially-weighted moving average of time spent polling scheduled a
-    /// task.
-    ///
-    /// Tracked in nanoseconds, stored as a f64 since that is what we use with
-    /// the EWMA calculations
-    task_poll_time_ewma: f64,
+    /// Used to ensure calls to start / stop batch are paired
+    #[cfg(debug_assertions)]
+    batch_started: bool,
+}
+
+impl Ephemeral {
+    pub(crate) fn new() -> Ephemeral {
+        Ephemeral {
+            processing_scheduled_tasks_started_at: Instant::now(),
+            tasks_polled_in_batch: 0,
+            #[cfg(debug_assertions)]
+            batch_started: false,
+        }
+    }
 }
 
 /// How to weigh each individual poll time, value is plucked from thin air.
@@ -40,6 +58,9 @@ const MAX_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL: u32 = 127;
 const TARGET_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL: u32 = 61;
 
 impl Stats {
+    pub(crate) const DEFAULT_GLOBAL_QUEUE_INTERVAL: u32 =
+        TARGET_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL;
+
     pub(crate) fn new(worker_metrics: &WorkerMetrics) -> Stats {
         // Seed the value with what we hope to see.
         let task_poll_time_ewma =
@@ -47,12 +68,14 @@ impl Stats {
 
         Stats {
             batch: MetricsBatch::new(worker_metrics),
-            processing_scheduled_tasks_started_at: Instant::now(),
-            tasks_polled_in_batch: 0,
             task_poll_time_ewma,
         }
     }
 
+    pub(crate) fn mean_task_poll_duration(&self) -> f64 {
+        self.task_poll_time_ewma
+    }
+
     pub(crate) fn tuned_global_queue_interval(&self, config: &Config) -> u32 {
         // If an interval is explicitly set, don't tune.
         if let Some(configured) = config.global_queue_interval {
@@ -85,24 +108,36 @@ impl Stats {
         self.batch.inc_local_schedule_count();
     }
 
-    pub(crate) fn start_processing_scheduled_tasks(&mut self) {
+    pub(crate) fn start_processing_scheduled_tasks(&mut self, ephemeral: &mut Ephemeral) {
         self.batch.start_processing_scheduled_tasks();
 
-        self.processing_scheduled_tasks_started_at = Instant::now();
-        self.tasks_polled_in_batch = 0;
+        #[cfg(debug_assertions)]
+        {
+            debug_assert!(!ephemeral.batch_started);
+            ephemeral.batch_started = true;
+        }
+
+        ephemeral.processing_scheduled_tasks_started_at = Instant::now();
+        ephemeral.tasks_polled_in_batch = 0;
     }
 
-    pub(crate) fn end_processing_scheduled_tasks(&mut self) {
+    pub(crate) fn end_processing_scheduled_tasks(&mut self, ephemeral: &mut Ephemeral) {
         self.batch.end_processing_scheduled_tasks();
 
+        #[cfg(debug_assertions)]
+        {
+            debug_assert!(ephemeral.batch_started);
+            ephemeral.batch_started = false;
+        }
+
         // Update the EWMA task poll time
-        if self.tasks_polled_in_batch > 0 {
+        if ephemeral.tasks_polled_in_batch > 0 {
             let now = Instant::now();
 
             // If we "overflow" this conversion, we have bigger problems than
             // slightly off stats.
-            let elapsed = (now - self.processing_scheduled_tasks_started_at).as_nanos() as f64;
-            let num_polls = self.tasks_polled_in_batch as f64;
+            let elapsed = (now - ephemeral.processing_scheduled_tasks_started_at).as_nanos() as f64;
+            let num_polls = ephemeral.tasks_polled_in_batch as f64;
 
             // Calculate the mean poll duration for a single task in the batch
             let mean_poll_duration = elapsed / num_polls;
@@ -116,10 +151,10 @@ impl Stats {
         }
     }
 
-    pub(crate) fn start_poll(&mut self) {
+    pub(crate) fn start_poll(&mut self, ephemeral: &mut Ephemeral) {
         self.batch.start_poll();
 
-        self.tasks_polled_in_batch += 1;
+        ephemeral.tasks_polled_in_batch += 1;
     }
 
     pub(crate) fn end_poll(&mut self) {

tokio/src/runtime/task/atomic_cell.rs

@@ -0,0 +1,99 @@
+use crate::loom::sync::atomic::AtomicPtr;
+use crate::loom::sync::atomic::Ordering::{AcqRel, Acquire, Release};
+use crate::runtime::task::{Header, Notified, RawTask};
+
+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.is_none()
+    }
+
+    pub(crate) fn is_none(&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(RawTask::from_raw(ptr)) })
+    }
+
+    pub(crate) fn swap_local(&self, task: Notified<S>) -> Option<Notified<S>> {
+        let next = task.into_raw().header_ptr().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(RawTask::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(RawTask::from_raw(NonNull::new_unchecked(task)))
+            });
+        }
+
+        None
+    }
+}

tokio/src/runtime/task/list.rs

@@ -53,6 +53,7 @@ cfg_not_has_atomic_u64! {
     }
 }
 
+#[repr(align(128))]
 pub(crate) struct OwnedTasks<S: 'static> {
     inner: Mutex<CountedOwnedTasksInner<S>>,
     id: u64,
@@ -119,7 +120,7 @@ impl<S: 'static> OwnedTasks<S> {
     /// a LocalNotified, giving the thread permission to poll this task.
     #[inline]
     pub(crate) fn assert_owner(&self, task: Notified<S>) -> LocalNotified<S> {
-        assert_eq!(task.header().get_owner_id(), self.id);
+        debug_assert_eq!(task.header().get_owner_id(), self.id);
 
         // safety: All tasks bound to this OwnedTasks are Send, so it is safe
         // to poll it on this thread no matter what thread we are on.

tokio/src/runtime/task/mod.rs

@@ -168,6 +168,9 @@
 // unstable. This should be removed once `JoinSet` is stabilized.
 #![cfg_attr(not(tokio_unstable), allow(dead_code))]
 
+mod atomic_cell;
+pub(crate) use atomic_cell::AtomicCell;
+
 mod core;
 use self::core::Cell;
 use self::core::Header;

tokio/src/runtime/tests/loom_pool.rs

@@ -412,8 +412,8 @@ async fn multi_gated() {
     }
 
     poll_fn(move |cx| {
+        gate.waker.register_by_ref(cx.waker());
         if gate.count.load(SeqCst) < 2 {
-            gate.waker.register_by_ref(cx.waker());
             Poll::Pending
         } else {
             Poll::Ready(())

tokio/src/runtime/tests/mod.rs

@@ -63,7 +63,7 @@ cfg_loom! {
 
     // Make sure debug assertions are enabled
     #[cfg(not(debug_assertions))]
-    compiler_error!("these tests require debug assertions to be enabled");
+    compile_error!("these tests require debug assertions to be enabled");
 }
 
 cfg_not_loom! {

tokio/src/util/atomic_cell.rs

@@ -1,7 +1,7 @@
 use crate::loom::sync::atomic::AtomicPtr;
 
 use std::ptr;
-use std::sync::atomic::Ordering::AcqRel;
+use std::sync::atomic::Ordering::{AcqRel, Acquire};
 
 pub(crate) struct AtomicCell<T> {
     data: AtomicPtr<T>,
@@ -27,8 +27,16 @@ impl<T> AtomicCell<T> {
     }
 
     pub(crate) fn take(&self) -> Option<Box<T>> {
+        if self.data.load(Acquire).is_null() {
+            return None;
+        }
+
         self.swap(None)
     }
+
+    pub(crate) fn is_none(&self) -> bool {
+        self.data.load(Acquire).is_null()
+    }
 }
 
 fn to_raw<T>(data: Option<Box<T>>) -> *mut T {