use core::task::{Poll, Waker};
use super::internal::Ordering;
use super::queue::{HANDOFF, Policy, stops_spin};
use super::spin::Gaps;
use super::waiter::{Cancel, Park, Wait};
#[cfg(not(all(loom, test)))]
pub const MAX_BYPASS: u32 = 16;
#[cfg(all(loom, test))]
pub const MAX_BYPASS: u32 = 2;
pub(super) const CLOSED_GRANT: usize = usize::MAX;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct Closed;
pub(super) trait Lock: Policy {
const SPINS: bool = true;
fn admit(&self, state: usize, need: usize) -> Option<usize>;
fn give_back(&self, need: usize);
fn closed(&self, _state: usize) -> bool {
false
}
fn try_take(&self, need: usize) -> Result<(), usize> {
take(self, need, false)
}
fn take_for_waiter(&self, need: usize) -> bool {
take(self, need, true).is_ok()
}
fn take_or_owe(&self, wait: &Wait, need: usize) -> Result<(), usize> {
self.owe_or_take(wait, |state| self.admit(state, need))
}
}
pub(super) fn take<L: Lock + ?Sized>(lock: &L, need: usize, for_waiter: bool) -> Result<(), usize> {
let failure = if for_waiter {
Ordering::Acquire
} else {
Ordering::Relaxed
};
take_from(lock, need, for_waiter, lock.queue().state.load(failure))
}
pub(super) fn take_from<L: Lock + ?Sized>(
lock: &L,
need: usize,
for_waiter: bool,
mut current: usize,
) -> Result<(), usize> {
let (success, failure) = if for_waiter {
(Ordering::AcqRel, Ordering::Acquire)
} else {
(Ordering::Acquire, Ordering::Relaxed)
};
let state = &lock.queue().state;
loop {
let next = match lock.admit(current, need) {
Some(next) if for_waiter || current & HANDOFF == 0 => next,
_ => return Err(current),
};
match state.compare_exchange_weak(current, next, success, failure) {
Ok(_) => return Ok(()),
Err(actual) => current = actual,
}
}
}
fn spin_take<L: Lock>(lock: &L, need: usize, mut seen: usize) -> bool {
let queue = lock.queue();
let rounds = queue.spin.rounds();
if rounds == 0 {
return false;
}
let mut gaps = Gaps::new(rounds);
let mut freed = false;
while gaps.wait() {
if stops_spin(seen) {
return false;
}
seen = queue.state.load(Ordering::Relaxed);
if !stops_spin(seen)
&& let Some(next) = lock.admit(seen, need)
{
freed = true;
match queue
.state
.compare_exchange(seen, next, Ordering::Acquire, Ordering::Relaxed)
{
Ok(_) => {
queue.spin.won(rounds);
return true;
}
Err(actual) => seen = actual,
}
}
}
if !freed {
queue.spin.ran_out(rounds);
}
false
}
pub(super) struct Contend {
wait: Wait,
bypass: u32,
}
impl Contend {
pub(super) const fn new() -> Self {
Self {
wait: Wait::new(),
bypass: 0,
}
}
#[inline]
pub(super) fn is_queued(&self) -> bool {
self.wait.is_queued()
}
#[inline]
pub(super) fn poll<L: Lock>(&mut self, lock: &L, need: usize, waker: &Waker) -> Poll<()> {
self.poll_from(lock, need, waker, None)
}
#[inline]
pub(super) fn poll_from<L: Lock>(
&mut self,
lock: &L,
need: usize,
waker: &Waker,
seen: Option<usize>,
) -> Poll<()> {
self.poll_closable_from(lock, need, waker, seen)
.map(|taken| debug_assert!(taken.is_ok(), "a lock that never closes was closed"))
}
pub(super) fn poll_closable<L: Lock>(
&mut self,
lock: &L,
need: usize,
waker: &Waker,
) -> Poll<Result<(), Closed>> {
self.poll_closable_from(lock, need, waker, None)
}
fn poll_closable_from<L: Lock>(
&mut self,
lock: &L,
need: usize,
waker: &Waker,
seen: Option<usize>,
) -> Poll<Result<(), Closed>> {
let queue = lock.queue();
let mut park = if self.wait.is_queued() {
self.wait.park(queue, Some(waker))
} else {
let first = match seen {
Some(seen) => take_from(lock, need, false, seen),
None => lock.try_take(need),
};
match first {
Ok(()) => return Poll::Ready(Ok(())),
Err(seen) => {
if lock.closed(seen) {
return Poll::Ready(Err(Closed));
}
if L::SPINS && spin_take(lock, need, seen) {
return Poll::Ready(Ok(()));
}
}
}
lock.enqueue(&mut self.wait, need, waker);
self.wait.park(queue, None)
};
loop {
match park {
Park::Granted(payload) => {
queue.consume_nudge();
return Poll::Ready(if payload == CLOSED_GRANT {
Err(Closed)
} else {
Ok(())
});
}
Park::Pending => return Poll::Pending,
Park::Nudged => {
queue.consume_nudge();
let attempt = if self.bypass + 1 == MAX_BYPASS {
lock.take_or_owe(&self.wait, need)
} else {
lock.try_take(need)
};
match attempt {
Ok(()) => {
if L::SPINS && queue.spin.rounds() == 0 {
queue.spin.won(0);
}
self.cancel(lock, need);
return Poll::Ready(Ok(()));
}
Err(seen) if seen & HANDOFF == 0 => {
self.bypass = (self.bypass + 1).min(MAX_BYPASS);
}
Err(_) => {}
}
park = if L::SPINS {
self.wait.spin_park(queue, waker)
} else {
self.wait.park(queue, Some(waker))
};
}
}
}
}
pub(super) fn cancel<L: Lock>(&mut self, lock: &L, need: usize) {
match self.wait.cancel(lock.queue()) {
Cancel::Idle => {}
Cancel::Withdrawn(woken) => lock.withdrawn(woken),
Cancel::Granted(payload) => {
lock.queue().consume_nudge();
if payload != CLOSED_GRANT {
lock.give_back(need);
}
}
}
}
}
#[cfg(all(test, not(loom)))]
mod tests {
use core::cell::Cell;
use super::super::queue::{Arrivals, WaitQueue};
use super::super::raw_semaphore::RawSemaphore;
use super::super::test_support::Polled;
use super::super::waiter::WakeList;
use super::*;
struct Gap<'a> {
sem: &'a RawSemaphore,
after_loss: Cell<Option<Box<dyn FnOnce() + 'a>>>,
}
impl Gap<'_> {
fn lost(&self, attempt: Result<(), usize>) -> Result<(), usize> {
if attempt.is_err()
&& let Some(between) = self.after_loss.take()
{
between();
}
attempt
}
}
impl Policy for Gap<'_> {
fn queue(&self) -> &WaitQueue {
self.sem.queue()
}
fn pass(&self, arrivals: Arrivals, woken: &mut WakeList<'_>, wake_front: bool) -> usize {
self.sem.pass(arrivals, woken, wake_front)
}
}
impl Lock for Gap<'_> {
fn admit(&self, state: usize, need: usize) -> Option<usize> {
self.sem.admit(state, need)
}
fn give_back(&self, need: usize) {
self.sem.give_back(need);
}
fn try_take(&self, need: usize) -> Result<(), usize> {
self.lost(self.sem.try_take(need))
}
fn take_or_owe(&self, wait: &Wait, need: usize) -> Result<(), usize> {
self.lost(self.sem.take_or_owe(wait, need))
}
}
#[test]
fn nothing_barges_between_a_waiters_last_loss_and_its_handoff() {
let sem = RawSemaphore::new(1);
let barged = Cell::new(false);
let gap = Gap {
sem: &sem,
after_loss: Cell::new(None),
};
assert!(sem.try_acquire(1));
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&gap, 1, cx.waker())
}));
waiter.pending();
for round in 1..=MAX_BYPASS {
sem.release(1);
assert_eq!(waiter.wakes() as u32, round, "a release nudges it");
assert!(sem.try_acquire(1), "a free permit is taken past it");
if round < MAX_BYPASS {
waiter.pending();
}
}
gap.after_loss.set(Some(Box::new(|| {
sem.release(1);
barged.set(sem.try_acquire(1));
})));
let served = waiter.poll().is_ready();
assert!(
!barged.get(),
"a barger took the permit freed after the waiter's last counted loss"
);
assert!(served, "the freed permit is handed to the waiter");
assert!(!sem.try_acquire(1), "the waiter holds it");
sem.release(1);
assert!(sem.try_acquire(1), "barging resumes once it is served");
}
struct Raced<'a> {
sem: &'a RawSemaphore,
passes: Cell<u32>,
pass_begins: Box<dyn Fn(u32) + 'a>,
pass_ends: Box<dyn Fn(u32) + 'a>,
before_try: Cell<Option<Box<dyn FnOnce() + 'a>>>,
before_owe: Cell<Option<Box<dyn FnOnce() + 'a>>>,
}
impl<'a> Raced<'a> {
fn new(
sem: &'a RawSemaphore,
pass_begins: impl Fn(u32) + 'a,
pass_ends: impl Fn(u32) + 'a,
) -> Self {
Self {
sem,
passes: Cell::new(0),
pass_begins: Box::new(pass_begins),
pass_ends: Box::new(pass_ends),
before_try: Cell::new(None),
before_owe: Cell::new(None),
}
}
}
impl Policy for Raced<'_> {
fn queue(&self) -> &WaitQueue {
self.sem.queue()
}
fn pass(&self, arrivals: Arrivals, woken: &mut WakeList<'_>, wake_front: bool) -> usize {
let pass = self.passes.get() + 1;
self.passes.set(pass);
(self.pass_begins)(pass);
let clear = self.sem.pass(arrivals, woken, wake_front);
(self.pass_ends)(pass);
clear
}
}
impl Lock for Raced<'_> {
fn admit(&self, state: usize, need: usize) -> Option<usize> {
self.sem.admit(state, need)
}
fn give_back(&self, need: usize) {
self.sem.give_back(need);
}
fn try_take(&self, need: usize) -> Result<(), usize> {
if let Some(first) = self.before_try.take() {
first();
}
self.sem.try_take(need)
}
fn take_or_owe(&self, wait: &Wait, need: usize) -> Result<(), usize> {
if let Some(first) = self.before_owe.take() {
first();
}
self.sem.take_or_owe(wait, need)
}
}
#[test]
fn a_waiter_serving_the_queue_is_not_passed_over_without_bound() {
let sem = RawSemaphore::new(1);
let barged = Cell::new(0u32);
let rounds = 4 * MAX_BYPASS;
let raced = Raced::new(
&sem,
|pass| {
if pass == 1 {
sem.release(1);
}
},
|pass| {
if pass <= rounds && sem.try_acquire(1) {
barged.set(barged.get() + 1);
sem.release(1);
}
},
);
assert!(sem.try_acquire(1));
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&raced, 1, cx.waker())
}));
let served = waiter.poll().is_ready();
assert!(
barged.get() <= MAX_BYPASS,
"{} callers took the permit past a queued waiter in {} passes of its own drain, \
and it never got to retry (the bound is {MAX_BYPASS})",
barged.get(),
raced.passes.get(),
);
assert!(
raced.passes.get() <= 3,
"the waiter's drain ran {} passes: releases that come while its nudge is out \
must not ask for more",
raced.passes.get(),
);
assert!(
served,
"the permit the last caller gave back is the waiter's"
);
assert!(!sem.try_acquire(1), "the waiter holds it");
}
#[test]
fn a_retry_lost_in_the_poll_that_queues_counts_as_a_bypass() {
let sem = RawSemaphore::new(1);
let raced = Raced::new(
&sem,
|pass| {
if pass == 1 {
sem.release(1);
}
},
|pass| {
if pass == 1 {
assert!(
sem.try_acquire(1),
"the freed permit is taken past the waiter"
);
}
},
);
assert!(sem.try_acquire(1));
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&raced, 1, cx.waker())
}));
waiter.pending();
drop(waiter);
assert_eq!(
contend.bypass, 1,
"the loss in the queueing poll is counted"
);
contend.cancel(&raced, 1);
sem.release(1);
assert!(sem.try_acquire(1), "nothing is left held or owed");
}
#[test]
fn a_loss_to_its_own_handoff_does_not_count_past_the_bound() {
let sem = RawSemaphore::new(1);
let raced = Raced::new(&sem, |_| {}, |_| {});
assert!(sem.try_acquire(1));
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&raced, 1, cx.waker())
}));
waiter.pending();
for _ in 1..MAX_BYPASS {
sem.release(1);
assert!(sem.try_acquire(1), "a free permit is taken past it");
waiter.pending();
}
sem.release(1);
assert!(sem.try_acquire(1), "a free permit is taken past it");
raced.before_owe.set(Some(Box::new(|| {
sem.release(1);
assert!(sem.try_acquire(1), "a free permit is taken past it");
})));
raced.before_try.set(Some(Box::new(|| sem.release(1))));
assert!(waiter.poll().is_ready(), "the handoff reached the waiter");
drop(waiter);
assert_eq!(
contend.bypass, MAX_BYPASS,
"a loss to the permit taken for the waiter itself was counted"
);
assert!(!sem.try_acquire(1), "the waiter holds it");
}
#[test]
fn a_drain_that_found_nobody_queued_does_not_pass_again_for_releases() {
let sem = RawSemaphore::new(2);
let hammering = Cell::new(false);
let rounds = 4 * MAX_BYPASS;
let raced = Raced::new(
&sem,
|_| {},
|pass| {
if hammering.get() && pass <= rounds {
sem.release(1);
assert!(sem.try_acquire(1), "the permit is free again");
}
},
);
assert!(sem.try_acquire(1), "the other holder's permit");
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&raced, 2, cx.waker())
}));
waiter.pending();
drop(waiter);
let before = raced.passes.get();
hammering.set(true);
contend.cancel(&raced, 2);
hammering.set(false);
assert_eq!(
raced.passes.get() - before,
1,
"the drain passed again for releases that had nobody to wake"
);
let mut next = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| next.poll(&raced, 2, cx.waker())));
waiter.pending();
sem.release(1);
assert_eq!(waiter.wakes(), 1, "the release nudges the waiter");
assert!(waiter.poll().is_ready(), "both permits are its");
drop(waiter);
assert_eq!(
next.bypass, 0,
"it retried on a request from before it queued"
);
sem.release(2);
assert_eq!(sem.available(), 2, "a permit was lost or made up");
}
#[test]
fn a_release_during_a_pass_that_found_a_waiter_reaches_it() {
let sem = RawSemaphore::new(1);
let raced = Raced::new(
&sem,
|_| {},
|pass| {
if pass == 1 {
sem.release(1);
}
},
);
assert!(sem.try_acquire(1));
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&raced, 1, cx.waker())
}));
assert!(
waiter.poll().is_ready(),
"a release that found the drain role taken was lost"
);
assert!(!sem.try_acquire(1), "the waiter holds it");
}
struct Watched<'a> {
sem: &'a RawSemaphore,
looks: Cell<u32>,
at_look: Box<dyn Fn(u32) + 'a>,
}
impl<'a> Watched<'a> {
fn new(sem: &'a RawSemaphore, at_look: impl Fn(u32) + 'a) -> Self {
Self {
sem,
looks: Cell::new(0),
at_look: Box::new(at_look),
}
}
}
impl Policy for Watched<'_> {
fn queue(&self) -> &WaitQueue {
self.sem.queue()
}
fn pass(&self, arrivals: Arrivals, woken: &mut WakeList<'_>, wake_front: bool) -> usize {
self.sem.pass(arrivals, woken, wake_front)
}
}
impl Lock for Watched<'_> {
fn admit(&self, state: usize, need: usize) -> Option<usize> {
let admitted = self.sem.admit(state, need);
let look = self.looks.get() + 1;
self.looks.set(look);
(self.at_look)(look);
admitted
}
fn give_back(&self, need: usize) {
self.sem.give_back(need);
}
}
#[test]
fn a_contender_takes_a_lock_let_go_during_its_spin_without_queueing() {
let sem = RawSemaphore::new(1);
assert!(sem.try_acquire(1), "the holder takes it");
let budget = sem.queue().spin.rounds();
assert!(budget > 0, "a fresh lock spins");
let watched = Watched::new(&sem, |look| {
if look == 2 {
sem.release(1);
}
});
let mut contend = Contend::new();
let mut contender = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&watched, 1, cx.waker())
}));
assert!(contender.poll().is_ready(), "the spin took the freed lock");
assert_eq!(contender.wakes(), 0, "a spinner needs no wake");
drop(contender);
assert!(!contend.is_queued(), "the contender never queued");
assert!(
watched.looks.get() >= 3,
"it looked again after the release"
);
assert_eq!(
sem.queue().spin.rounds(),
(2 * budget).min(u32::from(super::super::spin::MAX)),
"a won spin doubles the budget"
);
assert!(!sem.try_acquire(1), "the contender holds it");
}
#[test]
fn a_spin_that_never_sees_room_halves_the_budget_and_queues() {
let sem = RawSemaphore::new(1);
assert!(sem.try_acquire(1));
let budget = sem.queue().spin.rounds();
let watched = Watched::new(&sem, |_| {});
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&watched, 1, cx.waker())
}));
waiter.pending();
assert!(watched.looks.get() > 1, "it spun before queueing");
assert_eq!(
sem.queue().spin.rounds(),
budget / 2,
"a spin that ran out halves"
);
sem.release(1);
assert_eq!(waiter.wakes(), 1, "the release reaches the queued waiter");
assert!(waiter.poll().is_ready());
}
#[test]
fn a_spin_that_saw_room_but_lost_keeps_the_budget() {
let sem = RawSemaphore::new(1);
assert!(sem.try_acquire(1));
let budget = sem.queue().spin.rounds();
let watched = Watched::new(&sem, |look| match look {
2 => sem.release(1),
3 => assert!(sem.try_acquire(1), "the barger takes it"),
_ => {}
});
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&watched, 1, cx.waker())
}));
waiter.pending();
assert_eq!(
sem.queue().spin.rounds(),
budget,
"a busy lock keeps its budget"
);
drop(waiter);
contend.cancel(&watched, 1);
sem.release(1);
}
#[test]
fn a_contender_queues_at_once_behind_a_settled_queue() {
let sem = RawSemaphore::new(1);
assert!(sem.try_acquire(1));
let budget = sem.queue().spin.rounds();
let plain = Watched::new(&sem, |_| {});
let mut first = Contend::new();
let mut first_waiter = Polled::new(core::future::poll_fn(|cx| {
first.poll(&plain, 1, cx.waker())
}));
first_waiter.pending();
let after_first = sem.queue().spin.rounds();
let watched = Watched::new(&sem, |_| {});
let mut second = Contend::new();
let mut second_waiter = Polled::new(core::future::poll_fn(|cx| {
second.poll(&watched, 1, cx.waker())
}));
second_waiter.pending();
assert_eq!(watched.looks.get(), 1, "it looked once and did not spin");
assert_eq!(
sem.queue().spin.rounds(),
after_first,
"the budget is untouched"
);
assert!(after_first <= budget);
drop((first_waiter, second_waiter));
first.cancel(&plain, 1);
second.cancel(&watched, 1);
sem.release(1);
assert!(sem.try_acquire(1), "nothing is left held or owed");
}
#[test]
fn the_spin_takes_only_what_a_fresh_caller_may() {
use super::super::raw_semaphore::SHIFT;
const DRAINING_BIT: usize = 1;
const QUEUED_BIT: usize = 1 << 2;
const CLOSED_BIT: usize = 1 << super::super::queue::FIRST_BIT;
for permits in 0..=2usize {
for flags in 0..(1usize << SHIFT) {
let state = (permits << SHIFT) | flags;
let sem = RawSemaphore::new(0);
sem.queue().state.store(state, Ordering::Relaxed);
let took = spin_take(&sem, 1, state);
let settled = flags & (QUEUED_BIT | DRAINING_BIT) == QUEUED_BIT;
let may = permits >= 1 && flags & (HANDOFF | CLOSED_BIT) == 0 && !settled;
assert_eq!(took, may, "state {state:#x}: the spin took the wrong thing");
let after = sem.queue().state.load(Ordering::Relaxed);
let expected = if took { state - (1 << SHIFT) } else { state };
assert_eq!(
after, expected,
"state {state:#x}: the spin left the wrong word"
);
}
}
}
#[test]
fn a_won_retry_revives_a_spent_budget() {
let sem = RawSemaphore::new(1);
let spin = &sem.queue().spin;
while spin.rounds() > 0 {
spin.ran_out(spin.rounds());
}
assert!(sem.try_acquire(1));
let mut contend = Contend::new();
let mut waiter = Polled::new(core::future::poll_fn(|cx| {
contend.poll(&sem, 1, cx.waker())
}));
waiter.pending();
assert_eq!(spin.rounds(), 0, "a spent budget queues at once");
sem.release(1);
assert!(waiter.poll().is_ready(), "the retry wins");
assert_eq!(spin.rounds(), 1, "the won retry revives the budget");
}
}