use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc;
use std::task::{Poll, Wake, Waker};
use std::time::{Duration, Instant};
use super::{After, Hooks, OwnerPoll, Side};
#[derive(Default)]
struct State {
steps: u64,
seen: u8,
requests: u64,
}
struct Probe {
who: usize,
runs: usize,
}
impl Hooks<State> for Probe {
fn serve(&self, _: &mut State, _: &mut After) {}
fn deferred(&self, st: &mut State, bits: u8, _: &mut After) {
st.seen |= bits;
}
fn requests(&self, st: &mut State, _: &mut After) {
st.requests += 1;
}
fn owner(&self) -> usize {
self.who
}
fn runs(&self) -> usize {
self.runs
}
}
const OWNER: Probe = Probe { who: 2, runs: 2 };
const FOREIGN: Probe = Probe { who: 0, runs: 0 };
fn biased() -> Option<Side<State>> {
crate::rt::handoff::fence::available().then(|| Side::biased(State::default()))
}
struct Counted(AtomicUsize);
impl Wake for Counted {
fn wake(self: Arc<Self>) {
self.0.fetch_add(1, Ordering::SeqCst);
}
}
fn counted() -> (Arc<Counted>, Waker) {
let count = Arc::new(Counted(AtomicUsize::new(0)));
let waker = Waker::from(Arc::clone(&count));
(count, waker)
}
fn step(st: &mut State, _: &mut After) -> Poll<()> {
st.steps += 1;
Poll::Ready(())
}
fn poll(side: &Side<State>, waker: &Waker, hooks: &impl Hooks<State>) -> Poll<()> {
let mut after = After::default();
let polled = side.poll(&mut after, waker, hooks, step);
after.run();
polled
}
fn defer(side: &Side<State>, bits: u8, hooks: &impl Hooks<State>) {
let mut after = After::default();
side.defer(bits, &mut after, hooks);
after.run();
}
fn request(side: &Side<State>, hooks: &impl Hooks<State>) {
let mut after = After::default();
if let Some(held) = side.take_for_request(hooks) {
held.release(&mut after, hooks);
}
after.run();
}
fn steps(side: &mut Side<State>) -> u64 {
side.get_mut().steps
}
#[test]
fn the_owner_holds_a_biased_side_without_touching_the_word() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
let mut after = After::default();
let polled = side.poll(&mut after, &waker, &OWNER, |st, _| {
assert_eq!(side.words().0, 0, "the word was taken");
st.steps += 1;
Poll::Ready(())
});
after.run();
assert!(polled.is_ready());
assert!(!side.is_shared());
assert_eq!(side.words(), (0, 0));
defer(&side, 4, &OWNER);
assert_eq!(side.words(), (0, 0), "a defer of the owner left a flag");
assert_eq!(side.get_mut().seen, 4, "the owner's defer was not done");
assert!(!side.is_shared());
}
#[test]
fn a_foreign_take_of_a_free_side_switches_it_for_good() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
assert!(poll(&side, &waker, &FOREIGN).is_ready());
assert!(side.is_shared());
let mut after = After::default();
let polled = side.poll(&mut after, &waker, &OWNER, |st, _| {
assert_eq!(side.words().0, 1, "shared mode did not use the word");
st.steps += 1;
Poll::Ready(())
});
after.run();
assert!(polled.is_ready());
assert!(side.is_shared(), "shared mode went back");
assert_eq!(steps(&mut side), 3);
}
fn hold_until(
side: &Arc<Side<State>>,
) -> (
mpsc::Receiver<()>,
mpsc::Sender<()>,
std::thread::JoinHandle<()>,
) {
let (held_tx, held_rx) = mpsc::channel();
let (go_tx, go_rx) = mpsc::channel::<()>();
let side = Arc::clone(side);
let thread = std::thread::spawn(move || {
let (_, waker) = counted();
let mut after = After::default();
let polled = side.poll(&mut after, &waker, &OWNER, |st, _| {
st.steps += 1;
held_tx.send(()).unwrap();
go_rx.recv_timeout(Duration::from_secs(10)).unwrap();
Poll::Ready(())
});
after.run();
assert!(polled.is_ready());
});
(held_rx, go_tx, thread)
}
#[test]
fn a_foreign_poll_while_the_owner_holds_is_woken_at_its_release() {
let Some(side) = biased() else { return };
let side = Arc::new(side);
let (held, go, owner) = hold_until(&side);
held.recv_timeout(Duration::from_secs(10)).unwrap();
let (woke, waker) = counted();
assert!(poll(&side, &waker, &FOREIGN).is_pending());
assert_eq!(woke.0.load(Ordering::SeqCst), 0);
go.send(()).unwrap();
owner.join().unwrap();
assert_eq!(
woke.0.load(Ordering::SeqCst),
1,
"the waiting poll was not woken"
);
assert!(poll(&side, &waker, &FOREIGN).is_ready());
}
#[test]
fn work_a_foreign_thread_left_the_owner_is_done_at_its_release() {
let Some(side) = biased() else { return };
let side = Arc::new(side);
let (held, go, owner) = hold_until(&side);
held.recv_timeout(Duration::from_secs(10)).unwrap();
defer(&side, 4, &FOREIGN);
request(&side, &FOREIGN);
go.send(()).unwrap();
owner.join().unwrap();
let mut side = Arc::into_inner(side).unwrap();
assert_eq!(side.get_mut().seen, 4, "the deferred work was lost");
assert_eq!(side.get_mut().requests, 1, "the request was lost");
assert_eq!(side.words(), (0, 0));
}
#[test]
fn the_owner_calling_again_under_its_own_hold_leaves_its_work_for_its_release() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
let mut after = After::default();
let polled = side.poll(&mut after, &waker, &OWNER, |st, after| {
defer(&side, 2, &OWNER);
assert_eq!(st.seen, 0, "the nested call held the side too");
st.steps += 1;
let _ = after;
Poll::Ready(())
});
after.run();
assert!(polled.is_ready());
assert_eq!(side.get_mut().seen, 2, "the nested call's work was lost");
assert_eq!(side.words(), (0, 0));
}
struct Parker {
thread: std::thread::Thread,
woken: AtomicBool,
}
impl Wake for Parker {
fn wake(self: Arc<Self>) {
self.woken.store(true, Ordering::SeqCst);
self.thread.unpark();
}
}
thread_local! {
static PARKER: Arc<Parker> = Arc::new(Parker {
thread: std::thread::current(),
woken: AtomicBool::new(false),
});
}
fn poll_until_done(side: &Side<State>, hooks: &impl Hooks<State>) {
let parker = PARKER.with(Arc::clone);
let waker = Waker::from(Arc::clone(&parker));
let deadline = Instant::now() + Duration::from_secs(20);
while poll(side, &waker, hooks).is_pending() {
while !parker.woken.swap(false, Ordering::SeqCst) {
assert!(
Instant::now() < deadline,
"a wake was lost: word {:?}, shared {}, own {}, intent {}",
side.words(),
side.is_shared(),
side.own.load(Ordering::SeqCst),
side.intent.load(Ordering::SeqCst)
);
std::thread::park_timeout(Duration::from_millis(20));
}
}
}
#[test]
fn an_owner_and_two_foreign_threads_lose_no_step_wake_or_work() {
let Some(side) = biased() else { return };
let side = Arc::new(side);
const ROUNDS: u64 = 20_000;
let foreign = |bit: u8| {
let side = Arc::clone(&side);
std::thread::spawn(move || {
for round in 0..ROUNDS {
match round % 3 {
0 => poll_until_done(&side, &FOREIGN),
1 => {
defer(&side, bit, &FOREIGN);
poll_until_done(&side, &FOREIGN);
}
_ => {
request(&side, &FOREIGN);
poll_until_done(&side, &FOREIGN);
}
}
}
})
};
let a = foreign(1);
let b = foreign(2);
for round in 0..ROUNDS {
if round % 2 == 0 {
poll_until_done(&side, &OWNER);
} else {
defer(&side, 4, &OWNER);
poll_until_done(&side, &OWNER);
}
if round % 64 == 0 {
std::thread::yield_now();
}
}
a.join().unwrap();
b.join().unwrap();
let mut side = Arc::into_inner(side).unwrap();
assert_eq!(
steps(&mut side),
3 * ROUNDS,
"a step ran twice or not at all"
);
assert_eq!(side.get_mut().seen, 7, "deferred work was lost");
assert!(side.get_mut().requests > 0, "requests were lost");
assert_eq!(side.words(), (0, 0), "a flag was left up");
}
#[test]
fn a_foreign_thread_arriving_as_the_owner_lets_go_loses_nothing() {
if biased().is_none() {
return;
}
const SIDES: usize = 3_000;
const OWNER_OPS: u64 = 6;
let sides: Arc<Vec<Side<State>>> =
Arc::new((0..SIDES).map(|_| Side::biased(State::default())).collect());
let gate = Arc::new(std::sync::Barrier::new(3));
let foreign = |bit: u8| {
let (sides, gate) = (Arc::clone(&sides), Arc::clone(&gate));
std::thread::spawn(move || {
for side in sides.iter() {
gate.wait();
match bit {
1 => poll_until_done(side, &FOREIGN),
_ => {
defer(side, bit, &FOREIGN);
request(side, &FOREIGN);
poll_until_done(side, &FOREIGN);
}
}
}
})
};
let (a, b) = (foreign(1), foreign(2));
for side in sides.iter() {
gate.wait();
for _ in 0..OWNER_OPS {
poll_until_done(side, &OWNER);
defer(side, 4, &OWNER);
}
}
a.join().unwrap();
b.join().unwrap();
let mut sides = Arc::into_inner(sides).unwrap();
for (i, side) in sides.iter_mut().enumerate() {
assert_eq!(side.words(), (0, 0), "side {i} kept a flag");
let st = side.get_mut();
assert_eq!(st.steps, OWNER_OPS + 2, "side {i} lost or repeated a step");
assert_eq!(st.seen, 6, "side {i} lost deferred work");
}
assert!(
super::WITNESS[2].load(Ordering::Relaxed) > 0,
"the owner never met a foreign thread's intent"
);
}
struct Fenceless<'a>(&'a AtomicUsize);
impl Hooks<State> for Fenceless<'_> {
fn serve(&self, _: &mut State, _: &mut After) {}
fn nudge(&self) {
self.0.fetch_add(1, Ordering::SeqCst);
}
}
#[test]
fn a_foreign_poll_without_a_fence_is_served_by_the_owner_it_nudges() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
crate::rt::handoff::fence::inject_failure(true);
let nudges = AtomicUsize::new(0);
let (woke, fwaker) = counted();
assert!(poll(&side, &fwaker, &Fenceless(&nudges)).is_pending());
assert_eq!(nudges.load(Ordering::SeqCst), 1, "the owner was not asked");
assert!(!side.is_shared(), "the side switched without a fence");
assert_eq!(woke.0.load(Ordering::SeqCst), 0);
defer(&side, 0, &OWNER);
assert_eq!(woke.0.load(Ordering::SeqCst), 1, "the poll was not woken");
crate::rt::handoff::fence::inject_failure(false);
assert!(poll(&side, &fwaker, &FOREIGN).is_ready());
assert_eq!(steps(&mut side), 2);
}
#[test]
fn a_foreign_call_without_a_fence_on_an_unclaimed_side_goes_shared() {
let Some(side) = biased() else { return };
crate::rt::handoff::fence::inject_failure(true);
let nudges = AtomicUsize::new(0);
let (_, waker) = counted();
assert!(poll(&side, &waker, &Fenceless(&nudges)).is_ready());
assert!(side.is_shared());
assert_eq!(nudges.load(Ordering::SeqCst), 0);
crate::rt::handoff::fence::inject_failure(false);
}
struct FencelessOn<'a>(&'a AtomicUsize, usize);
impl Hooks<State> for FencelessOn<'_> {
fn serve(&self, _: &mut State, _: &mut After) {}
fn runs(&self) -> usize {
self.1
}
fn nudge(&self) {
self.0.fetch_add(1, Ordering::SeqCst);
}
}
#[test]
fn a_fenceless_call_on_a_side_whose_claimant_gave_it_up_takes_it_itself() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
let mut after = After::default();
side.give_up(OWNER.who, &mut after, &OWNER);
after.run();
assert!(
!side.is_shared(),
"a claim given up with no work left switched the side"
);
crate::rt::handoff::fence::inject_failure(true);
let nudges = AtomicUsize::new(0);
assert!(poll(&side, &waker, &Fenceless(&nudges)).is_ready());
crate::rt::handoff::fence::inject_failure(false);
assert_eq!(
nudges.load(Ordering::SeqCst),
0,
"a claimant that left was asked"
);
assert!(side.is_shared());
assert_eq!(steps(&mut side), 2);
}
#[test]
fn a_claimant_giving_its_claim_up_serves_what_a_fenceless_call_left() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
crate::rt::handoff::fence::inject_failure(true);
let nudges = AtomicUsize::new(0);
let (woke, fwaker) = counted();
assert!(poll(&side, &fwaker, &Fenceless(&nudges)).is_pending());
assert_eq!(nudges.load(Ordering::SeqCst), 1);
crate::rt::handoff::fence::inject_failure(false);
let mut after = After::default();
side.give_up(OWNER.who, &mut after, &OWNER);
after.run();
assert_eq!(
woke.0.load(Ordering::SeqCst),
1,
"the poll left was not woken"
);
assert!(side.is_shared());
assert!(poll(&side, &fwaker, &FOREIGN).is_ready());
assert_eq!(steps(&mut side), 2);
}
#[test]
fn the_claimants_own_fenceless_call_on_a_side_homed_elsewhere_takes_it_itself() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
crate::rt::handoff::fence::inject_failure(true);
let nudges = AtomicUsize::new(0);
assert!(poll(&side, &waker, &FencelessOn(&nudges, OWNER.who)).is_ready());
crate::rt::handoff::fence::inject_failure(false);
assert_eq!(
nudges.load(Ordering::SeqCst),
0,
"the claimant nudged itself"
);
assert!(side.is_shared());
assert_eq!(steps(&mut side), 2);
}
#[test]
fn a_claimant_that_gave_its_claim_up_claims_the_side_again() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
let mut after = After::default();
side.give_up(OWNER.who, &mut after, &OWNER);
after.run();
assert!(!side.claimed_by(OWNER.who));
assert!(poll(&side, &waker, &OWNER).is_ready());
assert!(
side.claimed_by(OWNER.who),
"the claimant did not claim the side again"
);
assert!(!side.is_shared());
assert_eq!(side.words(), (0, 0), "the take went through the word");
assert_eq!(steps(&mut side), 2);
}
#[test]
fn sides_made_after_a_fence_failure_start_shared() {
let Some(side) = biased() else { return };
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
assert!(!Side::biased(State::default()).is_shared());
crate::rt::handoff::fence::inject_failure(true);
let nudges = AtomicUsize::new(0);
assert!(poll(&side, &waker, &Fenceless(&nudges)).is_pending());
assert!(Side::biased(State::default()).is_shared());
assert!(!crate::rt::handoff::fence::available());
crate::rt::handoff::fence::inject_failure(false);
assert!(crate::rt::handoff::fence::available());
}
#[test]
fn foreign_polls_without_a_fence_never_share_the_side_and_are_all_served() {
if biased().is_none() {
return;
}
const ROUNDS: u64 = 2_000;
let side = Arc::new(Side::biased(State::default()));
let inside = Arc::new(AtomicUsize::new(0));
let done = Arc::new(AtomicBool::new(false));
let nudges = Arc::new(AtomicUsize::new(0));
let cs = |inside: &AtomicUsize| {
assert_eq!(inside.fetch_add(1, Ordering::SeqCst), 0, "two holders");
std::hint::spin_loop();
inside.fetch_sub(1, Ordering::SeqCst);
};
let foreign = {
let (side, inside, done, nudges) = (
Arc::clone(&side),
Arc::clone(&inside),
Arc::clone(&done),
Arc::clone(&nudges),
);
std::thread::spawn(move || {
crate::rt::handoff::fence::inject_failure(true);
let hooks = Fenceless(&nudges);
let parker = PARKER.with(Arc::clone);
let waker = Waker::from(Arc::clone(&parker));
let deadline = Instant::now() + Duration::from_secs(20);
for _ in 0..ROUNDS {
let mut after = After::default();
while side
.poll(&mut after, &waker, &hooks, |st, _| {
cs(&inside);
st.steps += 1;
Poll::Ready(())
})
.is_pending()
{
after.run();
after = After::default();
while !parker.woken.swap(false, Ordering::SeqCst) {
assert!(Instant::now() < deadline, "a wake was lost");
std::thread::park_timeout(Duration::from_millis(20));
}
}
after.run();
}
done.store(true, Ordering::SeqCst);
})
};
let (_, waker) = counted();
let mut handled = 0;
let mut owner_steps = 0;
while !done.load(Ordering::SeqCst) {
let mut after = After::default();
let polled = side.poll(&mut after, &waker, &OWNER, |st, _| {
cs(&inside);
st.steps += 1;
Poll::Ready(())
});
after.run();
owner_steps += u64::from(polled.is_ready());
let asked = nudges.load(Ordering::SeqCst);
if asked > handled {
handled = asked;
defer(&side, 0, &OWNER);
}
}
foreign.join().unwrap();
let mut side = Arc::into_inner(side).unwrap();
assert_eq!(steps(&mut side), ROUNDS + owner_steps, "a step was lost");
assert_eq!(side.words(), (0, 0), "a flag was left up");
}
#[test]
fn the_owners_poll_runs_its_step_once_without_touching_the_word() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
assert!(
!side.owned_by(OWNER.who),
"a side nobody claimed has no owner"
);
assert!(poll(&side, &waker, &OWNER).is_ready());
assert!(side.owned_by(OWNER.who) && !side.owned_by(OWNER.who + 2));
let mut after = After::default();
let done = matches!(
side.poll_owner(&mut after, &OWNER, |st, _| {
assert_eq!(side.words().0, 0, "the word was taken");
assert!(
!side.owned_by(OWNER.who),
"the owner's own hold is not another owner's take"
);
st.steps += 1;
Poll::Ready(())
}),
OwnerPoll::Done(Poll::Ready(()))
);
after.run();
assert!(done);
assert_eq!(steps(&mut side), 2);
assert_eq!(side.words(), (0, 0));
assert!(!side.is_shared());
}
#[test]
fn the_owners_poll_is_refused_once_a_foreign_thread_came() {
let Some(side) = biased() else { return };
let side = Arc::new(side);
let (held, go, owner) = hold_until(&side);
held.recv_timeout(Duration::from_secs(10)).unwrap();
defer(&side, 4, &FOREIGN);
go.send(()).unwrap();
owner.join().unwrap();
let mut side = Arc::into_inner(side).unwrap();
assert!(
side.owned_by(OWNER.who),
"the foreign thread switched the side"
);
let mut after = After::default();
let refused = matches!(
side.poll_owner(&mut after, &OWNER, step),
OwnerPoll::Refused
);
after.run();
assert!(refused);
assert_eq!(steps(&mut side), 1, "a refused poll ran its step");
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
assert_eq!(steps(&mut side), 2);
assert_eq!(side.get_mut().seen, 4, "the foreign thread's work was lost");
assert_eq!(side.words(), (0, 0));
}
#[test]
fn work_left_during_the_owners_poll_is_finished_with_the_step_again() {
let Some(mut side) = biased() else { return };
let (_, waker) = counted();
assert!(poll(&side, &waker, &OWNER).is_ready());
let mut step = |st: &mut State, after: &mut After| {
st.steps += 1;
if st.steps > 2 {
return Poll::Ready(());
}
defer(&side, 4, &FOREIGN);
#[cfg(readiness)]
side.drive(after, &FOREIGN, 0);
let _ = after;
Poll::Pending
};
let mut after = After::default();
let polled = match side.poll_owner(&mut after, &OWNER, |st, after| step(st, after)) {
OwnerPoll::Left(held, out) => held.finish_poll(out, &mut after, &OWNER, &mut step),
OwnerPoll::Done(_) => panic!("the foreign thread's work was not seen"),
OwnerPoll::Refused => panic!("the owner's poll was refused"),
};
after.run();
assert_eq!(side.get_mut().seen, 4, "the work left was lost");
#[cfg(readiness)]
{
assert!(
polled.is_ready(),
"the step did not run again after the edge"
);
assert_eq!(steps(&mut side), 3);
}
#[cfg(not(readiness))]
{
assert!(polled.is_pending());
assert_eq!(steps(&mut side), 2);
}
assert_eq!(side.words(), (0, 0), "a flag was left up");
}