use crate::{Scheduler, SessionId, Timer}; use futures::FutureExt as _; use std::{ future::Future, marker::PhantomData, mem::ManuallyDrop, panic::Location, pin::Pin, rc::Rc, sync::Arc, task::{Context, Poll}, thread::{self, ThreadId}, time::Duration, }; #[derive(Clone)] pub struct ForegroundExecutor { session_id: SessionId, scheduler: Arc, not_send: PhantomData>, } impl ForegroundExecutor { pub fn new(session_id: SessionId, scheduler: Arc) -> Self { Self { session_id, scheduler, not_send: PhantomData, } } #[track_caller] pub fn spawn(&self, future: F) -> Task where F: Future + 'static, F::Output: 'static, { let session_id = self.session_id; let scheduler = Arc::clone(&self.scheduler); let (runnable, task) = spawn_local_with_source_location(future, move |runnable| { scheduler.schedule_foreground(session_id, runnable); }); runnable.schedule(); Task(TaskState::Spawned(task)) } pub fn block_on(&self, future: Fut) -> Fut::Output { let mut output = None; self.scheduler.block( Some(self.session_id), async { output = Some(future.await) }.boxed_local(), None, ); output.unwrap() } pub fn block_with_timeout( &self, timeout: Duration, mut future: Fut, ) -> Result { let mut output = None; self.scheduler.block( Some(self.session_id), async { output = Some((&mut future).await) }.boxed_local(), Some(timeout), ); output.ok_or(future) } pub fn timer(&self, duration: Duration) -> Timer { self.scheduler.timer(duration) } } #[derive(Clone)] pub struct BackgroundExecutor { scheduler: Arc, } impl BackgroundExecutor { pub fn new(scheduler: Arc) -> Self { Self { scheduler } } pub fn spawn(&self, future: F) -> Task where F: Future + Send + 'static, F::Output: Send + 'static, { let scheduler = Arc::clone(&self.scheduler); let (runnable, task) = async_task::spawn(future, move |runnable| { scheduler.schedule_background(runnable); }); runnable.schedule(); Task(TaskState::Spawned(task)) } pub fn timer(&self, duration: Duration) -> Timer { self.scheduler.timer(duration) } pub fn scheduler(&self) -> &Arc { &self.scheduler } } /// Task is a primitive that allows work to happen in the background. /// /// It implements [`Future`] so you can `.await` on it. /// /// If you drop a task it will be cancelled immediately. Calling [`Task::detach`] allows /// the task to continue running, but with no way to return a value. #[must_use] #[derive(Debug)] pub struct Task(TaskState); #[derive(Debug)] enum TaskState { /// A task that is ready to return a value Ready(Option), /// A task that is currently running. Spawned(async_task::Task), } impl Task { /// Creates a new task that will resolve with the value pub fn ready(val: T) -> Self { Task(TaskState::Ready(Some(val))) } pub fn is_ready(&self) -> bool { match &self.0 { TaskState::Ready(_) => true, TaskState::Spawned(task) => task.is_finished(), } } /// Detaching a task runs it to completion in the background pub fn detach(self) { match self { Task(TaskState::Ready(_)) => {} Task(TaskState::Spawned(task)) => task.detach(), } } } impl Future for Task { type Output = T; fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll { match unsafe { self.get_unchecked_mut() } { Task(TaskState::Ready(val)) => Poll::Ready(val.take().unwrap()), Task(TaskState::Spawned(task)) => Pin::new(task).poll(cx), } } } /// Variant of `async_task::spawn_local` that includes the source location of the spawn in panics. /// /// Copy-modified from: /// #[track_caller] fn spawn_local_with_source_location( future: Fut, schedule: S, ) -> (async_task::Runnable, async_task::Task) where Fut: Future + 'static, Fut::Output: 'static, S: async_task::Schedule + Send + Sync + 'static, { #[inline] fn thread_id() -> ThreadId { std::thread_local! { static ID: ThreadId = thread::current().id(); } ID.try_with(|id| *id) .unwrap_or_else(|_| thread::current().id()) } struct Checked { id: ThreadId, inner: ManuallyDrop, location: &'static Location<'static>, } impl Drop for Checked { fn drop(&mut self) { assert!( self.id == thread_id(), "local task dropped by a thread that didn't spawn it. Task spawned at {}", self.location ); unsafe { ManuallyDrop::drop(&mut self.inner); } } } impl Future for Checked { type Output = F::Output; fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { assert!( self.id == thread_id(), "local task polled by a thread that didn't spawn it. Task spawned at {}", self.location ); unsafe { self.map_unchecked_mut(|c| &mut *c.inner).poll(cx) } } } // Wrap the future into one that checks which thread it's on. let future = Checked { id: thread_id(), inner: ManuallyDrop::new(future), location: Location::caller(), }; unsafe { async_task::spawn_unchecked(future, schedule) } }