time: revert "use sharding for timer implementation" related changes (#7226)

The work on sharding the timer implementation has caused a measurable performance regression due to increased contention. This patch reverts the current work on sharding. The next step will be to work on a per-worker timer wheel.

Author: carllerche

tokio/src/loom/mocked.rs

@@ -24,11 +24,6 @@ pub(crate) mod sync {
         pub(crate) fn try_lock(&self) -> Option<MutexGuard<'_, T>> {
             self.0.try_lock().ok()
         }
-
-        #[inline]
-        pub(crate) fn get_mut(&mut self) -> &mut T {
-            self.0.get_mut().unwrap()
-        }
     }
 
     #[derive(Debug)]

tokio/src/loom/std/mutex.rs

@@ -33,12 +33,4 @@ impl<T> Mutex<T> {
             Err(TryLockError::WouldBlock) => None,
         }
     }
-
-    #[inline]
-    pub(crate) fn get_mut(&mut self) -> &mut T {
-        match self.0.get_mut() {
-            Ok(val) => val,
-            Err(p_err) => p_err.into_inner(),
-        }
-    }
 }

tokio/src/runtime/builder.rs

@@ -924,7 +924,7 @@ impl Builder {
         }
     }
 
-    fn get_cfg(&self, workers: usize) -> driver::Cfg {
+    fn get_cfg(&self) -> driver::Cfg {
         driver::Cfg {
             enable_pause_time: match self.kind {
                 Kind::CurrentThread => true,
@@ -935,7 +935,6 @@ impl Builder {
             enable_time: self.enable_time,
             start_paused: self.start_paused,
             nevents: self.nevents,
-            workers,
         }
     }
 
@@ -1453,7 +1452,7 @@ impl Builder {
         use crate::runtime::scheduler;
         use crate::runtime::Config;
 
-        let (driver, driver_handle) = driver::Driver::new(self.get_cfg(1))?;
+        let (driver, driver_handle) = driver::Driver::new(self.get_cfg())?;
 
         // Blocking pool
         let blocking_pool = blocking::create_blocking_pool(self, self.max_blocking_threads);
@@ -1608,7 +1607,7 @@ cfg_rt_multi_thread! {
 
             let worker_threads = self.worker_threads.unwrap_or_else(num_cpus);
 
-            let (driver, driver_handle) = driver::Driver::new(self.get_cfg(worker_threads))?;
+            let (driver, driver_handle) = driver::Driver::new(self.get_cfg())?;
 
             // Create the blocking pool
             let blocking_pool =

tokio/src/runtime/context.rs

@@ -3,7 +3,7 @@ use crate::task::coop;
 
 use std::cell::Cell;
 
-#[cfg(any(feature = "rt", feature = "macros", feature = "time"))]
+#[cfg(any(feature = "rt", feature = "macros"))]
 use crate::util::rand::FastRand;
 
 cfg_rt! {
@@ -57,7 +57,7 @@ struct Context {
     #[cfg(feature = "rt")]
     runtime: Cell<EnterRuntime>,
 
-    #[cfg(any(feature = "rt", feature = "macros", feature = "time"))]
+    #[cfg(any(feature = "rt", feature = "macros"))]
     rng: Cell<Option<FastRand>>,
 
     /// Tracks the amount of "work" a task may still do before yielding back to
@@ -100,7 +100,7 @@ tokio_thread_local! {
             #[cfg(feature = "rt")]
             runtime: Cell::new(EnterRuntime::NotEntered),
 
-            #[cfg(any(feature = "rt", feature = "macros", feature = "time"))]
+            #[cfg(any(feature = "rt", feature = "macros"))]
             rng: Cell::new(None),
 
             budget: Cell::new(coop::Budget::unconstrained()),
@@ -121,11 +121,7 @@ tokio_thread_local! {
     }
 }
 
-#[cfg(any(
-    feature = "time",
-    feature = "macros",
-    all(feature = "sync", feature = "rt")
-))]
+#[cfg(any(feature = "macros", all(feature = "sync", feature = "rt")))]
 pub(crate) fn thread_rng_n(n: u32) -> u32 {
     CONTEXT.with(|ctx| {
         let mut rng = ctx.rng.get().unwrap_or_else(FastRand::new);

tokio/src/runtime/driver.rs

@@ -40,7 +40,6 @@ pub(crate) struct Cfg {
     pub(crate) enable_pause_time: bool,
     pub(crate) start_paused: bool,
     pub(crate) nevents: usize,
-    pub(crate) workers: usize,
 }
 
 impl Driver {
@@ -49,8 +48,7 @@ impl Driver {
 
         let clock = create_clock(cfg.enable_pause_time, cfg.start_paused);
 
-        let (time_driver, time_handle) =
-            create_time_driver(cfg.enable_time, io_stack, &clock, cfg.workers);
+        let (time_driver, time_handle) = create_time_driver(cfg.enable_time, io_stack, &clock);
 
         Ok((
             Self { inner: time_driver },
@@ -297,10 +295,9 @@ cfg_time! {
         enable: bool,
         io_stack: IoStack,
         clock: &Clock,
-        workers: usize,
     ) -> (TimeDriver, TimeHandle) {
         if enable {
-            let (driver, handle) = crate::runtime::time::Driver::new(io_stack, clock, workers as u32);
+            let (driver, handle) = crate::runtime::time::Driver::new(io_stack, clock);
 
             (TimeDriver::Enabled { driver }, Some(handle))
         } else {
@@ -346,7 +343,6 @@ cfg_not_time! {
         _enable: bool,
         io_stack: IoStack,
         _clock: &Clock,
-        _workers: usize,
     ) -> (TimeDriver, TimeHandle) {
         (io_stack, ())
     }

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

@@ -790,11 +790,6 @@ impl Context {
             self.defer.defer(waker);
         }
     }
-
-    #[allow(dead_code)]
-    pub(crate) fn get_worker_index(&self) -> usize {
-        self.worker.index
-    }
 }
 
 impl Core {

tokio/src/runtime/time/entry.rs

@@ -58,7 +58,6 @@ use crate::loom::cell::UnsafeCell;
 use crate::loom::sync::atomic::AtomicU64;
 use crate::loom::sync::atomic::Ordering;
 
-use crate::runtime::context;
 use crate::runtime::scheduler;
 use crate::sync::AtomicWaker;
 use crate::time::Instant;
@@ -329,8 +328,6 @@ pub(super) type EntryList = crate::util::linked_list::LinkedList<TimerShared, Ti
 ///
 /// Note that this structure is located inside the `TimerEntry` structure.
 pub(crate) struct TimerShared {
-    /// The shard id. We should never change it.
-    shard_id: u32,
     /// A link within the doubly-linked list of timers on a particular level and
     /// slot. Valid only if state is equal to Registered.
     ///
@@ -371,9 +368,8 @@ generate_addr_of_methods! {
 }
 
 impl TimerShared {
-    pub(super) fn new(shard_id: u32) -> Self {
+    pub(super) fn new() -> Self {
         Self {
-            shard_id,
             cached_when: AtomicU64::new(0),
             pointers: linked_list::Pointers::new(),
             state: StateCell::default(),
@@ -442,11 +438,6 @@ impl TimerShared {
     pub(super) fn might_be_registered(&self) -> bool {
         self.state.might_be_registered()
     }
-
-    /// Gets the shard id.
-    pub(super) fn shard_id(&self) -> u32 {
-        self.shard_id
-    }
 }
 
 unsafe impl linked_list::Link for TimerShared {
@@ -494,10 +485,8 @@ impl TimerEntry {
     fn inner(&self) -> &TimerShared {
         let inner = unsafe { &*self.inner.get() };
         if inner.is_none() {
-            let shard_size = self.driver.driver().time().inner.get_shard_size();
-            let shard_id = generate_shard_id(shard_size);
             unsafe {
-                *self.inner.get() = Some(TimerShared::new(shard_id));
+                *self.inner.get() = Some(TimerShared::new());
             }
         }
         return inner.as_ref().unwrap();
@@ -654,23 +643,3 @@ impl Drop for TimerEntry {
         unsafe { Pin::new_unchecked(self) }.as_mut().cancel();
     }
 }
-
-// Generates a shard id. If current thread is a worker thread, we use its worker index as a shard id.
-// Otherwise, we use a random number generator to obtain the shard id.
-cfg_rt! {
-    fn generate_shard_id(shard_size: u32) -> u32 {
-        let id = context::with_scheduler(|ctx| match ctx {
-            Some(scheduler::Context::CurrentThread(_ctx)) => 0,
-            #[cfg(feature = "rt-multi-thread")]
-            Some(scheduler::Context::MultiThread(ctx)) => ctx.get_worker_index() as u32,
-            None => context::thread_rng_n(shard_size),
-        });
-        id % shard_size
-    }
-}
-
-cfg_not_rt! {
-    fn generate_shard_id(shard_size: u32) -> u32 {
-        context::thread_rng_n(shard_size)
-    }
-}