use crate::{App, PlatformDispatcher, RunnableMeta, RunnableVariant, TaskTiming, profiler};use async_task::Runnable;use futures::channel::mpsc;use parking_lot::{Condvar, Mutex};use smol::prelude::*;use std::{ fmt::Debug, marker::PhantomData, mem::{self, ManuallyDrop}, num::NonZeroUsize, panic::Location, pin::Pin, rc::Rc, sync::{ Arc, atomic::{AtomicUsize, Ordering}, }, task::{Context, Poll}, thread::{self, ThreadId}, time::{Duration, Instant},};use util::TryFutureExt;use waker_fn::waker_fn; #[cfg(any(test, feature = "test-support"))]use rand::rngs::StdRng; /// A pointer to the executor that is currently running,/// for spawning background tasks.#[derive(Clone)]pub struct BackgroundExecutor { #[doc(hidden)] pub dispatcher: Arc<dyn PlatformDispatcher>,} /// A pointer to the executor that is currently running,/// for spawning tasks on the main thread.////// This is intentionally `!Send` via the `not_send` marker field. This is because/// `ForegroundExecutor::spawn` does not require `Send` but checks at runtime that the future is/// only polled from the same thread it was spawned from. These checks would fail when spawning/// foreground tasks from background threads.#[derive(Clone)]pub struct ForegroundExecutor { #[doc(hidden)] pub dispatcher: Arc<dyn PlatformDispatcher>, not_send: PhantomData<Rc<()>>,} /// Realtime task priority#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]#[repr(u8)]pub enum RealtimePriority { /// Audio task Audio, /// Other realtime task #[default] Other,} /// Task priority#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]#[repr(u8)]pub enum Priority { /// Realtime priority /// /// Spawning a task with this priority will spin it off on a separate thread dedicated just to that task. Realtime(RealtimePriority), /// High priority /// /// Only use for tasks that are critical to the user experience / responsiveness of the editor. High, /// Medium priority, probably suits most of your use cases. #[default] Medium, /// Low priority /// /// Prioritize this for background work that can come in large quantities /// to not starve the executor of resources for high priority tasks Low,} impl Priority { #[allow(dead_code)] pub(crate) const fn probability(&self) -> u32 { match self { // realtime priorities are not considered for probability scheduling Priority::Realtime(_) => 0, Priority::High => 60, Priority::Medium => 30, Priority::Low => 10, } }} /// 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<T>(TaskState<T>); #[derive(Debug)]enum TaskState<T> { /// A task that is ready to return a value Ready(Option<T>), /// A task that is currently running. Spawned(async_task::Task<T, RunnableMeta>),} impl<T> Task<T> { /// Creates a new task that will resolve with the value pub fn ready(val: T) -> Self { Task(TaskState::Ready(Some(val))) } /// 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(), } } /// Converts this task into a fallible task that returns `Option<T>`. /// /// Unlike the standard `Task<T>`, a [`FallibleTask`] will return `None` /// if the app was dropped while the task is executing. /// /// # Example /// /// ```ignore /// // Background task that gracefully handles app shutdown: /// cx.background_spawn(async move { /// let result = foreground_task.fallible().await; /// if let Some(value) = result { /// // Process the value /// } /// // If None, app was shut down - just exit gracefully /// }).detach(); /// ``` pub fn fallible(self) -> FallibleTask<T> { FallibleTask(match self.0 { TaskState::Ready(val) => FallibleTaskState::Ready(val), TaskState::Spawned(task) => FallibleTaskState::Spawned(task.fallible()), }) }} impl<E, T> Task<Result<T, E>>where T: 'static, E: 'static + Debug,{ /// Run the task to completion in the background and log any /// errors that occur. #[track_caller] pub fn detach_and_log_err(self, cx: &App) { let location = core::panic::Location::caller(); cx.foreground_executor() .spawn(self.log_tracked_err(*location)) .detach(); }} impl<T> Future for Task<T> { type Output = T; fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> { match unsafe { self.get_unchecked_mut() } { Task(TaskState::Ready(val)) => Poll::Ready(val.take().unwrap()), Task(TaskState::Spawned(task)) => task.poll(cx), } }} /// A task that returns `Option<T>` instead of panicking when cancelled.#[must_use]pub struct FallibleTask<T>(FallibleTaskState<T>); enum FallibleTaskState<T> { /// A task that is ready to return a value Ready(Option<T>), /// A task that is currently running (wraps async_task::FallibleTask). Spawned(async_task::FallibleTask<T, RunnableMeta>),} impl<T> FallibleTask<T> { /// Creates a new fallible task that will resolve with the value. pub fn ready(val: T) -> Self { FallibleTask(FallibleTaskState::Ready(Some(val))) } /// Detaching a task runs it to completion in the background. pub fn detach(self) { match self.0 { FallibleTaskState::Ready(_) => {} FallibleTaskState::Spawned(task) => task.detach(), } }} impl<T> Future for FallibleTask<T> { type Output = Option<T>; fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> { match unsafe { self.get_unchecked_mut() } { FallibleTask(FallibleTaskState::Ready(val)) => Poll::Ready(val.take()), FallibleTask(FallibleTaskState::Spawned(task)) => Pin::new(task).poll(cx), } }} impl<T> std::fmt::Debug for FallibleTask<T> { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match &self.0 { FallibleTaskState::Ready(_) => f.debug_tuple("FallibleTask::Ready").finish(), FallibleTaskState::Spawned(task) => { f.debug_tuple("FallibleTask::Spawned").field(task).finish() } } }} /// A task label is an opaque identifier that you can use to/// refer to a task in tests.#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]pub struct TaskLabel(NonZeroUsize); impl Default for TaskLabel { fn default() -> Self { Self::new() }} impl TaskLabel { /// Construct a new task label. pub fn new() -> Self { static NEXT_TASK_LABEL: AtomicUsize = AtomicUsize::new(1); Self( NEXT_TASK_LABEL .fetch_add(1, Ordering::SeqCst) .try_into() .unwrap(), ) }} type AnyLocalFuture<R> = Pin<Box<dyn 'static + Future<Output = R>>>; type AnyFuture<R> = Pin<Box<dyn 'static + Send + Future<Output = R>>>; /// BackgroundExecutor lets you run things on background threads./// In production this is a thread pool with no ordering guarantees./// In tests this is simulated by running tasks one by one in a deterministic/// (but arbitrary) order controlled by the `SEED` environment variable.impl BackgroundExecutor { #[doc(hidden)] pub fn new(dispatcher: Arc<dyn PlatformDispatcher>) -> Self { Self { dispatcher } } /// Enqueues the given future to be run to completion on a background thread. #[track_caller] pub fn spawn<R>(&self, future: impl Future<Output = R> + Send + 'static) -> Task<R> where R: Send + 'static, { self.spawn_with_priority(Priority::default(), future) } /// Enqueues the given future to be run to completion on a background thread. #[track_caller] pub fn spawn_with_priority<R>( &self, priority: Priority, future: impl Future<Output = R> + Send + 'static, ) -> Task<R> where R: Send + 'static, { self.spawn_internal::<R>(Box::pin(future), None, priority) } /// Enqueues the given future to be run to completion on a background thread and blocking the current task on it. /// /// This allows to spawn background work that borrows from its scope. Note that the supplied future will run to /// completion before the current task is resumed, even if the current task is slated for cancellation. pub async fn await_on_background<R>(&self, future: impl Future<Output = R> + Send) -> R where R: Send, { // We need to ensure that cancellation of the parent task does not drop the environment // before the our own task has completed or got cancelled. struct NotifyOnDrop<'a>(&'a (Condvar, Mutex<bool>)); impl Drop for NotifyOnDrop<'_> { fn drop(&mut self) { *self.0.1.lock() = true; self.0.0.notify_all(); } } struct WaitOnDrop<'a>(&'a (Condvar, Mutex<bool>)); impl Drop for WaitOnDrop<'_> { fn drop(&mut self) { let mut done = self.0.1.lock(); if !*done { self.0.0.wait(&mut done); } } } let dispatcher = self.dispatcher.clone(); let location = core::panic::Location::caller(); let pair = &(Condvar::new(), Mutex::new(false)); let _wait_guard = WaitOnDrop(pair); let (runnable, task) = unsafe { async_task::Builder::new() .metadata(RunnableMeta { location, app: None, }) .spawn_unchecked( move |_| async { let _notify_guard = NotifyOnDrop(pair); future.await }, move |runnable| { dispatcher.dispatch( RunnableVariant::Meta(runnable), None, Priority::default(), ) }, ) }; runnable.schedule(); task.await } /// Enqueues the given future to be run to completion on a background thread. /// The given label can be used to control the priority of the task in tests. #[track_caller] pub fn spawn_labeled<R>( &self, label: TaskLabel, future: impl Future<Output = R> + Send + 'static, ) -> Task<R> where R: Send + 'static, { self.spawn_internal::<R>(Box::pin(future), Some(label), Priority::default()) } #[track_caller] fn spawn_internal<R: Send + 'static>( &self, future: AnyFuture<R>, label: Option<TaskLabel>, priority: Priority, ) -> Task<R> { let dispatcher = self.dispatcher.clone(); let (runnable, task) = if let Priority::Realtime(realtime) = priority { let location = core::panic::Location::caller(); let (mut tx, rx) = flume::bounded::<Runnable<RunnableMeta>>(1); dispatcher.spawn_realtime( realtime, Box::new(move || { while let Ok(runnable) = rx.recv() { let start = Instant::now(); let location = runnable.metadata().location; let mut timing = TaskTiming { location, start, end: None, }; profiler::add_task_timing(timing); runnable.run(); let end = Instant::now(); timing.end = Some(end); profiler::add_task_timing(timing); } }), ); async_task::Builder::new() .metadata(RunnableMeta { location, app: None, }) .spawn( move |_| future, move |runnable| { let _ = tx.send(runnable); }, ) } else { let location = core::panic::Location::caller(); async_task::Builder::new() .metadata(RunnableMeta { location, app: None, }) .spawn( move |_| future, move |runnable| { dispatcher.dispatch(RunnableVariant::Meta(runnable), label, priority) }, ) }; runnable.schedule(); Task(TaskState::Spawned(task)) } /// Used by the test harness to run an async test in a synchronous fashion. #[cfg(any(test, feature = "test-support"))] #[track_caller] pub fn block_test<R>(&self, future: impl Future<Output = R>) -> R { if let Ok(value) = self.block_internal(false, future, None) { value } else { unreachable!() } } /// Block the current thread until the given future resolves. /// Consider using `block_with_timeout` instead. pub fn block<R>(&self, future: impl Future<Output = R>) -> R { if let Ok(value) = self.block_internal(true, future, None) { value } else { unreachable!() } } #[cfg(not(any(test, feature = "test-support")))] pub(crate) fn block_internal<Fut: Future>( &self, _background_only: bool, future: Fut, timeout: Option<Duration>, ) -> Result<Fut::Output, impl Future<Output = Fut::Output> + use<Fut>> { use std::time::Instant; let mut future = Box::pin(future); if timeout == Some(Duration::ZERO) { return Err(future); } let deadline = timeout.map(|timeout| Instant::now() + timeout); let parker = parking::Parker::new(); let unparker = parker.unparker(); let waker = waker_fn(move || { unparker.unpark(); }); let mut cx = std::task::Context::from_waker(&waker); loop { match future.as_mut().poll(&mut cx) { Poll::Ready(result) => return Ok(result), Poll::Pending => { let timeout = deadline.map(|deadline| deadline.saturating_duration_since(Instant::now())); if let Some(timeout) = timeout { if !parker.park_timeout(timeout) && deadline.is_some_and(|deadline| deadline < Instant::now()) { return Err(future); } } else { parker.park(); } } } } } #[cfg(any(test, feature = "test-support"))] #[track_caller] pub(crate) fn block_internal<Fut: Future>( &self, background_only: bool, future: Fut, timeout: Option<Duration>, ) -> Result<Fut::Output, impl Future<Output = Fut::Output> + use<Fut>> { use std::sync::atomic::AtomicBool; use std::time::Instant;Lines 1–500 of 1239. Load more
use crate::{App, PlatformDispatcher, RunnableMeta, RunnableVariant, TaskTiming, profiler}; use async_task::Runnable; use futures::channel::mpsc; use parking_lot::{Condvar, Mutex}; use smol::prelude::*; use std::{ fmt::Debug, marker::PhantomData, mem::{self, ManuallyDrop}, num::NonZeroUsize, panic::Location, pin::Pin, rc::Rc, sync::{ Arc, atomic::{AtomicUsize, Ordering}, }, task::{Context, Poll}, thread::{self, ThreadId}, time::{Duration, Instant}, }; use util::TryFutureExt; use waker_fn::waker_fn; #[cfg(any(test, feature = "test-support"))] use rand::rngs::StdRng; /// A pointer to the executor that is currently running, /// for spawning background tasks. #[derive(Clone)] pub struct BackgroundExecutor { #[doc(hidden)] pub dispatcher: Arc<dyn PlatformDispatcher>, } /// A pointer to the executor that is currently running, /// for spawning tasks on the main thread. /// /// This is intentionally `!Send` via the `not_send` marker field. This is because /// `ForegroundExecutor::spawn` does not require `Send` but checks at runtime that the future is /// only polled from the same thread it was spawned from. These checks would fail when spawning /// foreground tasks from background threads. #[derive(Clone)] pub struct ForegroundExecutor { #[doc(hidden)] pub dispatcher: Arc<dyn PlatformDispatcher>, not_send: PhantomData<Rc<()>>, } /// Realtime task priority #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] #[repr(u8)] pub enum RealtimePriority { /// Audio task Audio, /// Other realtime task #[default] Other, } /// Task priority #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] #[repr(u8)] pub enum Priority { /// Realtime priority /// /// Spawning a task with this priority will spin it off on a separate thread dedicated just to that task. Realtime(RealtimePriority), /// High priority /// /// Only use for tasks that are critical to the user experience / responsiveness of the editor. High, /// Medium priority, probably suits most of your use cases. #[default] Medium, /// Low priority /// /// Prioritize this for background work that can come in large quantities /// to not starve the executor of resources for high priority tasks Low, } impl Priority { #[allow(dead_code)] pub(crate) const fn probability(&self) -> u32 { match self { // realtime priorities are not considered for probability scheduling Priority::Realtime(_) => 0, Priority::High => 60, Priority::Medium => 30, Priority::Low => 10, } } } /// 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<T>(TaskState<T>); #[derive(Debug)] enum TaskState<T> { /// A task that is ready to return a value Ready(Option<T>), /// A task that is currently running. Spawned(async_task::Task<T, RunnableMeta>), } impl<T> Task<T> { /// Creates a new task that will resolve with the value pub fn ready(val: T) -> Self { Task(TaskState::Ready(Some(val))) } /// 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(), } } /// Converts this task into a fallible task that returns `Option<T>`. /// /// Unlike the standard `Task<T>`, a [`FallibleTask`] will return `None` /// if the app was dropped while the task is executing. /// /// # Example /// /// ```ignore /// // Background task that gracefully handles app shutdown: /// cx.background_spawn(async move { /// let result = foreground_task.fallible().await; /// if let Some(value) = result { /// // Process the value /// } /// // If None, app was shut down - just exit gracefully /// }).detach(); /// ``` pub fn fallible(self) -> FallibleTask<T> { FallibleTask(match self.0 { TaskState::Ready(val) => FallibleTaskState::Ready(val), TaskState::Spawned(task) => FallibleTaskState::Spawned(task.fallible()), }) } } impl<E, T> Task<Result<T, E>> where T: 'static, E: 'static + Debug, { /// Run the task to completion in the background and log any /// errors that occur. #[track_caller] pub fn detach_and_log_err(self, cx: &App) { let location = core::panic::Location::caller(); cx.foreground_executor() .spawn(self.log_tracked_err(*location)) .detach(); } } impl<T> Future for Task<T> { type Output = T; fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> { match unsafe { self.get_unchecked_mut() } { Task(TaskState::Ready(val)) => Poll::Ready(val.take().unwrap()), Task(TaskState::Spawned(task)) => task.poll(cx), } } } /// A task that returns `Option<T>` instead of panicking when cancelled. #[must_use] pub struct FallibleTask<T>(FallibleTaskState<T>); enum FallibleTaskState<T> { /// A task that is ready to return a value Ready(Option<T>), /// A task that is currently running (wraps async_task::FallibleTask). Spawned(async_task::FallibleTask<T, RunnableMeta>), } impl<T> FallibleTask<T> { /// Creates a new fallible task that will resolve with the value. pub fn ready(val: T) -> Self { FallibleTask(FallibleTaskState::Ready(Some(val))) } /// Detaching a task runs it to completion in the background. pub fn detach(self) { match self.0 { FallibleTaskState::Ready(_) => {} FallibleTaskState::Spawned(task) => task.detach(), } } } impl<T> Future for FallibleTask<T> { type Output = Option<T>; fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> { match unsafe { self.get_unchecked_mut() } { FallibleTask(FallibleTaskState::Ready(val)) => Poll::Ready(val.take()), FallibleTask(FallibleTaskState::Spawned(task)) => Pin::new(task).poll(cx), } } } impl<T> std::fmt::Debug for FallibleTask<T> { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match &self.0 { FallibleTaskState::Ready(_) => f.debug_tuple("FallibleTask::Ready").finish(), FallibleTaskState::Spawned(task) => { f.debug_tuple("FallibleTask::Spawned").field(task).finish() } } } } /// A task label is an opaque identifier that you can use to /// refer to a task in tests. #[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)] pub struct TaskLabel(NonZeroUsize); impl Default for TaskLabel { fn default() -> Self { Self::new() } } impl TaskLabel { /// Construct a new task label. pub fn new() -> Self { static NEXT_TASK_LABEL: AtomicUsize = AtomicUsize::new(1); Self( NEXT_TASK_LABEL .fetch_add(1, Ordering::SeqCst) .try_into() .unwrap(), ) } } type AnyLocalFuture<R> = Pin<Box<dyn 'static + Future<Output = R>>>; type AnyFuture<R> = Pin<Box<dyn 'static + Send + Future<Output = R>>>; /// BackgroundExecutor lets you run things on background threads. /// In production this is a thread pool with no ordering guarantees. /// In tests this is simulated by running tasks one by one in a deterministic /// (but arbitrary) order controlled by the `SEED` environment variable. impl BackgroundExecutor { #[doc(hidden)] pub fn new(dispatcher: Arc<dyn PlatformDispatcher>) -> Self { Self { dispatcher } } /// Enqueues the given future to be run to completion on a background thread. #[track_caller] pub fn spawn<R>(&self, future: impl Future<Output = R> + Send + 'static) -> Task<R> where R: Send + 'static, { self.spawn_with_priority(Priority::default(), future) } /// Enqueues the given future to be run to completion on a background thread. #[track_caller] pub fn spawn_with_priority<R>( &self, priority: Priority, future: impl Future<Output = R> + Send + 'static, ) -> Task<R> where R: Send + 'static, { self.spawn_internal::<R>(Box::pin(future), None, priority) } /// Enqueues the given future to be run to completion on a background thread and blocking the current task on it. /// /// This allows to spawn background work that borrows from its scope. Note that the supplied future will run to /// completion before the current task is resumed, even if the current task is slated for cancellation. pub async fn await_on_background<R>(&self, future: impl Future<Output = R> + Send) -> R where R: Send, { // We need to ensure that cancellation of the parent task does not drop the environment // before the our own task has completed or got cancelled. struct NotifyOnDrop<'a>(&'a (Condvar, Mutex<bool>)); impl Drop for NotifyOnDrop<'_> { fn drop(&mut self) { *self.0.1.lock() = true; self.0.0.notify_all(); } } struct WaitOnDrop<'a>(&'a (Condvar, Mutex<bool>)); impl Drop for WaitOnDrop<'_> { fn drop(&mut self) { let mut done = self.0.1.lock(); if !*done { self.0.0.wait(&mut done); } } } let dispatcher = self.dispatcher.clone(); let location = core::panic::Location::caller(); let pair = &(Condvar::new(), Mutex::new(false)); let _wait_guard = WaitOnDrop(pair); let (runnable, task) = unsafe { async_task::Builder::new() .metadata(RunnableMeta { location, app: None, }) .spawn_unchecked( move |_| async { let _notify_guard = NotifyOnDrop(pair); future.await }, move |runnable| { dispatcher.dispatch( RunnableVariant::Meta(runnable), None, Priority::default(), ) }, ) }; runnable.schedule(); task.await } /// Enqueues the given future to be run to completion on a background thread. /// The given label can be used to control the priority of the task in tests. #[track_caller] pub fn spawn_labeled<R>( &self, label: TaskLabel, future: impl Future<Output = R> + Send + 'static, ) -> Task<R> where R: Send + 'static, { self.spawn_internal::<R>(Box::pin(future), Some(label), Priority::default()) } #[track_caller] fn spawn_internal<R: Send + 'static>( &self, future: AnyFuture<R>, label: Option<TaskLabel>, priority: Priority, ) -> Task<R> { let dispatcher = self.dispatcher.clone(); let (runnable, task) = if let Priority::Realtime(realtime) = priority { let location = core::panic::Location::caller(); let (mut tx, rx) = flume::bounded::<Runnable<RunnableMeta>>(1); dispatcher.spawn_realtime( realtime, Box::new(move || { while let Ok(runnable) = rx.recv() { let start = Instant::now(); let location = runnable.metadata().location; let mut timing = TaskTiming { location, start, end: None, }; profiler::add_task_timing(timing); runnable.run(); let end = Instant::now(); timing.end = Some(end); profiler::add_task_timing(timing); } }), ); async_task::Builder::new() .metadata(RunnableMeta { location, app: None, }) .spawn( move |_| future, move |runnable| { let _ = tx.send(runnable); }, ) } else { let location = core::panic::Location::caller(); async_task::Builder::new() .metadata(RunnableMeta { location, app: None, }) .spawn( move |_| future, move |runnable| { dispatcher.dispatch(RunnableVariant::Meta(runnable), label, priority) }, ) }; runnable.schedule(); Task(TaskState::Spawned(task)) } /// Used by the test harness to run an async test in a synchronous fashion. #[cfg(any(test, feature = "test-support"))] #[track_caller] pub fn block_test<R>(&self, future: impl Future<Output = R>) -> R { if let Ok(value) = self.block_internal(false, future, None) { value } else { unreachable!() } } /// Block the current thread until the given future resolves. /// Consider using `block_with_timeout` instead. pub fn block<R>(&self, future: impl Future<Output = R>) -> R { if let Ok(value) = self.block_internal(true, future, None) { value } else { unreachable!() } } #[cfg(not(any(test, feature = "test-support")))] pub(crate) fn block_internal<Fut: Future>( &self, _background_only: bool, future: Fut, timeout: Option<Duration>, ) -> Result<Fut::Output, impl Future<Output = Fut::Output> + use<Fut>> { use std::time::Instant; let mut future = Box::pin(future); if timeout == Some(Duration::ZERO) { return Err(future); } let deadline = timeout.map(|timeout| Instant::now() + timeout); let parker = parking::Parker::new(); let unparker = parker.unparker(); let waker = waker_fn(move || { unparker.unpark(); }); let mut cx = std::task::Context::from_waker(&waker); loop { match future.as_mut().poll(&mut cx) { Poll::Ready(result) => return Ok(result), Poll::Pending => { let timeout = deadline.map(|deadline| deadline.saturating_duration_since(Instant::now())); if let Some(timeout) = timeout { if !parker.park_timeout(timeout) && deadline.is_some_and(|deadline| deadline < Instant::now()) { return Err(future); } } else { parker.park(); } } } } } #[cfg(any(test, feature = "test-support"))] #[track_caller] pub(crate) fn block_internal<Fut: Future>( &self, background_only: bool, future: Fut, timeout: Option<Duration>, ) -> Result<Fut::Output, impl Future<Output = Fut::Output> + use<Fut>> { use std::sync::atomic::AtomicBool; use std::time::Instant;