From 42251fafe727a5aa63d8e555963122fd94d9a629 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Wed, 17 May 2023 10:56:29 -0700 Subject: [PATCH] remove atomic-ness --- .../runtime/scheduler/multi_thread/worker.rs | 52 +++++-------------- tokio/src/runtime/task/atomic_cell.rs | 22 ++++++-- 2 files changed, 32 insertions(+), 42 deletions(-) diff --git a/tokio/src/runtime/scheduler/multi_thread/worker.rs b/tokio/src/runtime/scheduler/multi_thread/worker.rs index 93f4dc99e..75d701b0f 100644 --- a/tokio/src/runtime/scheduler/multi_thread/worker.rs +++ b/tokio/src/runtime/scheduler/multi_thread/worker.rs @@ -626,53 +626,29 @@ impl Core { .or_else(|| self.run_queue.pop()) } - /// Function responsible for stealing tasks from another worker - /// - /// Note: Only if less than half the workers are searching for tasks to steal - /// a new worker will actually try to steal. The idea is to make sure not all - /// workers will be trying to steal at the same time. fn steal_work(&mut self, worker: &Worker) -> Option { - const ATTEMPTS: usize = 4; - if !self.transition_to_searching(worker) { return None; } - // Number of remotes let num = worker.handle.shared.remotes.len(); + // Start from a random worker + let start = self.rand.fastrand_n(num as u32) as usize; - // Run this a few times - for i in 0..ATTEMPTS { - // Only try stealing the LIFO slot - let steal_lifo = i == ATTEMPTS - 1; + for i in 0..num { + let i = (start + i) % num; - // Start from a random worker - let start = self.rand.fastrand_n(num as u32) as usize; + // Don't steal from ourself! We know we don't have work. + if i == worker.index { + continue; + } - for i in 0..num { - let i = (start + i) % num; - - // Don't steal from ourself! We know we don't have work. - if i == worker.index { - continue; - } - - let target = &worker.handle.shared.remotes[i]; - if let Some(task) = target - .steal - .steal_into(&mut self.run_queue, &mut self.metrics) - { - 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); - } - } + let target = &worker.handle.shared.remotes[i]; + if let Some(task) = target + .steal + .steal_into(&mut self.run_queue, &mut self.metrics) + { + return Some(task); } } diff --git a/tokio/src/runtime/task/atomic_cell.rs b/tokio/src/runtime/task/atomic_cell.rs index 54b00d868..57e421ce9 100644 --- a/tokio/src/runtime/task/atomic_cell.rs +++ b/tokio/src/runtime/task/atomic_cell.rs @@ -5,25 +5,33 @@ use crate::runtime::task::{Header, Notified}; use std::marker::PhantomData; use std::ptr::{self, NonNull}; -pub(crate) struct AtomicCell { - task: AtomicPtr
, +pub(crate) struct AtomicCell { + // task: AtomicPtr
, + task: std::cell::UnsafeCell>>, _p: PhantomData, } +unsafe impl Send for AtomicCell {} +unsafe impl Sync for AtomicCell {} + impl AtomicCell { pub(crate) fn new() -> AtomicCell { AtomicCell { - task: AtomicPtr::default(), + // task: AtomicPtr::default(), + task: Default::default(), _p: PhantomData, } } /// Should be called from a local context pub(crate) fn is_some(&self) -> bool { - !self.task.load(Acquire).is_null() + // !self.task.load(Acquire).is_null() + unsafe { (*self.task.get()).is_some() } } pub(crate) fn take_local(&self) -> Option> { + unsafe { (*self.task.get()).take() } + /* let ptr = self.task.load(Acquire); if ptr.is_null() { @@ -39,9 +47,12 @@ impl AtomicCell { } NonNull::new(ptr).map(|ptr| unsafe { Notified::from_raw(ptr) }) + */ } pub(crate) fn swap_local(&self, task: Notified) -> Option> { + std::mem::replace(unsafe { &mut (*self.task.get()) }, Some(task)) + /* let next = task.into_raw().as_ptr(); let prev = self.task.load(Acquire); @@ -66,8 +77,10 @@ impl AtomicCell { // this is the only thread that could set the cell to !null. self.task.store(next, Release); None + */ } + /* pub(crate) fn take_remote(&self) -> Option> { let task = self.task.load(Acquire); @@ -87,4 +100,5 @@ impl AtomicCell { return None; } + */ }