use crate::loom::sync::atomic::AtomicUsize;
use crate::loom::sync::Arc;
use crate::loom::thread;
use crate::runtime::{Builder, Runtime};
use crate::sync::oneshot::{self, Receiver};
use crate::task;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::Ordering::{Acquire, Release};
use std::task::{Context, Poll};
fn assert_at_most_num_polls(rt: Arc<Runtime>, at_most_polls: usize) {
let (tx, rx) = oneshot::channel();
let num_polls = Arc::new(AtomicUsize::new(0));
rt.spawn(async move {
for _ in 0..12 {
task::yield_now().await;
}
tx.send(()).unwrap();
});
rt.block_on(async {
BlockedFuture {
rx,
num_polls: num_polls.clone(),
}
.await;
});
let polls = num_polls.load(Acquire);
assert!(polls <= at_most_polls);
}
#[test]
fn block_on_num_polls() {
loom::model(|| {
let at_most = 4;
let rt1 = Arc::new(Builder::new_current_thread().build().unwrap());
let rt2 = rt1.clone();
let rt3 = rt1.clone();
let th1 = thread::spawn(move || assert_at_most_num_polls(rt1, at_most));
let th2 = thread::spawn(move || assert_at_most_num_polls(rt2, at_most));
let th3 = thread::spawn(move || assert_at_most_num_polls(rt3, at_most));
th1.join().unwrap();
th2.join().unwrap();
th3.join().unwrap();
});
}
#[test]
fn assert_no_unnecessary_polls() {
loom::model(|| {
let rt = Builder::new_current_thread().build().unwrap();
let (tx, rx) = oneshot::channel();
let pending_cnt = Arc::new(AtomicUsize::new(0));
rt.spawn(async move {
for _ in 0..24 {
task::yield_now().await;
}
tx.send(()).unwrap();
});
let pending_cnt_clone = pending_cnt.clone();
rt.block_on(async move {
task::yield_now().await;
ResetFuture {
rx,
pending_cnt: pending_cnt_clone,
}
.await;
});
let pending_cnt = pending_cnt.load(Acquire);
assert!(pending_cnt <= 1);
});
}
struct BlockedFuture {
rx: Receiver<()>,
num_polls: Arc<AtomicUsize>,
}
impl Future for BlockedFuture {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.num_polls.fetch_add(1, Release);
match Pin::new(&mut self.rx).poll(cx) {
Poll::Pending => Poll::Pending,
_ => Poll::Ready(()),
}
}
}
struct ResetFuture {
rx: Receiver<()>,
pending_cnt: Arc<AtomicUsize>,
}
impl Future for ResetFuture {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match Pin::new(&mut self.rx).poll(cx) {
Poll::Pending => {
self.pending_cnt.fetch_add(1, Release);
Poll::Pending
}
_ => Poll::Ready(()),
}
}
}