From 8917e9936282f855a08808ed8874c4117210da6e Mon Sep 17 00:00:00 2001 From: Ibraheem Ahmed Date: Wed, 11 Jan 2023 21:29:14 -0500 Subject: [PATCH] rework and document backoff behavior of `sync::mpsc` --- library/std/src/sync/mpmc/array.rs | 14 +++++++------- library/std/src/sync/mpmc/list.rs | 16 ++++++++-------- library/std/src/sync/mpmc/utils.rs | 29 ++++++++++++++--------------- library/std/src/sync/mpmc/zero.rs | 2 +- 4 files changed, 30 insertions(+), 31 deletions(-) diff --git a/library/std/src/sync/mpmc/array.rs b/library/std/src/sync/mpmc/array.rs index f71edc6c525..c1e3e48b044 100644 --- a/library/std/src/sync/mpmc/array.rs +++ b/library/std/src/sync/mpmc/array.rs @@ -168,7 +168,7 @@ impl Channel { return true; } Err(_) => { - backoff.spin(); + backoff.spin_light(); tail = self.tail.load(Ordering::Relaxed); } } @@ -182,11 +182,11 @@ impl Channel { 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 Channel { return true; } Err(_) => { - backoff.spin(); + backoff.spin_light(); head = self.head.load(Ordering::Relaxed); } } @@ -273,11 +273,11 @@ impl Channel { } } - 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 Channel { if backoff.is_completed() { break; } else { - backoff.spin(); + backoff.spin_light(); } } diff --git a/library/std/src/sync/mpmc/list.rs b/library/std/src/sync/mpmc/list.rs index 2d5b2fb3b23..ec6c0726ac7 100644 --- a/library/std/src/sync/mpmc/list.rs +++ b/library/std/src/sync/mpmc/list.rs @@ -46,7 +46,7 @@ impl Slot { 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 Block { if !next.is_null() { return next; } - backoff.snooze(); + backoff.spin_heavy(); } } @@ -191,7 +191,7 @@ impl Channel { // 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 Channel { 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 Channel { // 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 Channel { // 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 Channel { 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 Channel { // 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); } diff --git a/library/std/src/sync/mpmc/utils.rs b/library/std/src/sync/mpmc/utils.rs index a9b365daeec..cfe42750d52 100644 --- a/library/std/src/sync/mpmc/utils.rs +++ b/library/std/src/sync/mpmc/utils.rs @@ -91,9 +91,8 @@ impl DerefMut for CachePadded { } 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, } @@ -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 diff --git a/library/std/src/sync/mpmc/zero.rs b/library/std/src/sync/mpmc/zero.rs index fccd6c29a7e..33f768dcbe9 100644 --- a/library/std/src/sync/mpmc/zero.rs +++ b/library/std/src/sync/mpmc/zero.rs @@ -57,7 +57,7 @@ impl Packet { fn wait_ready(&self) { let backoff = Backoff::new(); while !self.ready.load(Ordering::Acquire) { - backoff.snooze(); + backoff.spin_heavy(); } } }