wip

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)]

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;

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.
     ///

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<Notified> {
+        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<Core>) -> 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);
 

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,

tokio/src/util/atomic_cell.rs

@@ -27,6 +27,10 @@ impl<T> AtomicCell<T> {
     }
 
     pub(crate) fn take(&self) -> Option<Box<T>> {
+        if self.data.load(Acquire).is_null() {
+            return None;
+        }
+
         self.swap(None)
     }