use run;
use atomicmonitor::{AtomMonitor, Ordering};
use std::mem;
use std::marker::PhantomData;
use std::sync::Arc;
use std::sync::atomic::fence;
use futures::Future;
pub unsafe fn scoped<'env, R>(operation: impl FnOnce(&Scope<'env>) -> R) -> R {
let scope = Scope {
running_count: Arc::new(AtomMonitor::new(0)),
not_static: PhantomData,
};
let result = operation(&scope);
scope.running_count.wait_until(|count| {
debug!("count is currently {}", count);
count == 0
});
result
}
pub struct Scope<'env> {
running_count: Arc<AtomMonitor<usize>>,
not_static: PhantomData<&'env ()>,
}
impl<'env> Scope<'env> {
#[must_use]
pub fn wrap<'scope, F: Future<Item=(), Error=()> + Send + 'env>(
&'scope self, future_factory: impl FnOnce() -> F)
-> Box<dyn Future<Item=(), Error=()> + Send + 'static> {
fence(Ordering::SeqCst);
let future = future_factory();
self.running_count.mutate(|count| {
count.fetch_add(1, Ordering::SeqCst)
});
let running_count = self.running_count.clone();
let future = future
.then(move |result| {
running_count.mutate(|count| {
count.fetch_sub(1, Ordering::SeqCst)
});
result
});
let future: Box<dyn Future<Item=(), Error=()> + Send + 'env> =
Box::new(future);
let future: Box<dyn Future<Item=(), Error=()> + Send + 'static> =
unsafe { mem::transmute(future) };
future
}
#[must_use]
pub fn work<'scope>(&'scope self, work: impl FnOnce() + Send + 'env)
-> Box<dyn Future<Item=(), Error=()> + Send + 'static> {
self.wrap(move || run::run(work))
}
}