wip

tokio/src/runtime/builder.rs

@@ -1198,7 +1198,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::MultiThread;
+            use crate::runtime::scheduler::{self, MultiThread};
 
             let core_threads = self.worker_threads.unwrap_or_else(num_cpus);
 
@@ -1213,7 +1213,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) = MultiThread::new(
+            let (scheduler, handle, launch) = MultiThread::new(
                 core_threads,
                 driver,
                 driver_handle,
@@ -1232,6 +1232,12 @@ 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/scheduler/mod.rs

@@ -160,13 +160,6 @@ cfg_rt! {
         }
 
         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_unstable! {
                 pub(crate) fn expect_multi_thread_alt(&self) -> &Arc<multi_thread_alt::Handle> {
                     match self {

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

@@ -5,63 +5,24 @@ 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 (remote): {}", notifies_remote);
+            println!("notifies (local): {}", notifies_local);
-            println!(" notifies (local): {}", notifies_local);
+            println!(" unparks (local): {}", unparks_local);
-            println!("  unparks (local): {}", unparks_local);
+            println!("     maintenance: {}", maintenance);
-            println!(" unparks (remote): {}", unparks_remote);
+            println!("  LIFO schedules: {}", lifo_scheds);
-            println!("  notify, no core: {}", num_no_avail_core);
+            println!("     LIFO capped: {}", lifo_capped);
-            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);
         }
     }
 
@@ -69,18 +30,10 @@ 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);
     }
@@ -92,72 +45,15 @@ 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,8 +43,7 @@ impl Handle {
     }
 
     pub(crate) fn shutdown(&self) {
-        self.shared.close();
+        self.close();
-        self.driver.unpark();
     }
 
     pub(super) fn bind_new_task<T>(me: &Arc<Self>, future: T, id: task::Id) -> JoinHandle<T::Output>
@@ -55,7 +54,7 @@ impl Handle {
         let (handle, notified) = me.shared.owned.bind(future, me.clone(), id);
 
         if let Some(notified) = notified {
-            me.shared.schedule_task(notified, false);
+            me.schedule_task(notified, false);
         }
 
         handle

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

@@ -1,425 +1,240 @@
 //! Coordinates idling workers
 
-use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
+use crate::loom::sync::atomic::AtomicUsize;
-use crate::loom::sync::MutexGuard;
+use crate::runtime::scheduler::multi_thread::Shared;
-use crate::runtime::scheduler::multi_thread::{worker, Core, Shared};
 
-use std::sync::atomic::Ordering::{AcqRel, Acquire, Release};
+use std::fmt;
+use std::sync::atomic::Ordering::{self, SeqCst};
 
 pub(super) struct Idle {
-    /// Number of searching cores
+    /// Tracks both the number of searching workers and the number of unparked
-    num_searching: AtomicUsize,
+    /// workers.
+    ///
+    /// Used as a fast-path to avoid acquiring the lock when needed.
+    state: AtomicUsize,
 
-    /// Number of idle cores
+    /// Total number of workers.
-    num_idle: AtomicUsize,
+    num_workers: usize,
-
-    /// 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 {
-    /// Worker IDs that are currently sleeping
+    /// Sleeping workers
     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(cores: Vec<Box<Core>>, num_workers: usize) -> (Idle, Synced) {
+    pub(super) fn new(num_workers: usize) -> (Idle, Synced) {
+        let init = State::new(num_workers);
+
         let idle = Idle {
-            num_searching: AtomicUsize::new(0),
+            state: AtomicUsize::new(init.into()),
-            num_idle: AtomicUsize::new(cores.len()),
+            num_workers,
-            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)
     }
 
-    pub(super) fn num_idle(&self, synced: &Synced) -> usize {
+    /// If there are no workers actively searching, returns the index of a
-        debug_assert_eq!(synced.available_cores.len(), self.num_idle.load(Acquire));
+    /// worker currently sleeping.
-        synced.available_cores.len()
+    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
-    pub(super) fn num_searching(&self) -> usize {
+        // notify another worker, eventually leading to our work being found.
-        self.num_searching.load(Acquire)
+        //
-    }
+        // For this to happen, this load must happen before the thread
-
+        // transitioning `num_searching` to zero. Acquire / Release does not
-    pub(super) fn snapshot(&self, snapshot: &mut Snapshot) {
+        // provide sufficient guarantees, so this load is done with `SeqCst` and
-        snapshot.update(&self.idle_map)
+        // will pair with the `fetch_sub(1)` when transitioning out of
-    }
+        // searching.
-
+        if !self.notify_should_wakeup() {
-    /// Try to acquire an available core
+            return None;
-    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
     }
 
-    /// We need at least one searching worker
+    /// Returns `true` if the worker needs to do a final check for submitted
-    pub(super) fn notify_local(&self, shared: &Shared) {
+    /// work.
-        if self.num_searching.load(Acquire) != 0 {
+    pub(super) fn transition_worker_to_parked(
-            // 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,
-        synced: &mut worker::Synced,
+        shared: &Shared,
-        workers: &mut Vec<usize>,
+        worker: usize,
-        num: usize,
+        is_searching: bool,
-    ) {
-        debug_assert!(workers.is_empty());
-
-        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);
-
-                    self.idle_map.unset(core.index);
-
-                    synced.assigned_cores[worker] = Some(core);
-
-                    workers.push(worker);
-
-                    continue;
-                } else {
-                    synced.idle.sleepers.push(worker);
-                }
-            }
-
-            break;
-        }
-
-        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) {
+        // Acquire the lock
-            // Needs to be called while holding the lock
+        let mut lock = shared.synced.lock();
-            self.transition_worker_to_searching(core);
+
-            true
+        // Decrement the number of unparked threads
-        } else {
+        let ret = State::dec_num_unparked(&self.state, is_searching);
-            false
+
-        }
+        // Track the sleeping worker
+        lock.idle.sleepers.push(worker);
+
+        ret
     }
 
-    fn transition_worker_to_searching(&self, core: &mut Core) {
+    pub(super) fn transition_worker_to_searching(&self) -> bool {
-        core.is_searching = true;
+        let state = State::load(&self.state, SeqCst);
-        self.num_searching.fetch_add(1, AcqRel);
+        if 2 * state.num_searching() >= self.num_workers {
-        self.needs_searching.store(false, Release);
+            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);
+        true
     }
 
     /// 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, core: &mut Core) -> bool {
+    pub(super) fn transition_worker_from_searching(&self) -> bool {
-        debug_assert!(core.is_searching);
+        State::dec_num_searching(&self.state)
-        core.is_searching = false;
-
-        let prev = self.num_searching.fetch_sub(1, AcqRel);
-        debug_assert!(prev > 0);
-
-        prev == 1
-    }
-}
-
-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 new_n(n: usize) -> IdleMap {
+    /// Unpark a specific worker. This happens if tasks are submitted from
-        let chunks = (0..n).map(|_| AtomicUsize::new(0)).collect();
+    /// within the worker's park routine.
-        IdleMap { chunks }
+    ///
-    }
+    /// 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;
 
-    fn set(&self, index: usize) {
+        for index in 0..sleepers.len() {
-        let (chunk, mask) = index_to_mask(index);
+            if sleepers[index] == worker_id {
-        let prev = self.chunks[chunk].load(Acquire);
+                sleepers.swap_remove(index);
-        let next = prev | mask;
-        self.chunks[chunk].store(next, Release);
-    }
 
-    fn set_all(&self, cores: &[Box<Core>]) {
+                // Update the state accordingly while the lock is held.
-        for core in cores {
+                State::unpark_one(&self.state, 0);
-            self.set(core.index);
-        }
-    }
 
-    fn unset(&self, index: usize) {
+                return true;
-        let (chunk, mask) = index_to_mask(index);
-        let prev = self.chunks[chunk].load(Acquire);
-        let next = prev & !mask;
-        self.chunks[chunk].store(next, Release);
-    }
-
-    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;
             }
         }
 
-        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
     }
 }
 
-impl Snapshot {
+impl State {
-    pub(crate) fn new(idle: &Idle) -> Snapshot {
+    fn new(num_workers: usize) -> State {
-        let chunks = vec![0; idle.idle_map.chunks.len()];
+        // All workers start in the unparked state
-        let mut ret = Snapshot { chunks };
+        let ret = State(num_workers << UNPARK_SHIFT);
-        ret.update(&idle.idle_map);
+        debug_assert_eq!(num_workers, ret.num_unparked());
+        debug_assert_eq!(0, ret.num_searching());
         ret
     }
 
-    fn update(&mut self, idle_map: &IdleMap) {
+    fn load(cell: &AtomicUsize, ordering: Ordering) -> State {
-        for i in 0..self.chunks.len() {
+        State(cell.load(ordering))
-            self.chunks[i] = idle_map.chunks[i].load(Acquire);
+    }
+
+    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
     }
 
-    pub(super) fn is_idle(&self, index: usize) -> bool {
+    /// Number of workers currently searching
-        let (chunk, mask) = index_to_mask(index);
+    fn num_searching(self) -> usize {
-        debug_assert!(
+        self.0 & SEARCH_MASK
-            chunk < self.chunks.len(),
+    }
-            "index={}; chunks={}",
+
-            index,
+    /// Number of workers currently unparked
-            self.chunks.len()
+    fn num_unparked(self) -> usize {
-        );
+        (self.0 & UNPARK_MASK) >> UNPARK_SHIFT
-        self.chunks[chunk] & mask == mask
     }
 }
 
-fn num_chunks(max_cores: usize) -> usize {
+impl From<usize> for State {
-    (max_cores / BITS) + 1
+    fn from(src: usize) -> State {
+        State(src)
+    }
 }
 
-fn index_to_mask(index: usize) -> (usize, usize) {
+impl From<State> for usize {
-    let mask = 1 << (index & BIT_MASK);
+    fn from(src: State) -> usize {
-    let chunk = index / BITS;
+        src.0
-
+    }
-    (chunk, mask)
 }
 
-fn num_active_workers(synced: &Synced) -> usize {
+impl fmt::Debug for State {
-    synced.available_cores.capacity() - synced.available_cores.len()
+    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

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

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

@@ -33,7 +33,6 @@ 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.
     ///
@@ -106,6 +105,11 @@ pub(crate) fn local<T: 'static>() -> (Steal<T>, Local<T>) {
 }
 
 impl<T> Local<T> {
+    /// Returns the number of entries in the queue
+    pub(crate) fn len(&self) -> usize {
+        self.inner.len() as usize
+    }
+
     /// How many tasks can be pushed into the queue
     pub(crate) fn remaining_slots(&self) -> usize {
         self.inner.remaining_slots()
@@ -115,9 +119,12 @@ impl<T> Local<T> {
         LOCAL_QUEUE_CAPACITY
     }
 
-    /// Returns `true` if there are no entries in the queue
+    /// Returns false if there are any entries in the queue
-    pub(crate) fn is_empty(&self) -> bool {
+    ///
-        self.inner.is_empty()
+    /// Separate to is_stealable so that refactors of is_stealable to "protect"
+    /// some tasks from stealing won't affect this
+    pub(crate) fn has_tasks(&self) -> bool {
+        !self.inner.is_empty()
     }
 
     /// Pushes a batch of tasks to the back of the queue. All tasks must fit in
@@ -192,13 +199,11 @@ 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) {
@@ -381,6 +386,10 @@ impl<T> Local<T> {
 }
 
 impl<T> Steal<T> {
+    pub(crate) fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
     /// Steals half the tasks from self and place them into `dst`.
     pub(crate) fn steal_into(
         &self,
@@ -411,8 +420,6 @@ 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,16 +10,6 @@ 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
@@ -29,20 +19,12 @@ pub(crate) struct Ephemeral {
     /// Number of tasks polled in the batch of scheduled tasks
     tasks_polled_in_batch: usize,
 
-    /// Used to ensure calls to start / stop batch are paired
+    /// Exponentially-weighted moving average of time spent polling scheduled a
-    #[cfg(debug_assertions)]
+    /// task.
-    batch_started: bool,
+    ///
-}
+    /// Tracked in nanoseconds, stored as a f64 since that is what we use with
-
+    /// the EWMA calculations
-impl Ephemeral {
+    task_poll_time_ewma: f64,
-    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.
@@ -58,9 +40,6 @@ 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 =
@@ -68,6 +47,8 @@ impl Stats {
 
         Stats {
             batch: MetricsBatch::new(worker_metrics),
+            processing_scheduled_tasks_started_at: Instant::now(),
+            tasks_polled_in_batch: 0,
             task_poll_time_ewma,
         }
     }
@@ -104,36 +85,24 @@ impl Stats {
         self.batch.inc_local_schedule_count();
     }
 
-    pub(crate) fn start_processing_scheduled_tasks(&mut self, ephemeral: &mut Ephemeral) {
+    pub(crate) fn start_processing_scheduled_tasks(&mut self) {
         self.batch.start_processing_scheduled_tasks();
 
-        #[cfg(debug_assertions)]
+        self.processing_scheduled_tasks_started_at = Instant::now();
-        {
+        self.tasks_polled_in_batch = 0;
-            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, ephemeral: &mut Ephemeral) {
+    pub(crate) fn end_processing_scheduled_tasks(&mut self) {
         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 ephemeral.tasks_polled_in_batch > 0 {
+        if self.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 - ephemeral.processing_scheduled_tasks_started_at).as_nanos() as f64;
+            let elapsed = (now - self.processing_scheduled_tasks_started_at).as_nanos() as f64;
-            let num_polls = ephemeral.tasks_polled_in_batch as f64;
+            let num_polls = self.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;
@@ -147,10 +116,10 @@ impl Stats {
         }
     }
 
-    pub(crate) fn start_poll(&mut self, ephemeral: &mut Ephemeral) {
+    pub(crate) fn start_poll(&mut self) {
         self.batch.start_poll();
 
-        ephemeral.tasks_polled_in_batch += 1;
+        self.tasks_polled_in_batch += 1;
     }
 
     pub(crate) fn end_poll(&mut self) {

tokio/src/util/mod.rs

@@ -68,6 +68,11 @@ cfg_rt! {
     pub(crate) use rc_cell::RcCell;
 }
 
+cfg_rt_multi_thread! {
+    mod try_lock;
+    pub(crate) use try_lock::TryLock;
+}
+
 pub(crate) mod trace;
 
 pub(crate) mod error;