use std::collections::BTreeSet;
use super::explore::FailedState;
use super::observer::{Observer, RaceOutcome, Step, StepCx};
use crate::model::{State, StateView, Transition};
#[derive(Default)]
pub(super) struct Wut {
children: Vec<(Transition, Wut)>,
}
impl Wut {
pub(super) fn children(&self) -> &[(Transition, Wut)] {
&self.children
}
fn graft(&mut self, seq: &[Transition]) {
let Some((&head, rest)) = seq.split_first() else {
return;
};
debug_assert!(
self.children.iter().all(|(t, _)| t.pid != head.pid),
"wakeup-tree siblings must have distinct pids"
);
let mut branch = Wut::default();
branch.graft(rest);
self.children.push((head, branch));
}
}
pub(super) struct Frame {
sleep: Vec<usize>,
pending: Vec<Transition>,
}
impl Frame {
pub(super) fn sleep(&self) -> &[usize] {
&self.sleep
}
pub(super) fn pending(&self) -> &[Transition] {
&self.pending
}
}
pub(super) fn run<'a>(
root: State<'a>,
observer: &mut impl Observer,
) -> Result<(), FailedState<'a>> {
let mut tree = Wut::default();
let mut frames: Vec<Frame> = vec![Frame {
sleep: Vec::new(),
pending: root.enabled().to_vec(),
}];
observer.step(Step::Visit, StepCx::new(&tree, &frames, &[], &root));
if root.is_failed() {
let (reason, view) = root.into_failure();
return Err(FailedState::new(reason, view));
}
let view = root.view();
if let Some(p) = seed(&root, &[]) {
let seeded = resolve(&root, p).expect("a seeded process must be runnable");
tree.graft(&[seeded]);
observer.step(
Step::RootSeed { seeded },
StepCx::new(&tree, &frames, &[], &root),
);
} else {
observer.step(Step::RootEmpty, StepCx::new(&tree, &frames, &[], &root));
observer.step(Step::Done, StepCx::new(&tree, &frames, &[], &root));
}
let mut cur = root;
let mut prefix: Vec<Transition> = Vec::new();
let mut need_replay = false;
loop {
if node_at(&tree, prefix.len()).children.is_empty() {
if pop_exhausted(&mut tree, &mut frames, &mut prefix, &cur, observer) {
observer.step(Step::Done, StepCx::new(&tree, &frames, &prefix, &cur));
return Ok(());
}
need_replay = true;
continue;
}
if need_replay {
cur = view.replay(&prefix);
need_replay = false;
observer.step(
Step::Replay { prefix: &prefix },
StepCx::new(&tree, &frames, &prefix, &cur),
);
}
let p = node_at(&tree, prefix.len()).children[0].0.pid;
let p_t = resolve(&cur, p).expect("a wakeup-tree branch must be runnable");
debug_assert!(
cur.enabled().contains(&p_t),
"a committed op must be enabled"
);
let child_sleep = child_sleep_set(&cur, frames.last().unwrap(), p_t);
cur.apply(p_t);
observer.step(Step::Visit, StepCx::new(&tree, &frames, &prefix, &cur));
if cur.is_failed() {
observer.step(
Step::Maximal {
trace: cur.trace(),
failure: true,
},
StepCx::new(&tree, &frames, &prefix, &cur),
);
let (reason, view) = cur.into_failure();
return Err(FailedState::new(reason, view));
}
observer.step(
Step::Descend {
depth: prefix.len(),
committed: p_t,
parent_sleep: frames.last().unwrap().sleep(),
child_sleep: &child_sleep,
},
StepCx::new(&tree, &frames, &prefix, &cur),
);
frames.push(Frame {
sleep: child_sleep,
pending: cur.enabled().to_vec(),
});
prefix.push(p_t);
if cur.enabled().is_empty() {
observer.step(
Step::Maximal {
trace: &prefix,
failure: false,
},
StepCx::new(&tree, &frames, &prefix, &cur),
);
plan_reversals(&mut tree, &frames, &prefix, &cur, &view, observer);
debug_assert!(
node_at(&tree, prefix.len()).children.is_empty(),
"a maximal trace's wakeup-tree node has no continuations"
);
need_replay = true;
} else {
seed_child(&mut tree, &cur, &frames, &prefix, observer);
}
}
}
fn pop_exhausted(
tree: &mut Wut,
frames: &mut Vec<Frame>,
prefix: &mut Vec<Transition>,
cur: &State,
observer: &mut impl Observer,
) -> bool {
let from_depth = prefix.len();
frames.pop();
let Some(finished_t) = prefix.pop() else {
return true;
};
let finished = finished_t.pid;
let parent = node_at_mut(tree, prefix.len());
debug_assert_eq!(
parent.children.first().map(|(t, _)| t.pid),
Some(finished),
"the finished branch must be the ≺-minimal child"
);
parent.children.remove(0);
frames.last_mut().unwrap().sleep.push(finished);
observer.step(
Step::Pop {
finished_pid: finished,
from_depth,
into_depth: from_depth - 1,
},
StepCx::new(tree, frames, prefix, cur),
);
false
}
fn child_sleep_set(cur: &State, parent: &Frame, p_t: Transition) -> Vec<usize> {
parent
.sleep
.iter()
.copied()
.filter(|&q| match resolve(cur, q) {
Some(q_t) => !cur.depends(p_t, q_t),
None => false,
})
.collect()
}
fn seed_child(
tree: &mut Wut,
cur: &State,
frames: &[Frame],
prefix: &[Transition],
observer: &mut impl Observer,
) {
let depth = prefix.len();
let child_sleep = &frames.last().unwrap().sleep;
if node_at(tree, depth).children.is_empty()
&& let Some(q) = seed(cur, child_sleep)
{
debug_assert!(
!child_sleep.contains(&q),
"sleep-set-blocked state under Optimal DPOR"
);
let q_t = resolve(cur, q).expect("a seeded process must be runnable");
node_at_mut(tree, depth).graft(&[q_t]);
}
let seeded = node_at(tree, depth).children.first().map(|(t, _)| *t);
observer.step(
Step::SeedChild { depth, seeded },
StepCx::new(tree, frames, prefix, cur),
);
}
fn plan_reversals(
tree: &mut Wut,
frames: &[Frame],
trace: &[Transition],
state: &State,
view: &StateView,
observer: &mut impl Observer,
) {
let n = trace.len();
let clocks = event_clocks(state, trace);
let non_blocking = state.non_blocking();
for j in 1..=n {
for i in 1..j {
if !reversible_race(state, &clocks, trace, i, j) {
continue;
}
let mut v = notdep(&clocks, trace, i);
v.push(trace[j - 1]);
let prefix = &trace[..i - 1];
let outcome = if !non_blocking && !runnable_after(view, trace, &v, i) {
RaceOutcome::Disabling
} else if let Some(covering_pid) =
covered_by_sleeper(view, prefix, &frames[i - 1].sleep, &v)
{
RaceOutcome::CoveredBySleeper {
insert_depth: i - 1,
covering_pid,
}
} else {
match insert(node_at_mut(tree, i - 1), view, prefix, &v) {
InsertResult::ExistingLeaf => RaceOutcome::ExistingLeaf {
insert_depth: i - 1,
},
InsertResult::Grafted => RaceOutcome::Grafted {
insert_depth: i - 1,
},
}
};
observer.step(
Step::Race {
i,
j,
e: trace[i - 1],
ep: trace[j - 1],
notdep: &v[..v.len() - 1],
v: &v,
outcome,
},
StepCx::new(tree, frames, trace, state),
);
}
}
}
fn covered_by_sleeper(
view: &StateView,
prefix: &[Transition],
sleep: &[usize],
v: &[Transition],
) -> Option<usize> {
if sleep.is_empty() {
return None;
}
let mut state = view.replay(prefix);
weak_initial_walk(&mut state, sleep, v).map(|m| sleep[m.child])
}
enum InsertResult {
Grafted,
ExistingLeaf,
}
fn insert(
node: &mut Wut,
view: &StateView,
prefix: &[Transition],
v: &[Transition],
) -> InsertResult {
let state = view.replay(prefix);
insert_rec(node, view, prefix.to_vec(), state, v)
}
fn insert_rec(
node: &mut Wut,
view: &StateView,
prefix: Vec<Transition>,
mut state: State,
v: &[Transition],
) -> InsertResult {
let pids: Vec<usize> = node.children.iter().map(|(t, _)| t.pid).collect();
let Some(m) = weak_initial_walk(&mut state, &pids, v) else {
node.graft(v); return InsertResult::Grafted;
};
if node.children[m.child].1.children.is_empty() {
return InsertResult::ExistingLeaf; }
let mut child_prefix = prefix;
child_prefix.push(m.q_t);
let subtree = &mut node.children[m.child].1;
if m.clean {
state.apply(m.q_t);
insert_rec(subtree, view, child_prefix, state, &m.rest)
} else {
let child_state = view.replay(&child_prefix);
insert_rec(subtree, view, child_prefix, child_state, &m.rest)
}
}
struct WeakInitial {
child: usize,
rest: Vec<Transition>,
q_t: Transition,
clean: bool,
}
fn weak_initial_walk(
state: &mut State,
candidates: &[usize],
v: &[Transition],
) -> Option<WeakInitial> {
if candidates.is_empty() {
return None;
}
let q_ts: Vec<Option<Transition>> = candidates.iter().map(|&q| resolve(state, q)).collect();
let mut matched: Vec<Option<usize>> = vec![None; candidates.len()];
let mut killed: Vec<bool> = q_ts.iter().map(Option::is_none).collect();
let mut applies = 0usize;
let win = |matched: &[Option<usize>], killed: &[bool], applies: usize| -> Option<WeakInitial> {
let child = (0..candidates.len()).find(|&ci| !killed[ci])?;
let rest = match matched[child] {
Some(k) => {
let mut r = v.to_vec();
r.remove(k);
r
}
None => v.to_vec(),
};
Some(WeakInitial {
child,
rest,
q_t: q_ts[child].unwrap(),
clean: applies == 0,
})
};
for (k, &vk) in v.iter().enumerate() {
for ci in 0..candidates.len() {
if !killed[ci] && matched[ci].is_none() && vk.pid == candidates[ci] {
matched[ci] = Some(k);
}
}
if let Some(ci) = (0..candidates.len()).find(|&ci| !killed[ci])
&& matched[ci].is_some()
{
return win(&matched, &killed, applies);
}
let Some(vk_t) = resolve(state, vk.pid) else {
for ci in 0..candidates.len() {
if matched[ci].is_none() {
killed[ci] = true;
}
}
break;
};
for ci in 0..candidates.len() {
if !killed[ci] && matched[ci].is_none() && state.depends(vk_t, q_ts[ci].unwrap()) {
killed[ci] = true;
}
}
state.apply(vk_t);
applies += 1;
}
win(&matched, &killed, applies)
}
struct Clocks {
flat: Vec<usize>,
procs: usize,
}
impl Clocks {
fn get(&self, k: usize, pid: usize) -> usize {
self.flat[k * self.procs + pid]
}
}
fn notdep(clocks: &Clocks, trace: &[Transition], i: usize) -> Vec<Transition> {
let e = trace[i - 1];
(i + 1..=trace.len())
.filter(|&k| clocks.get(k, e.pid) < i)
.map(|k| trace[k - 1])
.collect()
}
fn reversible_race(
state: &State,
clocks: &Clocks,
trace: &[Transition],
i: usize,
j: usize,
) -> bool {
let (e, ep) = (trace[i - 1], trace[j - 1]);
if e.pid == ep.pid || !state.depends(e, ep) || !happens_before(clocks, trace, i, j) {
return false;
}
!(i + 1..j).any(|m| happens_before(clocks, trace, i, m) && happens_before(clocks, trace, m, j))
}
fn runnable_after(view: &StateView, trace: &[Transition], v: &[Transition], i: usize) -> bool {
let (ep, notdep) = v.split_last().expect("v ends with e'");
let mut state = view.replay(&trace[..i - 1]);
for nd in notdep {
let Some(t) = resolve(&state, nd.pid) else {
return false;
};
state.apply(t);
if state.is_failed() {
return false; }
}
state.enabled().iter().any(|t| t.pid == ep.pid)
}
fn happens_before(clocks: &Clocks, trace: &[Transition], i: usize, j: usize) -> bool {
i <= clocks.get(j, trace[i - 1].pid)
}
fn event_clocks(state: &State, trace: &[Transition]) -> Clocks {
let procs = state.world().processes().len();
let n = trace.len();
let mut flat = vec![0usize; (n + 1) * procs];
let mut last = vec![0usize; procs];
for k in 1..=n {
let t = trace[k - 1];
let src = last[t.pid];
flat.copy_within(src * procs..src * procs + procs, k * procs);
for j in 1..k {
if state.depends(trace[j - 1], t) {
for p in 0..procs {
let pred = flat[j * procs + p];
if pred > flat[k * procs + p] {
flat[k * procs + p] = pred;
}
}
}
}
flat[k * procs + t.pid] = k;
last[t.pid] = k;
}
Clocks { flat, procs }
}
fn seed(state: &State, sleep: &[usize]) -> Option<usize> {
let enabled = enabled_pids(state);
if enabled.is_empty() {
return None;
}
debug_assert!(
enabled.iter().any(|p| !sleep.contains(p)),
"sleep-set-blocked state under Optimal DPOR"
);
enabled
.iter()
.find(|p| !sleep.contains(p))
.or_else(|| enabled.iter().next())
.copied()
}
fn enabled_pids(state: &State) -> BTreeSet<usize> {
state.enabled().iter().map(|t| t.pid).collect()
}
fn resolve(state: &State, p: usize) -> Option<Transition> {
state.enabled().iter().copied().find(|t| t.pid == p)
}
fn node_at(root: &Wut, depth: usize) -> &Wut {
let mut n = root;
for _ in 0..depth {
n = &n.children[0].1;
}
n
}
fn node_at_mut(root: &mut Wut, depth: usize) -> &mut Wut {
let mut n = root;
for _ in 0..depth {
n = &mut n.children[0].1;
}
n
}
#[cfg(test)]
mod tests {
use std::collections::{BTreeMap, BTreeSet};
use std::fmt::Debug;
use super::{Frame, State, StateView, Wut, event_clocks, happens_before};
use crate::Atomic;
use crate::model::World;
use crate::search::{FailedState, Observer, Step, StepCx, explore};
fn dfs<'a>(state: State<'a>, observer: &mut impl Observer) -> Result<(), FailedState<'a>> {
let tree = Wut::default();
let frames: Vec<Frame> = Vec::new();
observer.step(
Step::Visit,
StepCx::new(&tree, &frames, state.trace(), &state),
);
if state.is_failed() {
let (reason, view) = state.into_failure();
return Err(FailedState::new(reason, view));
}
for &t in state.enabled() {
let mut next = state.fork();
next.apply(t);
dfs(next, observer)?;
}
Ok(())
}
fn dfs_explore<'a>(
setup: &'a dyn Fn(&mut World<'a>),
observer: &mut impl Observer,
) -> Result<(), FailedState<'a>> {
dfs(StateView::new(setup).state(), observer)
}
fn spawn_store<'a, T>(world: &mut World<'a>, name: impl Into<String>, cell: Atomic<T>, value: T)
where
T: Copy + PartialEq + Debug + 'static,
{
world.spawn(name, async move {
cell.store(value).await;
Ok(())
});
}
fn spawn_load<'a, T>(world: &mut World<'a>, name: impl Into<String>, cell: Atomic<T>)
where
T: Copy + PartialEq + Debug + 'static,
{
world.spawn(name, async move {
cell.load().await;
Ok(())
});
}
fn spawn_increment<'a>(
world: &mut World<'a>,
name: impl Into<String>,
src: Atomic<i32>,
dst: Atomic<i32>,
) {
world.spawn(name, async move {
let v = src.load().await;
dst.store(v + 1).await;
Ok(())
});
}
fn cells<'a>(world: &mut World<'a>, prefix: &str, count: usize) -> Vec<Atomic<i32>> {
(0..count)
.map(|i| world.atomic(format!("{prefix}{i}"), 0i32))
.collect()
}
#[derive(Default)]
struct Leaves(usize);
impl Observer for Leaves {
fn step(&mut self, step: Step<'_>, cx: StepCx<'_, '_>) {
if matches!(step, Step::Visit) && cx.state().is_terminal() {
self.0 += 1;
}
}
}
fn leaves<'a>(setup: &'a dyn Fn(&mut World<'a>)) -> usize {
let mut obs = Leaves::default();
let _ = explore(setup, &mut obs);
obs.0
}
fn dfs_leaves<'a>(setup: &'a dyn Fn(&mut World<'a>)) -> usize {
let mut obs = Leaves::default();
let _ = dfs_explore(setup, &mut obs);
obs.0
}
type Label = (usize, usize, usize); type Canon = BTreeSet<(Label, Label)>;
fn canon(state: &State) -> Canon {
let trace = state.trace();
let n = trace.len();
let clocks = event_clocks(state, trace);
let mut seen = BTreeMap::<usize, usize>::new();
let mut label = vec![(0usize, 0usize, 0usize); n + 1];
for k in 1..=n {
let t = trace[k - 1];
let idx = seen.entry(t.pid).or_default();
label[k] = (t.pid, *idx, t.oid);
*idx += 1;
}
let mut hb = Canon::new();
for j in 1..=n {
for i in 1..j {
if happens_before(&clocks, trace, i, j) {
hb.insert((label[i], label[j]));
}
}
}
hb
}
#[derive(Default)]
struct Classes(BTreeSet<Canon>);
impl Observer for Classes {
fn step(&mut self, step: Step<'_>, cx: StepCx<'_, '_>) {
if matches!(step, Step::Visit) && cx.state().is_terminal() {
self.0.insert(canon(cx.state()));
}
}
}
fn classes<'a>(setup: &'a dyn Fn(&mut World<'a>)) -> usize {
let mut obs = Classes::default();
let _ = dfs_explore(setup, &mut obs);
obs.0.len()
}
fn assert_optimal<'a>(setup: &'a dyn Fn(&mut World<'a>)) {
let opt = leaves(setup);
assert_eq!(
opt,
classes(setup),
"Optimal must explore one trace per class"
);
assert!(
opt <= dfs_leaves(setup),
"Optimal must never explore more than DFS"
);
}
fn assert_leaves<'a>(setup: &'a dyn Fn(&mut World<'a>), dfs: usize, optimal: usize) {
assert_eq!(dfs_leaves(setup), dfs, "DFS leaf count");
assert_eq!(leaves(setup), optimal, "Optimal leaf count");
assert_optimal(setup);
}
fn two_loaders(world: &mut World) {
let x = world.atomic("x", 0u32);
spawn_load(world, "reader-1", x.clone());
spawn_load(world, "reader-2", x);
}
fn two_objects(world: &mut World) {
let x = world.atomic("x", 0u32);
let y = world.atomic("y", 0u32);
spawn_store(world, "writer-x", x, 1);
spawn_store(world, "writer-y", y, 1);
}
fn two_writers(world: &mut World) {
let x = world.atomic("x", 0u32);
spawn_store(world, "writer-1", x.clone(), 1);
spawn_store(world, "writer-2", x, 2);
}
fn racy(world: &mut World) {
let x = world.atomic("x", 0u32);
spawn_store(world, "writer", x.clone(), 1);
world.spawn("reader", async move {
if x.load().await == 1 {
Ok(())
} else {
Err("unexpected value".into())
}
});
}
fn never_finishes(world: &mut World) {
world.spawn("stuck", async {
std::future::pending::<()>().await;
Ok(())
});
}
fn three_writers(world: &mut World) {
let x = world.atomic("x", 0u32);
for i in 1..=3u32 {
spawn_store(world, format!("w{i}"), x.clone(), i);
}
}
fn one_writer_two_readers(world: &mut World) {
let x = world.atomic("x", 0u32);
spawn_store(world, "writer", x.clone(), 1);
spawn_load(world, "reader-1", x.clone());
spawn_load(world, "reader-2", x);
}
fn branch_changes_ops(world: &mut World) {
let x = world.atomic("x", 0u32);
let y = world.atomic("y", 0u32);
spawn_store(world, "write-x", x.clone(), 1);
spawn_store(world, "write-y", y.clone(), 1);
world.spawn("reader", async move {
if x.load().await == 0 {
y.load().await;
} else {
y.store(2).await;
}
Ok(())
});
}
fn producer_consumer(world: &mut World) {
let (tx, rx) = world.channel::<i32>("ch");
let tx2 = tx.clone();
world.spawn("producer-1", async move {
tx.send(1).await;
Ok(())
});
world.spawn("producer-2", async move {
tx2.send(2).await;
Ok(())
});
world.spawn("consumer", async move {
rx.recv().await;
rx.recv().await;
Ok(())
});
}
fn rpc_mux(world: &mut World) {
#[derive(Debug, Clone, Copy)]
struct Reply {
id: i32,
result: i32,
}
let in_flight = world.atomic("in_flight", -1);
let (conn, reader) = world.channel::<Reply>("conn");
for id in 0..2 {
let (in_flight, conn) = (in_flight.clone(), conn.clone());
world.spawn(format!("caller-{id}"), async move {
in_flight.store(id).await;
conn.send(Reply {
id,
result: id * 10,
})
.await;
Ok(())
});
}
world.spawn("reader", async move {
for _ in 0..2 {
let frame = reader.recv().await;
let routed_to = in_flight.load().await;
if frame.result != routed_to * 10 {
return Err(format!(
"call {routed_to} received call {}'s result ({})",
frame.id, frame.result
)
.into());
}
}
Ok(())
});
}
fn three_producers(world: &mut World) {
let (tx, rx) = world.channel::<i32>("ch");
for i in 1..=3i32 {
let tx = tx.clone();
world.spawn(format!("producer-{i}"), async move {
tx.send(i).await;
Ok(())
});
}
world.spawn("consumer", async move {
rx.recv().await;
rx.recv().await;
rx.recv().await;
Ok(())
});
}
fn interleaved(world: &mut World) {
let (tx, rx) = world.channel::<i32>("ch");
let tx_a = tx.clone();
world.spawn("producer-a", async move {
tx_a.send(1).await;
tx_a.send(2).await;
Ok(())
});
world.spawn("producer-b", async move {
tx.send(3).await;
Ok(())
});
world.spawn("consumer", async move {
rx.recv().await;
rx.recv().await;
rx.recv().await;
Ok(())
});
}
fn mixed(world: &mut World) {
let x = world.atomic("x", 0i32);
spawn_store(world, "writer-1", x.clone(), 1);
spawn_store(world, "writer-2", x, 2);
let (tx, rx) = world.channel::<i32>("ch");
let tx2 = tx.clone();
world.spawn("producer-1", async move {
tx.send(1).await;
Ok(())
});
world.spawn("producer-2", async move {
tx2.send(2).await;
Ok(())
});
world.spawn("consumer", async move {
rx.recv().await;
rx.recv().await;
Ok(())
});
}
fn branch_consumer(world: &mut World) {
let x = world.atomic("x", 0i32);
let writer = x.clone();
spawn_store(world, "writer", writer, 5);
let (tx, rx) = world.channel::<i32>("ch");
let tx2 = tx.clone();
world.spawn("producer-1", async move {
tx.send(0).await; Ok(())
});
world.spawn("producer-2", async move {
tx2.send(1).await;
Ok(())
});
world.spawn("consumer", async move {
if rx.recv().await == 0 {
x.load().await;
} else {
x.store(9).await;
}
Ok(())
});
}
fn seq_drift(world: &mut World) {
let g = world.atomic("g", 0i32);
let (tx, rx) = world.channel::<i32>("ch");
let tx1 = tx.clone();
world.spawn("p1", async move {
tx1.send(1).await;
Ok(())
});
let gw = g.clone();
spawn_store(world, "writer-g", gw, 5);
let tx3 = tx.clone();
world.spawn("p3", async move {
g.load().await;
tx3.send(3).await;
Ok(())
});
world.spawn("p2", async move {
tx.send(2).await;
Ok(())
});
world.spawn("consumer", async move {
rx.recv().await;
rx.recv().await;
rx.recv().await;
Ok(())
});
}
fn readers(world: &mut World, n: usize) {
let x = world.atomic("x", 0u32);
spawn_store(world, "writer", x.clone(), 42);
for i in 1..=n {
spawn_load(world, format!("reader-{i}"), x.clone());
}
}
fn lastzero(world: &mut World, n: usize) {
let cells = cells(world, "a", n + 1);
let rc = cells.to_vec();
world.spawn("reader", async move {
let mut i = n;
loop {
if rc[i].load().await == 0 {
break;
}
if i == 0 {
break; }
i -= 1;
}
Ok(())
});
for j in 1..=n {
spawn_increment(
world,
format!("writer-{j}"),
cells[j - 1].clone(),
cells[j].clone(),
);
}
}
fn indexer(world: &mut World, num_threads: usize) {
let table = cells(world, "t", 128);
for tid in 0..num_threads {
let tab = table.to_vec();
world.spawn(format!("thread-{tid}"), async move {
let mut m = 0i32;
for _ in 0..10 {
if m < 4 {
m += 1;
let w = m * 11 + tid as i32;
let mut h = ((w * 7) % 128) as usize;
loop {
match tab[h].compare_exchange(0, w).await {
Ok(_) => break,
Err(_) => h = (h + 1) % 128,
}
}
} else {
break;
}
}
Ok(())
});
}
}
fn indexer_collision(world: &mut World, n: usize) {
let cells = cells(world, "c", n + 1);
for tid in 0..2 {
let tab = cells.to_vec();
world.spawn(format!("t{tid}"), async move {
let w = tid + 1;
let mut h = 0usize;
loop {
match tab[h].compare_exchange(0, w).await {
Ok(_) => break,
Err(_) => h += 1,
}
}
Ok(())
});
}
}
#[test]
fn reduces_read_read() {
assert_leaves(&two_loaders, 2, 1);
}
#[test]
fn reduces_disjoint_objects() {
assert_leaves(&two_objects, 2, 1);
}
#[test]
fn keeps_dependent_writes() {
assert_leaves(&two_writers, 2, 2);
}
#[test]
fn keeps_three_writers() {
assert_leaves(&three_writers, 6, 6);
}
#[test]
fn reduces_writer_two_readers() {
assert!(leaves(&one_writer_two_readers) < 6);
assert_optimal(&one_writer_two_readers);
}
#[test]
fn handles_branch_changing_ops() {
assert_optimal(&branch_changes_ops);
}
#[test]
fn finds_the_race() {
let dfs = dfs_explore(&racy, &mut ()).unwrap_err();
let opt = explore(&racy, &mut ()).unwrap_err();
assert_eq!(opt.to_string(), dfs.to_string());
}
#[test]
fn detects_deadlock() {
let failed = explore(&never_finishes, &mut ()).unwrap_err();
assert_eq!(failed.to_string(), "deadlock");
}
#[test]
fn channel_one_per_class() {
assert_optimal(&producer_consumer);
}
#[test]
fn rpc_mux_bug_found_identically() {
let dfs = dfs_explore(&rpc_mux, &mut ()).unwrap_err();
let opt = explore(&rpc_mux, &mut ()).unwrap_err();
assert_eq!(opt.to_string(), dfs.to_string());
}
#[test]
fn channel_three_producers() {
assert_leaves(&three_producers, 30, 6);
}
#[test]
fn channel_interleaved_independent_count() {
assert_leaves(&interleaved, 15, 3);
}
#[test]
fn channel_mixed_objects() {
assert_leaves(&mixed, 120, 4);
}
#[test]
fn channel_branch_consumer() {
assert_optimal(&branch_consumer);
}
#[test]
fn channel_seq_drift_no_panic() {
assert_optimal(&seq_drift);
assert_leaves(&seq_drift, 608, 12);
}
#[test]
fn readers_small() {
for n in 2..=4 {
assert_optimal(&|w| readers(w, n));
}
assert_eq!(leaves(&|w| readers(w, 2)), 4);
}
#[test]
fn readers_paper_counts() {
assert_eq!(leaves(&|w| readers(w, 8)), 256);
assert_eq!(leaves(&|w| readers(w, 13)), 8192);
}
#[test]
fn lastzero_small() {
assert_optimal(&|w| lastzero(w, 2));
assert_optimal(&|w| lastzero(w, 3));
}
#[cfg(not(debug_assertions))]
#[test]
fn lastzero_4_ground_truth() {
assert_optimal(&|w| lastzero(w, 4));
}
#[test]
fn lastzero_paper_counts() {
assert_eq!(leaves(&|w| lastzero(w, 5)), 64);
assert_eq!(leaves(&|w| lastzero(w, 10)), 3328);
}
#[cfg(not(debug_assertions))]
#[test]
fn lastzero_15_paper_count() {
assert_eq!(leaves(&|w| lastzero(w, 15)), 147456);
}
#[test]
fn indexer_collision_ground_truth() {
assert_optimal(&|w| indexer_collision(w, 2));
assert_optimal(&|w| indexer_collision(w, 3));
}
#[test]
fn indexer_paper_counts() {
assert_eq!(leaves(&|w| indexer(w, 12)), 8);
}
#[cfg(not(debug_assertions))]
#[test]
fn indexer_15_paper_count() {
assert_eq!(leaves(&|w| indexer(w, 15)), 4096);
}
fn read_modify_reader(world: &mut World) {
let x = world.atomic("x", 0u32);
let w = x.clone();
world.spawn("writer", async move {
let _ = w.load().await;
w.store(1).await;
Ok(())
});
spawn_load(world, "reader", x);
}
fn double_counter(world: &mut World) {
let x = world.atomic("x", 0i32);
for i in 0..2 {
let c = x.clone();
world.spawn(format!("inc-{i}"), async move {
let v = c.load().await;
c.store(v + 1).await;
Ok(())
});
}
}
fn read_modify_two_readers(world: &mut World) {
let x = world.atomic("x", 0u32);
let w = x.clone();
world.spawn("writer", async move {
let _ = w.load().await;
w.store(1).await;
Ok(())
});
spawn_load(world, "reader-1", x.clone());
spawn_load(world, "reader-2", x);
}
fn read_modify_killer(world: &mut World) {
let x = world.atomic("x", 0u32);
let w = x.clone();
world.spawn("writer", async move {
let _ = w.load().await;
w.store(1).await;
Ok(())
});
world.spawn("reader", async move {
if x.load().await == 1 {
Ok(())
} else {
Err("reader saw the pre-store value".into())
}
});
}
#[test]
fn read_modify_reader_keeps_both_classes() {
assert_leaves(&read_modify_reader, 3, 2);
}
#[test]
fn read_modify_two_readers_optimal() {
assert_optimal(&read_modify_two_readers);
}
#[test]
fn double_counter_optimal() {
assert_optimal(&double_counter);
}
#[test]
fn read_modify_finds_stale_read_identically() {
let dfs = dfs_explore(&read_modify_killer, &mut ()).unwrap_err();
let opt = explore(&read_modify_killer, &mut ()).unwrap_err();
assert_eq!(opt.to_string(), dfs.to_string());
}
fn lcg(s: &mut u64) -> u64 {
*s = s
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
*s >> 33
}
fn random_atomics(seed: u64, world: &mut World) {
let mut s = seed.wrapping_add(1);
let cells_n = 1 + (lcg(&mut s) % 2) as usize;
let cells: Vec<Atomic<i32>> = (0..cells_n)
.map(|i| world.atomic(format!("c{i}"), 0))
.collect();
let procs_n = 2 + (lcg(&mut s) % 2) as usize;
for p in 0..procs_n {
let ops_n = 1 + (lcg(&mut s) % 2) as usize;
let ops: Vec<(usize, u64, i32)> = (0..ops_n)
.map(|_| {
(
(lcg(&mut s) as usize) % cells_n,
lcg(&mut s) % 3,
(lcg(&mut s) % 3) as i32,
)
})
.collect();
let cs = cells.clone();
world.spawn(format!("p{p}"), async move {
for (ci, kind, val) in ops {
match kind {
0 => {
cs[ci].load().await;
}
1 => {
cs[ci].store(val).await;
}
_ => {
let _ = cs[ci].compare_exchange(0, val).await;
}
}
}
Ok(())
});
}
}
#[test]
fn random_atomics_optimal_matches_classes() {
for seed in 0..64u64 {
assert_optimal(&|w| random_atomics(seed, w));
}
}
fn two_producers_two_sends(world: &mut World) {
let (tx, rx) = world.channel::<i32>("ch");
let tx2 = tx.clone();
world.spawn("producer-a", async move {
tx.send(1).await;
tx.send(2).await;
Ok(())
});
world.spawn("producer-b", async move {
tx2.send(3).await;
tx2.send(4).await;
Ok(())
});
world.spawn("consumer", async move {
for _ in 0..4 {
rx.recv().await;
}
Ok(())
});
}
#[test]
fn channel_two_producers_two_sends_optimal() {
assert_optimal(&two_producers_two_sends);
assert_eq!(leaves(&two_producers_two_sends), 6);
}
fn branch_same_cell(world: &mut World) {
let x = world.atomic("x", 0i32);
let y = world.atomic("y", 0i32);
spawn_store(world, "writer", x.clone(), 1);
world.spawn("reader", async move {
if x.load().await == 0 {
x.store(2).await;
} else {
y.load().await;
}
Ok(())
});
}
#[test]
fn branch_same_cell_optimal() {
assert_optimal(&branch_same_cell);
}
mod custom_object {
use std::cell::{Cell, RefCell};
use std::future::poll_fn;
use std::rc::Rc;
use std::task::{Poll, Waker};
use crate::model::{Object, ObjectID, Transition, World};
#[derive(Clone, Copy)]
enum Op {
Inc,
Get,
}
struct Req {
transition: Transition,
waker: Waker,
op: Op,
done: Rc<Cell<bool>>,
}
#[derive(Default)]
struct Inner {
count: usize,
seq: usize,
requests: Vec<Req>,
history: Vec<(Transition, Op)>,
}
#[derive(Clone)]
struct Counter {
id: ObjectID,
inner: Rc<RefCell<Inner>>,
}
impl Counter {
fn new(id: ObjectID) -> Self {
Self {
id,
inner: Rc::new(RefCell::new(Inner::default())),
}
}
async fn op(&self, op: Op) {
let done = Rc::new(Cell::new(false));
let mut pending = Some(op);
poll_fn(move |cx| {
if done.get() {
return Poll::Ready(());
}
if let Some(op) = pending.take() {
let mut st = self.inner.borrow_mut();
let transition = Transition::new(self.id, st.seq);
st.seq += 1;
st.requests.push(Req {
transition,
waker: cx.waker().clone(),
op,
done: Rc::clone(&done),
});
}
Poll::Pending
})
.await
}
fn op_of(&self, t: Transition) -> Op {
let st = self.inner.borrow();
st.requests
.iter()
.map(|r| (r.transition, r.op))
.chain(st.history.iter().map(|(tt, op)| (*tt, *op)))
.find(|(tt, _)| *tt == t)
.map(|(_, op)| op)
.expect("transition registered on this counter")
}
}
impl Object for Counter {
fn apply(&mut self, t: Transition) {
let mut st = self.inner.borrow_mut();
let i = st
.requests
.iter()
.position(|r| r.transition == t)
.expect("transition must be enabled");
let req = st.requests.remove(i);
if let Op::Inc = req.op {
st.count += 1;
}
st.history.push((t, req.op));
req.done.set(true);
req.waker.wake();
}
fn enabled(&self) -> Vec<Transition> {
self.inner
.borrow()
.requests
.iter()
.map(|r| r.transition)
.collect()
}
fn label(&self, t: Transition) -> String {
match self.op_of(t) {
Op::Inc => "inc".into(),
Op::Get => "get".into(),
}
}
fn depends(&self, t1: Transition, t2: Transition) -> bool {
!matches!((self.op_of(t1), self.op_of(t2)), (Op::Get, Op::Get))
}
}
fn program(world: &mut World) {
let c = world.register("counter", Counter::new);
let (a, b, r) = (c.clone(), c.clone(), c);
world.spawn("inc-a", async move {
a.op(Op::Inc).await;
Ok(())
});
world.spawn("inc-b", async move {
b.op(Op::Inc).await;
Ok(())
});
world.spawn("reader", async move {
r.op(Op::Get).await;
Ok(())
});
}
#[test]
fn register_extension_point_explored_optimally() {
super::assert_optimal(&program);
assert_eq!(super::leaves(&program), 6);
}
}
}