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(),        }    }} 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 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 })                .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>,        #[cfg_attr(            target_os = "windows",            expect(                unused_variables,                reason = "Multi priority scheduler is broken on windows"            )        )]        priority: Priority,    ) -> Task<R> {        let dispatcher = self.dispatcher.clone();        #[cfg(target_os = "windows")]        let priority = Priority::Medium; // multi-prio scheduler is broken on windows         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 })                .spawn(                    move |_| future,                    move |runnable| {                        let _ = tx.send(runnable);                    },                )        } else {            let location = core::panic::Location::caller();            async_task::Builder::new()                .metadata(RunnableMeta { location })                .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 parking::Parker;         let mut future = Box::pin(future);        if timeout == Some(Duration::ZERO) {            return Err(future);        }        let Some(dispatcher) = self.dispatcher.as_test() else {            return Err(future);        };         let mut max_ticks = if timeout.is_some() {            dispatcher.gen_block_on_ticks()        } else {            usize::MAX        };         let parker = Parker::new();        let unparker = parker.unparker();         let awoken = Arc::new(AtomicBool::new(false));        let waker = waker_fn({            let awoken = awoken.clone();            let unparker = unparker.clone();            move || {                awoken.store(true, Ordering::SeqCst);                unparker.unpark();            }        });        let mut cx = std::task::Context::from_waker(&waker);         let duration = Duration::from_secs(            option_env!("GPUI_TEST_TIMEOUT")                .and_then(|s| s.parse::<u64>().ok())                .unwrap_or(180),        );        let mut test_should_end_by = Instant::now() + duration;         loop {            match future.as_mut().poll(&mut cx) {                Poll::Ready(result) => return Ok(result),                Poll::Pending => {                    if max_ticks == 0 {                        return Err(future);                    }                    max_ticks -= 1;                     if !dispatcher.tick(background_only) {                        if awoken.swap(false, Ordering::SeqCst) {                            continue;                        }                         if !dispatcher.parking_allowed() {                            if dispatcher.advance_clock_to_next_delayed() {                                continue;                            }                            let mut backtrace_message = String::new();                            let mut waiting_message = String::new();                            if let Some(backtrace) = dispatcher.waiting_backtrace() {                                backtrace_message =                                    format!("\nbacktrace of waiting future:\n{:?}", backtrace);                            }                            if let Some(waiting_hint) = dispatcher.waiting_hint() {                                waiting_message = format!("\n  waiting on: {}\n", waiting_hint);                            }                            panic!(                                "parked with nothing left to run{waiting_message}{backtrace_message}",                            )                        }                        dispatcher.push_unparker(unparker.clone());                        parker.park_timeout(Duration::from_millis(1));                        if Instant::now() > test_should_end_by {                            panic!("test timed out after {duration:?} with allow_parking")

Lines 1–500 of 849. Load more