From cbade610734c5741191fd232f9554fa144947c13 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Wed, 17 May 2023 15:00:42 -0700 Subject: [PATCH] add backoff to stealing --- .../scheduler/multi_thread/counters.rs | 51 +++++++++++++++++++ .../runtime/scheduler/multi_thread/idle.rs | 2 + .../src/runtime/scheduler/multi_thread/mod.rs | 1 + .../runtime/scheduler/multi_thread/worker.rs | 45 +++++++++++----- tokio/src/runtime/task/atomic_cell.rs | 2 + 5 files changed, 88 insertions(+), 13 deletions(-) create mode 100644 tokio/src/runtime/scheduler/multi_thread/counters.rs diff --git a/tokio/src/runtime/scheduler/multi_thread/counters.rs b/tokio/src/runtime/scheduler/multi_thread/counters.rs new file mode 100644 index 000000000..5bc58d83b --- /dev/null +++ b/tokio/src/runtime/scheduler/multi_thread/counters.rs @@ -0,0 +1,51 @@ +#[cfg(tokio_mt_counters)] +mod imp { + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering::Relaxed; + + static NUM_NOTIFY_LOCAL: AtomicUsize = AtomicUsize::new(0); + static NUM_UNPARKS_LOCAL: AtomicUsize = AtomicUsize::new(0); + static NUM_NEED_SEARCHERS: AtomicUsize = AtomicUsize::new(0); + + impl Drop for super::Counters { + fn drop(&mut self) { + println!("---"); + println!( + "NUM_NOTIFY: {:>10}", + NUM_NOTIFY_LOCAL.load(Relaxed) + ); + println!( + "NUM_UNPARKS: {:>10}", + NUM_UNPARKS_LOCAL.load(Relaxed) + ); + println!( + "NUM_NEED_SEARCHERS: {:>10}", + NUM_NEED_SEARCHERS.load(Relaxed) + ); + } + } + + pub(crate) fn inc_num_need_searchers() { + NUM_NEED_SEARCHERS.fetch_add(1, Relaxed); + } + + 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); + } +} + +#[cfg(not(tokio_mt_counters))] +mod imp { + pub(crate) fn inc_num_need_searchers() {} + pub(crate) fn inc_num_inc_notify_local() {} + pub(crate) fn inc_num_unparks_local() {} +} + +#[derive(Debug)] +pub(crate) struct Counters; + +pub(super) use imp::*; diff --git a/tokio/src/runtime/scheduler/multi_thread/idle.rs b/tokio/src/runtime/scheduler/multi_thread/idle.rs index d0b313cfd..1046a0b6e 100644 --- a/tokio/src/runtime/scheduler/multi_thread/idle.rs +++ b/tokio/src/runtime/scheduler/multi_thread/idle.rs @@ -108,6 +108,8 @@ impl Idle { 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 diff --git a/tokio/src/runtime/scheduler/multi_thread/mod.rs b/tokio/src/runtime/scheduler/multi_thread/mod.rs index 47cd1f3d7..86329ff44 100644 --- a/tokio/src/runtime/scheduler/multi_thread/mod.rs +++ b/tokio/src/runtime/scheduler/multi_thread/mod.rs @@ -1,4 +1,5 @@ //! Multi-threaded runtime +mod counters; mod handle; pub(crate) use handle::Handle; diff --git a/tokio/src/runtime/scheduler/multi_thread/worker.rs b/tokio/src/runtime/scheduler/multi_thread/worker.rs index 1bd2e280c..5b1702fc9 100644 --- a/tokio/src/runtime/scheduler/multi_thread/worker.rs +++ b/tokio/src/runtime/scheduler/multi_thread/worker.rs @@ -144,6 +144,9 @@ 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. @@ -239,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, @@ -621,12 +625,11 @@ impl Core { fn next_task(&mut self, worker: &Worker) -> Option { if self.tick % worker.handle.shared.config.global_queue_interval == 0 { worker - .inject() - .pop() + .next_injected_task() .or_else(|| self.next_local_task(&worker.handle)) } else { self.next_local_task(&worker.handle) - .or_else(|| worker.inject().pop()) + .or_else(|| worker.next_injected_task()) } } @@ -651,6 +654,8 @@ impl Core { // Number of remotes let num = worker.handle.shared.remotes.len(); + let mut backoff = 1; + // Run this a few times for i in 0..ATTEMPTS { // Only try stealing the LIFO slot @@ -675,19 +680,26 @@ impl Core { 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); - } - } + // 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); + // } + // } } + + // Fallback on checking the global queue + if let Some(task) = worker.handle.shared.inject.pop() { + return Some(task); + } + + std::thread::sleep(std::time::Duration::from_micros(backoff)); + backoff *= 2; } - // Fallback on checking the global queue - worker.handle.shared.inject.pop() + None } fn transition_to_searching(&mut self, worker: &Worker) -> bool { @@ -813,6 +825,10 @@ impl Core { } impl Worker { + fn next_injected_task(&self) -> Option { + self.inject().pop() + } + /// Returns a reference to the scheduler's injection queue. fn inject(&self) -> &Inject> { &self.handle.shared.inject @@ -900,7 +916,10 @@ impl Handle { /// 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); } } diff --git a/tokio/src/runtime/task/atomic_cell.rs b/tokio/src/runtime/task/atomic_cell.rs index 737aa302b..71beeb97b 100644 --- a/tokio/src/runtime/task/atomic_cell.rs +++ b/tokio/src/runtime/task/atomic_cell.rs @@ -75,6 +75,8 @@ impl AtomicCell { return None; } + std::thread::sleep(std::time::Duration::from_micros(3)); + // Try to take it once if self .task