use alloc::sync::Arc;
use core::mem;
use crate::sync::Mutex;
use crate::prelude::*;
#[cfg(feature = "std")]
use crate::sync::Condvar;
#[cfg(feature = "std")]
use std::time::Duration;
use core::future::Future as StdFuture;
use core::task::{Context, Poll};
use core::pin::Pin;
pub(crate) struct Notifier {
notify_pending: Mutex<(bool, Option<Arc<Mutex<FutureState>>>)>,
}
impl Notifier {
pub(crate) fn new() -> Self {
Self {
notify_pending: Mutex::new((false, None)),
}
}
pub(crate) fn notify(&self) {
let mut lock = self.notify_pending.lock().unwrap();
if let Some(future_state) = &lock.1 {
if complete_future(future_state) {
lock.1 = None;
return;
}
}
lock.0 = true;
}
pub(crate) fn get_future(&self) -> Future {
let mut lock = self.notify_pending.lock().unwrap();
if let Some(existing_state) = &lock.1 {
if existing_state.lock().unwrap().callbacks_made {
lock.1.take();
lock.0 = false;
}
}
if let Some(existing_state) = &lock.1 {
Future { state: Arc::clone(&existing_state) }
} else {
let state = Arc::new(Mutex::new(FutureState {
callbacks: Vec::new(),
callbacks_with_state: Vec::new(),
complete: lock.0,
callbacks_made: false,
}));
lock.1 = Some(Arc::clone(&state));
Future { state }
}
}
#[cfg(any(test, feature = "_test_utils"))]
pub fn notify_pending(&self) -> bool {
self.notify_pending.lock().unwrap().0
}
}
macro_rules! define_callback { ($($bounds: path),*) => {
pub trait FutureCallback : $($bounds +)* {
fn call(&self);
}
impl<F: Fn() $(+ $bounds)*> FutureCallback for F {
fn call(&self) { (self)(); }
}
} }
#[cfg(feature = "std")]
define_callback!(Send);
#[cfg(not(feature = "std"))]
define_callback!();
pub(crate) struct FutureState {
callbacks: Vec<(bool, Box<dyn FutureCallback>)>,
callbacks_with_state: Vec<(bool, Box<dyn Fn(&Arc<Mutex<FutureState>>) -> () + Send>)>,
complete: bool,
callbacks_made: bool,
}
fn complete_future(this: &Arc<Mutex<FutureState>>) -> bool {
let mut state_lock = this.lock().unwrap();
let state = &mut *state_lock;
for (counts_as_call, callback) in state.callbacks.drain(..) {
callback.call();
state.callbacks_made |= counts_as_call;
}
for (counts_as_call, callback) in state.callbacks_with_state.drain(..) {
(callback)(this);
state.callbacks_made |= counts_as_call;
}
state.complete = true;
state.callbacks_made
}
#[derive(Clone)]
pub struct Future {
state: Arc<Mutex<FutureState>>,
}
impl Future {
pub fn register_callback(&self, callback: Box<dyn FutureCallback>) {
let mut state = self.state.lock().unwrap();
if state.complete {
state.callbacks_made = true;
mem::drop(state);
callback.call();
} else {
state.callbacks.push((true, callback));
}
}
#[cfg(c_bindings)]
pub fn register_callback_fn<F: 'static + FutureCallback>(&self, callback: F) {
self.register_callback(Box::new(callback));
}
#[cfg(feature = "std")]
pub fn wait(self) {
Sleeper::from_single_future(self).wait();
}
#[cfg(feature = "std")]
pub fn wait_timeout(self, max_wait: Duration) -> bool {
Sleeper::from_single_future(self).wait_timeout(max_wait)
}
#[cfg(test)]
pub fn poll_is_complete(&self) -> bool {
let mut state = self.state.lock().unwrap();
if state.complete {
state.callbacks_made = true;
true
} else { false }
}
}
use core::task::Waker;
struct StdWaker(pub Waker);
impl FutureCallback for StdWaker {
fn call(&self) { self.0.wake_by_ref() }
}
impl<'a> StdFuture for Future {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut state = self.state.lock().unwrap();
if state.complete {
state.callbacks_made = true;
Poll::Ready(())
} else {
let waker = cx.waker().clone();
state.callbacks.push((false, Box::new(StdWaker(waker))));
Poll::Pending
}
}
}
#[cfg(feature = "std")]
pub struct Sleeper {
notifiers: Vec<Arc<Mutex<FutureState>>>,
}
#[cfg(feature = "std")]
impl Sleeper {
pub fn from_single_future(future: Future) -> Self {
Self { notifiers: vec![future.state] }
}
pub fn from_two_futures(fut_a: Future, fut_b: Future) -> Self {
Self { notifiers: vec![fut_a.state, fut_b.state] }
}
pub fn new(futures: Vec<Future>) -> Self {
Self { notifiers: futures.into_iter().map(|f| f.state).collect() }
}
fn setup_wait(&self) -> (Arc<Condvar>, Arc<Mutex<Option<Arc<Mutex<FutureState>>>>>) {
let cv = Arc::new(Condvar::new());
let notified_fut_mtx = Arc::new(Mutex::new(None));
{
for notifier_mtx in self.notifiers.iter() {
let cv_ref = Arc::clone(&cv);
let notified_fut_ref = Arc::clone(¬ified_fut_mtx);
let mut notifier = notifier_mtx.lock().unwrap();
if notifier.complete {
*notified_fut_mtx.lock().unwrap() = Some(Arc::clone(¬ifier_mtx));
break;
}
notifier.callbacks_with_state.push((false, Box::new(move |notifier_ref| {
*notified_fut_ref.lock().unwrap() = Some(Arc::clone(notifier_ref));
cv_ref.notify_all();
})));
}
}
(cv, notified_fut_mtx)
}
pub fn wait(&self) {
let (cv, notified_fut_mtx) = self.setup_wait();
let notified_fut = cv.wait_while(notified_fut_mtx.lock().unwrap(), |fut_opt| fut_opt.is_none())
.unwrap().take().expect("CV wait shouldn't have returned until the notifying future was set");
notified_fut.lock().unwrap().callbacks_made = true;
}
pub fn wait_timeout(&self, max_wait: Duration) -> bool {
let (cv, notified_fut_mtx) = self.setup_wait();
let notified_fut =
match cv.wait_timeout_while(notified_fut_mtx.lock().unwrap(), max_wait, |fut_opt| fut_opt.is_none()) {
Ok((_, e)) if e.timed_out() => return false,
Ok((mut notified_fut, _)) =>
notified_fut.take().expect("CV wait shouldn't have returned until the notifying future was set"),
Err(_) => panic!("Previous panic while a lock was held led to a lock panic"),
};
notified_fut.lock().unwrap().callbacks_made = true;
true
}
}
#[cfg(test)]
mod tests {
use super::*;
use core::sync::atomic::{AtomicBool, Ordering};
use core::future::Future as FutureTrait;
use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
#[test]
fn notifier_pre_notified_future() {
let notifier = Notifier::new();
notifier.notify();
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(callback.load(Ordering::SeqCst));
}
#[test]
fn notifier_future_completes_wake() {
let notifier = Notifier::new();
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(!callback.load(Ordering::SeqCst));
notifier.notify();
assert!(callback.load(Ordering::SeqCst));
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(!callback.load(Ordering::SeqCst));
notifier.notify();
assert!(callback.load(Ordering::SeqCst));
let future = notifier.get_future();
notifier.notify();
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(callback.load(Ordering::SeqCst));
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(!callback.load(Ordering::SeqCst));
}
#[test]
fn new_future_wipes_notify_bit() {
let notifier = Notifier::new();
notifier.notify();
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(callback.load(Ordering::SeqCst));
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(!callback.load(Ordering::SeqCst));
notifier.notify();
assert!(callback.load(Ordering::SeqCst));
}
#[cfg(feature = "std")]
#[test]
fn test_wait_timeout() {
use crate::sync::Arc;
use std::thread;
let persistence_notifier = Arc::new(Notifier::new());
let thread_notifier = Arc::clone(&persistence_notifier);
let exit_thread = Arc::new(AtomicBool::new(false));
let exit_thread_clone = exit_thread.clone();
thread::spawn(move || {
loop {
thread_notifier.notify();
if exit_thread_clone.load(Ordering::SeqCst) {
break
}
}
});
let _ = persistence_notifier.get_future().wait();
loop {
if persistence_notifier.get_future().wait_timeout(Duration::from_millis(100)) {
break
}
}
exit_thread.store(true, Ordering::SeqCst);
loop {
if !persistence_notifier.get_future().wait_timeout(Duration::from_millis(100)) {
break
}
}
}
#[cfg(feature = "std")]
#[test]
fn test_state_drops() {
use crate::sync::Arc;
use std::thread;
let notifier_a = Arc::new(Notifier::new());
let notifier_b = Arc::new(Notifier::new());
let thread_notifier_a = Arc::clone(¬ifier_a);
let future_a = notifier_a.get_future();
let future_state_a = Arc::downgrade(&future_a.state);
let future_b = notifier_b.get_future();
let future_state_b = Arc::downgrade(&future_b.state);
let join_handle = thread::spawn(move || {
std::thread::sleep(Duration::from_millis(50));
thread_notifier_a.notify();
});
Sleeper::from_two_futures(future_a, future_b).wait();
join_handle.join().unwrap();
mem::drop(notifier_a);
mem::drop(notifier_b);
assert!(future_state_a.upgrade().is_none() && future_state_b.upgrade().is_none());
}
#[test]
fn test_future_callbacks() {
let future = Future {
state: Arc::new(Mutex::new(FutureState {
callbacks: Vec::new(),
callbacks_with_state: Vec::new(),
complete: false,
callbacks_made: false,
}))
};
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(!callback.load(Ordering::SeqCst));
complete_future(&future.state);
assert!(callback.load(Ordering::SeqCst));
complete_future(&future.state);
}
#[test]
fn test_pre_completed_future_callbacks() {
let future = Future {
state: Arc::new(Mutex::new(FutureState {
callbacks: Vec::new(),
callbacks_with_state: Vec::new(),
complete: false,
callbacks_made: false,
}))
};
complete_future(&future.state);
let callback = Arc::new(AtomicBool::new(false));
let callback_ref = Arc::clone(&callback);
future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
assert!(callback.load(Ordering::SeqCst));
assert!(future.state.lock().unwrap().callbacks.is_empty());
}
const WAKER_V_TABLE: RawWakerVTable = RawWakerVTable::new(waker_clone, wake, wake_by_ref, drop);
unsafe fn wake_by_ref(ptr: *const ()) { let p = ptr as *const Arc<AtomicBool>; assert!(!(*p).fetch_or(true, Ordering::SeqCst)); }
unsafe fn drop(ptr: *const ()) { let p = ptr as *mut Arc<AtomicBool>; let _freed = Box::from_raw(p); }
unsafe fn wake(ptr: *const ()) { wake_by_ref(ptr); drop(ptr); }
unsafe fn waker_clone(ptr: *const ()) -> RawWaker {
let p = ptr as *const Arc<AtomicBool>;
RawWaker::new(Box::into_raw(Box::new(Arc::clone(&*p))) as *const (), &WAKER_V_TABLE)
}
fn create_waker() -> (Arc<AtomicBool>, Waker) {
let a = Arc::new(AtomicBool::new(false));
let waker = unsafe { Waker::from_raw(waker_clone((&a as *const Arc<AtomicBool>) as *const ())) };
(a, waker)
}
#[test]
fn test_future() {
let mut future = Future {
state: Arc::new(Mutex::new(FutureState {
callbacks: Vec::new(),
callbacks_with_state: Vec::new(),
complete: false,
callbacks_made: false,
}))
};
let mut second_future = Future { state: Arc::clone(&future.state) };
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
assert!(!woken.load(Ordering::SeqCst));
let (second_woken, second_waker) = create_waker();
assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Pending);
assert!(!second_woken.load(Ordering::SeqCst));
complete_future(&future.state);
assert!(woken.load(Ordering::SeqCst));
assert!(second_woken.load(Ordering::SeqCst));
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Ready(()));
}
#[test]
#[cfg(feature = "std")]
fn test_dropped_future_doesnt_count() {
let notifier = Notifier::new();
notifier.notify();
notifier.get_future();
assert!(notifier.get_future().wait_timeout(Duration::from_millis(1)));
assert!(!notifier.get_future().wait_timeout(Duration::from_millis(1)));
let mut future = notifier.get_future();
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
notifier.notify();
assert!(woken.load(Ordering::SeqCst));
assert!(notifier.get_future().wait_timeout(Duration::from_millis(1)));
let mut future = notifier.get_future();
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
notifier.notify();
assert!(woken.load(Ordering::SeqCst));
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
assert!(!notifier.get_future().wait_timeout(Duration::from_millis(1)));
}
#[test]
fn test_poll_post_notify_completes() {
let notifier = Notifier::new();
notifier.notify();
let mut future = notifier.get_future();
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
assert!(!woken.load(Ordering::SeqCst));
notifier.notify();
let mut future = notifier.get_future();
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
assert!(!woken.load(Ordering::SeqCst));
let mut future = notifier.get_future();
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
assert!(!woken.load(Ordering::SeqCst));
notifier.notify();
assert!(woken.load(Ordering::SeqCst));
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
}
#[test]
fn test_poll_post_notify_completes_initial_notified() {
let notifier = Notifier::new();
let mut future = notifier.get_future();
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
notifier.notify();
assert!(woken.load(Ordering::SeqCst));
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
notifier.notify();
let mut future = notifier.get_future();
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
assert!(!woken.load(Ordering::SeqCst));
let mut future = notifier.get_future();
let (woken, waker) = create_waker();
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
assert!(!woken.load(Ordering::SeqCst));
notifier.notify();
assert!(woken.load(Ordering::SeqCst));
assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
}
#[test]
#[cfg(feature = "std")]
fn test_multi_future_sleep() {
let notifier_a = Notifier::new();
let notifier_b = Notifier::new();
notifier_a.notify();
notifier_b.notify();
Sleeper::from_two_futures(notifier_a.get_future(), notifier_b.get_future()).wait();
Sleeper::from_two_futures(notifier_a.get_future(), notifier_b.get_future()).wait();
assert!(!Sleeper::from_two_futures(notifier_a.get_future(), notifier_b.get_future())
.wait_timeout(Duration::from_millis(10)));
notifier_a.notify();
Sleeper::from_two_futures(notifier_a.get_future(), notifier_b.get_future()).wait();
}
#[test]
#[cfg(feature = "std")]
fn sleeper_with_pending_callbacks() {
let notifier_a = Notifier::new();
let notifier_b = Notifier::new();
notifier_a.notify();
notifier_b.notify();
Sleeper::from_two_futures(notifier_a.get_future(), notifier_b.get_future()).wait();
let callback_a = Arc::new(AtomicBool::new(false));
let callback_b = Arc::new(AtomicBool::new(false));
let callback_a_ref = Arc::clone(&callback_a);
let callback_b_ref = Arc::clone(&callback_b);
notifier_a.get_future().register_callback(Box::new(move || assert!(!callback_a_ref.fetch_or(true, Ordering::SeqCst))));
notifier_b.get_future().register_callback(Box::new(move || assert!(!callback_b_ref.fetch_or(true, Ordering::SeqCst))));
assert!(callback_a.load(Ordering::SeqCst) ^ callback_b.load(Ordering::SeqCst));
notifier_a.notify();
notifier_b.notify();
assert!(callback_a.load(Ordering::SeqCst) && callback_b.load(Ordering::SeqCst));
Sleeper::from_two_futures(notifier_a.get_future(), notifier_b.get_future()).wait();
assert!(!Sleeper::from_two_futures(notifier_a.get_future(), notifier_b.get_future())
.wait_timeout(Duration::from_millis(10)));
}
}