itsscb/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2025-09-25 12:00:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

rt(alt): track which workers are idle. (#5886)

Browse Source 
The scheduler uses this map to avoid trying to steal from idle workers.
This commit is contained in:
Carl Lerche 2023-07-21 14:00:33 -07:00 committed by GitHub
parent 4165601b1b
commit a58beb3aca
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 18 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
tokio/src/runtime/scheduler/multi_thread_alt/idle.rs
View File

@ -16,7 +16,7 @@ pub(super) struct Idle {
num_idle: AtomicUsize,
/// Map of idle cores
// idle_map: IdleMap,
idle_map: IdleMap,
/// Used to catch false-negatives when waking workers
needs_searching: AtomicBool,
@ -30,7 +30,7 @@ pub(super) struct IdleMap {
}
pub(super) struct Snapshot {
// chunks: Vec<usize>,
chunks: Vec<usize>,
}
/// Data synchronized by the scheduler mutex
@ -47,7 +47,7 @@ impl Idle {
let idle = Idle {
num_searching: AtomicUsize::new(0),
num_idle: AtomicUsize::new(cores.len()),
// idle_map: IdleMap::new(&cores),
idle_map: IdleMap::new(&cores),
needs_searching: AtomicBool::new(false),
num_cores: cores.len(),
};
@ -69,22 +69,22 @@ impl Idle {
self.num_searching.load(Acquire)
}
pub(super) fn snapshot(&self, _snapshot: &mut Snapshot) {
// snapshot.update(&self.idle_map)
pub(super) fn snapshot(&self, snapshot: &mut Snapshot) {
snapshot.update(&self.idle_map)
}
/// Try to acquire an available core
pub(super) fn try_acquire_available_core(&self, synced: &mut Synced) -> Option<Box<Core>> {
let ret = synced.available_cores.pop();
if let Some(_core) = &ret {
if let Some(core) = &ret {
// Decrement the number of idle cores
let num_idle = self.num_idle.load(Acquire) - 1;
debug_assert_eq!(num_idle, synced.available_cores.len());
self.num_idle.store(num_idle, Release);
// self.idle_map.unset(core.index);
// debug_assert!(self.idle_map.matches(&synced.available_cores));
self.idle_map.unset(core.index);
debug_assert!(self.idle_map.matches(&synced.available_cores));
}
ret
@ -151,8 +151,8 @@ impl Idle {
debug_assert!(!core.is_searching);
core.is_searching = true;
// self.idle_map.unset(core.index);
// debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
self.idle_map.unset(core.index);
debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
// Assign the core to the worker
synced.assigned_cores[worker] = Some(core);
@ -201,7 +201,7 @@ impl Idle {
if let Some(core) = synced.idle.available_cores.pop() {
debug_assert!(!core.is_searching);
// self.idle_map.unset(core.index);
self.idle_map.unset(core.index);
synced.assigned_cores[worker] = Some(core);
@ -217,7 +217,7 @@ impl Idle {
}
if !workers.is_empty() {
// debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
let num_idle = synced.idle.available_cores.len();
self.num_idle.store(num_idle, Release);
} else {
@ -237,7 +237,7 @@ impl Idle {
let worker = synced.idle.sleepers.pop().unwrap();
let core = synced.idle.available_cores.pop().unwrap();
// self.idle_map.unset(core.index);
self.idle_map.unset(core.index);
synced.assigned_cores[worker] = Some(core);
shared.condvars[worker].notify_one();
@ -246,7 +246,7 @@ impl Idle {
.store(synced.idle.available_cores.len(), Release);
}
// debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
// Wake up any other workers
while let Some(index) = synced.idle.sleepers.pop() {
@ -267,12 +267,12 @@ impl Idle {
let num_idle = synced.idle.available_cores.len();
debug_assert_eq!(num_idle, self.num_idle.load(Acquire));
// self.idle_map.set(core.index);
self.idle_map.set(core.index);
// Store the core in the list of available cores
synced.idle.available_cores.push(core);
// debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
debug_assert!(self.idle_map.matches(&synced.idle.available_cores));
// Update `num_idle`
self.num_idle.store(num_idle + 1, Release);
@ -386,25 +386,19 @@ impl IdleMap {
}
impl Snapshot {
pub(crate) fn new(_idle: &Idle) -> Snapshot {
/*
pub(crate) fn new(idle: &Idle) -> Snapshot {
let chunks = vec![0; idle.idle_map.chunks.len()];
let mut ret = Snapshot { chunks };
ret.update(&idle.idle_map);
ret
*/
Snapshot {}
}
fn update(&mut self, _idle_map: &IdleMap) {
/*
fn update(&mut self, idle_map: &IdleMap) {
for i in 0..self.chunks.len() {
self.chunks[i] = idle_map.chunks[i].load(Acquire);
}
*/
}
/*
pub(super) fn is_idle(&self, index: usize) -> bool {
let (chunk, mask) = index_to_mask(index);
debug_assert!(
@ -415,7 +409,6 @@ impl Snapshot {
);
self.chunks[chunk] & mask == mask
}
*/
}
fn num_chunks(max_cores: usize) -> usize {

2
tokio/src/runtime/scheduler/multi_thread_alt/worker.rs
View File

@ -865,12 +865,10 @@ impl Worker {
continue;
}
/*
// If the core is currently idle, then there is nothing to steal.
if self.idle_snapshot.is_idle(i) {
continue;
}
*/
let target = &cx.shared().remotes[i];