squiid/rust
1
0
Fork 0
Code Issues Pull requests Projects Activity

rework and document backoff behavior of sync::mpsc

Browse source
This commit is contained in:
Ibraheem Ahmed 2023-01-11 21:29:14 -05:00
parent f8276c94ac
commit 8917e99362
4 changed files with 30 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
library/std/src/sync/mpmc/array.rs
View file

@ -168,7 +168,7 @@ impl<T> Channel<T> {
return true;
}
Err(_) => {
backoff.spin();
backoff.spin_light();
tail = self.tail.load(Ordering::Relaxed);
}
}
@ -182,11 +182,11 @@ impl<T> Channel<T> {
return false;
}
backoff.spin();
backoff.spin_light();
tail = self.tail.load(Ordering::Relaxed);
} else {
// Snooze because we need to wait for the stamp to get updated.
backoff.snooze();
backoff.spin_heavy();
tail = self.tail.load(Ordering::Relaxed);
}
}
@ -251,7 +251,7 @@ impl<T> Channel<T> {
return true;
}
Err(_) => {
backoff.spin();
backoff.spin_light();
head = self.head.load(Ordering::Relaxed);
}
}
@ -273,11 +273,11 @@ impl<T> Channel<T> {
}
}
backoff.spin();
backoff.spin_light();
head = self.head.load(Ordering::Relaxed);
} else {
// Snooze because we need to wait for the stamp to get updated.
backoff.snooze();
backoff.spin_heavy();
head = self.head.load(Ordering::Relaxed);
}
}
@ -330,7 +330,7 @@ impl<T> Channel<T> {
if backoff.is_completed() {
break;
} else {
backoff.spin();
backoff.spin_light();
}
}

16
library/std/src/sync/mpmc/list.rs
View file

@ -46,7 +46,7 @@ impl<T> Slot<T> {
fn wait_write(&self) {
let backoff = Backoff::new();
while self.state.load(Ordering::Acquire) & WRITE == 0 {
backoff.snooze();
backoff.spin_heavy();
}
}
}
@ -82,7 +82,7 @@ impl<T> Block<T> {
if !next.is_null() {
return next;
}
backoff.snooze();
backoff.spin_heavy();
}
}
@ -191,7 +191,7 @@ impl<T> Channel<T> {
// If we reached the end of the block, wait until the next one is installed.
if offset == BLOCK_CAP {
backoff.snooze();
backoff.spin_heavy();
tail = self.tail.index.load(Ordering::Acquire);
block = self.tail.block.load(Ordering::Acquire);
continue;
@ -247,7 +247,7 @@ impl<T> Channel<T> {
return true;
},
Err(_) => {
backoff.spin();
backoff.spin_light();
tail = self.tail.index.load(Ordering::Acquire);
block = self.tail.block.load(Ordering::Acquire);
}
@ -286,7 +286,7 @@ impl<T> Channel<T> {
// If we reached the end of the block, wait until the next one is installed.
if offset == BLOCK_CAP {
backoff.snooze();
backoff.spin_heavy();
head = self.head.index.load(Ordering::Acquire);
block = self.head.block.load(Ordering::Acquire);
continue;
@ -320,7 +320,7 @@ impl<T> Channel<T> {
// The block can be null here only if the first message is being sent into the channel.
// In that case, just wait until it gets initialized.
if block.is_null() {
backoff.snooze();
backoff.spin_heavy();
head = self.head.index.load(Ordering::Acquire);
block = self.head.block.load(Ordering::Acquire);
continue;
@ -351,7 +351,7 @@ impl<T> Channel<T> {
return true;
},
Err(_) => {
backoff.spin();
backoff.spin_light();
head = self.head.index.load(Ordering::Acquire);
block = self.head.block.load(Ordering::Acquire);
}
@ -542,7 +542,7 @@ impl<T> Channel<T> {
// New updates to tail will be rejected by MARK_BIT and aborted unless it's
// at boundary. We need to wait for the updates take affect otherwise there
// can be memory leaks.
backoff.snooze();
backoff.spin_heavy();
tail = self.tail.index.load(Ordering::Acquire);
}

29
library/std/src/sync/mpmc/utils.rs
View file

@ -91,9 +91,8 @@ impl<T> DerefMut for CachePadded<T> {
}
const SPIN_LIMIT: u32 = 6;
const YIELD_LIMIT: u32 = 10;
/// Performs exponential backoff in spin loops.
/// Performs quadratic backoff in spin loops.
pub struct Backoff {
step: Cell<u32>,
}
@ -104,25 +103,27 @@ impl Backoff {
Backoff { step: Cell::new(0) }
}
/// Backs off in a lock-free loop.
/// Backs off using lightweight spinning.
///
/// This method should be used when we need to retry an operation because another thread made
/// progress.
/// This method should be used for:
/// - Retrying an operation because another thread made progress. i.e. on CAS failure.
/// - Waiting for an operation to complete by spinning optimistically for a few iterations
/// before falling back to parking the thread (see `Backoff::is_completed`).
#[inline]
pub fn spin(&self) {
pub fn spin_light(&self) {
let step = self.step.get().min(SPIN_LIMIT);
for _ in 0..step.pow(2) {
crate::hint::spin_loop();
}
if self.step.get() <= SPIN_LIMIT {
self.step.set(self.step.get() + 1);
}
self.step.set(self.step.get() + 1);
}
/// Backs off in a blocking loop.
/// Backs off using heavyweight spinning.
///
/// This method should be used in blocking loops where parking the thread is not an option.
#[inline]
pub fn snooze(&self) {
pub fn spin_heavy(&self) {
if self.step.get() <= SPIN_LIMIT {
for _ in 0..self.step.get().pow(2) {
crate::hint::spin_loop()
@ -131,12 +132,10 @@ impl Backoff {
crate::thread::yield_now();
}
if self.step.get() <= YIELD_LIMIT {
self.step.set(self.step.get() + 1);
}
self.step.set(self.step.get() + 1);
}
/// Returns `true` if quadratic backoff has completed and blocking the thread is advised.
/// Returns `true` if quadratic backoff has completed and parking the thread is advised.
#[inline]
pub fn is_completed(&self) -> bool {
self.step.get() > SPIN_LIMIT

2
library/std/src/sync/mpmc/zero.rs
View file

@ -57,7 +57,7 @@ impl<T> Packet<T> {
fn wait_ready(&self) {
let backoff = Backoff::new();
while !self.ready.load(Ordering::Acquire) {
backoff.snooze();
backoff.spin_heavy();
}
}
}