From ec3570ecf54e9285ea6bb07b2c3d3e8acab72526 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Thu, 22 Jun 2023 00:32:32 +0000 Subject: [PATCH] wip --- .../runtime/scheduler/multi_thread/counters.rs | 16 ++++++++++++++++ tokio/src/runtime/scheduler/multi_thread/idle.rs | 2 ++ .../src/runtime/scheduler/multi_thread/queue.rs | 1 + .../src/runtime/scheduler/multi_thread/worker.rs | 13 +++++++------ tokio/src/runtime/task/list.rs | 1 + tokio/src/util/atomic_cell.rs | 4 ++++ 6 files changed, 31 insertions(+), 6 deletions(-) diff --git a/tokio/src/runtime/scheduler/multi_thread/counters.rs b/tokio/src/runtime/scheduler/multi_thread/counters.rs index 2284b5ffc..8a74882ef 100644 --- a/tokio/src/runtime/scheduler/multi_thread/counters.rs +++ b/tokio/src/runtime/scheduler/multi_thread/counters.rs @@ -7,6 +7,7 @@ mod imp { 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); @@ -14,12 +15,14 @@ mod imp { 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); 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); @@ -28,11 +31,13 @@ mod imp { 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); println!("---"); println!("notifies (remote): {}", notifies_remote); println!(" notifies (local): {}", notifies_local); println!(" unparks (local): {}", unparks_local); + println!(" unparks (remote): {}", unparks_remote); println!(" maintenance: {}", maintenance); println!(" LIFO schedules: {}", lifo_scheds); println!(" LIFO capped: {}", lifo_capped); @@ -41,6 +46,7 @@ mod imp { println!(" parks: {}", num_park); println!(" polls: {}", num_polls); println!(" polls (LIFO): {}", num_lifo_polls); + println!("remote task batch: {}", num_remote_batch); } } @@ -56,6 +62,10 @@ mod imp { 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); } @@ -87,6 +97,10 @@ mod imp { 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); + } } #[cfg(not(tokio_internal_mt_counters))] @@ -94,6 +108,7 @@ 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() {} @@ -102,6 +117,7 @@ mod imp { 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() {} } #[derive(Debug)] diff --git a/tokio/src/runtime/scheduler/multi_thread/idle.rs b/tokio/src/runtime/scheduler/multi_thread/idle.rs index ee38761f8..664878998 100644 --- a/tokio/src/runtime/scheduler/multi_thread/idle.rs +++ b/tokio/src/runtime/scheduler/multi_thread/idle.rs @@ -164,6 +164,8 @@ impl Idle { // Drop the lock before notifying the condvar. drop(synced); + super::counters::inc_num_unparks_remote(); + // Notify the worker shared.condvars[worker].notify_one(); return; diff --git a/tokio/src/runtime/scheduler/multi_thread/queue.rs b/tokio/src/runtime/scheduler/multi_thread/queue.rs index b12b6584f..dc34b7ed4 100644 --- a/tokio/src/runtime/scheduler/multi_thread/queue.rs +++ b/tokio/src/runtime/scheduler/multi_thread/queue.rs @@ -33,6 +33,7 @@ pub(crate) struct Local { /// Consumer handle. May be used from many threads. pub(crate) struct Steal(Arc>); +#[repr(align(128))] pub(crate) struct Inner { /// Concurrently updated by many threads. /// diff --git a/tokio/src/runtime/scheduler/multi_thread/worker.rs b/tokio/src/runtime/scheduler/multi_thread/worker.rs index 58211d0a4..8a0b206da 100644 --- a/tokio/src/runtime/scheduler/multi_thread/worker.rs +++ b/tokio/src/runtime/scheduler/multi_thread/worker.rs @@ -712,7 +712,6 @@ impl Worker { if !cx.defer.borrow().is_empty() { core = try_task_new_batch!(self, self.park_yield(cx, core)); } else { - super::counters::inc_num_parks(); core = try_task_new_batch!(self, self.park(cx, core)); } } @@ -780,6 +779,8 @@ impl Worker { core: &mut Core, max: usize, ) -> Option { + super::counters::inc_num_remote_batch(); + // The worker is currently idle, pull a batch of work from the // injection queue. We don't want to pull *all* the work so other // workers can also get some. @@ -1143,11 +1144,6 @@ impl Worker { } fn do_park(&mut self, cx: &Context, mut core: Box) -> NextTaskResult { - core.stats.about_to_park(); - - // Flush metrics to the runtime metrics aggregator - self.flush_metrics(cx, &mut core); - let was_searching = core.is_searching; // Before we park, if we are searching, we need to transition away from searching @@ -1181,6 +1177,11 @@ impl Worker { } } + super::counters::inc_num_parks(); + core.stats.about_to_park(); + // Flush metrics to the runtime metrics aggregator + self.flush_metrics(cx, &mut core); + // If the runtime is shutdown, skip parking self.update_global_flags(cx, &mut synced, &mut core); diff --git a/tokio/src/runtime/task/list.rs b/tokio/src/runtime/task/list.rs index da9ea92a0..448310f64 100644 --- a/tokio/src/runtime/task/list.rs +++ b/tokio/src/runtime/task/list.rs @@ -53,6 +53,7 @@ cfg_not_has_atomic_u64! { } } +#[repr(align(128))] pub(crate) struct OwnedTasks { inner: Mutex>, id: u64, diff --git a/tokio/src/util/atomic_cell.rs b/tokio/src/util/atomic_cell.rs index e3cc2dd1a..d889654a7 100644 --- a/tokio/src/util/atomic_cell.rs +++ b/tokio/src/util/atomic_cell.rs @@ -27,6 +27,10 @@ impl AtomicCell { } pub(crate) fn take(&self) -> Option> { + if self.data.load(Acquire).is_null() { + return None; + } + self.swap(None) }