inset_foundation/
executor.rs

1//! The foreground executor: `!Send` futures polled on the main thread at the microtask
2//! checkpoint. Dart's counterpart is the isolate's microtask queue, where `await`
3//! continuations run — one queue with `scheduleMicrotask`'s callbacks, in scheduling order.
4//!
5//! The shape is gpui's `ForegroundExecutor` over `async-task`: a woken task's runnable lands on
6//! the ready queue, and [`AppCell::checkpoint`](crate::AppCell::checkpoint) polls the queue
7//! with no borrow of the [`App`](crate::App) held, so each poll can borrow it through its own
8//! [`AsyncApp`](crate::AsyncApp). Waking also asks the host to wake the app, since the task may
9//! have been woken on another thread while the main thread sleeps in the host's wait — what
10//! gpui's dispatch to the main queue does for every runnable. Work for another thread goes
11//! to the same host dispatcher and comes back as a task.
12
13use std::collections::VecDeque;
14use std::future::Future;
15use std::pin::Pin;
16use std::sync::{Arc, Mutex, MutexGuard};
17use std::task::{Context, Poll};
18
19use async_task::Runnable;
20use inset_embedder::Dispatcher;
21
22use crate::background;
23
24/// Poll budget for one checkpoint's run: two tasks waking each other forever must fail loud,
25/// not starve the event loop — the counterpart of the microtask budget.
26const POLL_BUDGET: usize = 100_000;
27
28/// Tasks woken and not yet polled. A `Waker` may be called from any thread, so the queue is
29/// shared; the runnables themselves only ever run on the main thread.
30type ReadyQueue = Arc<Mutex<VecDeque<Runnable>>>;
31
32/// Spawns `!Send` futures for the main thread. Lives on the `App`, and owns the queue: the
33/// `App` going away cancels the tasks still queued. Everyone else — the cell's checkpoint,
34/// every `AsyncApp` — holds an [`ExecutorHandle`] onto the same queue.
35pub(crate) struct ForegroundExecutor {
36    shared: ExecutorHandle,
37}
38
39impl ForegroundExecutor {
40    /// An executor whose woken tasks ask `dispatcher` to wake the app, and whose background work
41    /// runs on it.
42    pub(crate) fn new(dispatcher: Arc<dyn Dispatcher>) -> ForegroundExecutor {
43        ForegroundExecutor {
44            shared: ExecutorHandle {
45                ready: Arc::default(),
46                dispatcher,
47            },
48        }
49    }
50
51    /// Queues `future` for its first poll at the next checkpoint — never inline, since the
52    /// caller holds the `App` the future will borrow.
53    pub(crate) fn spawn<R: 'static>(&self, future: impl Future<Output = R> + 'static) -> Task<R> {
54        self.shared.spawn(future)
55    }
56
57    /// [`ExecutorHandle::run_in_background`].
58    pub(crate) fn run_in_background<T: Send + 'static>(
59        &self,
60        work: impl FnOnce() -> T + Send + 'static,
61    ) -> Task<T> {
62        self.shared.run_in_background(work)
63    }
64
65    pub(crate) fn handle(&self) -> ExecutorHandle {
66        self.shared.clone()
67    }
68}
69
70/// A queued runnable and the queue own each other through the task's schedule closure, so the
71/// queue is emptied by hand; dropping a runnable can wake further tasks, hence the loop.
72impl Drop for ForegroundExecutor {
73    fn drop(&mut self) {
74        loop {
75            let queued: Vec<_> = lock(&self.shared.ready).drain(..).collect();
76            if queued.is_empty() {
77                break;
78            }
79            drop(queued);
80        }
81    }
82}
83
84/// A [`ForegroundExecutor`] without its ownership: the cell's checkpoint polls through one,
85/// and every `AsyncApp` carries one to spawn with.
86#[derive(Clone)]
87pub(crate) struct ExecutorHandle {
88    ready: ReadyQueue,
89    dispatcher: Arc<dyn Dispatcher>,
90}
91
92impl ExecutorHandle {
93    /// Queues `future` from outside the `App`: the door for a host callback that arrives while
94    /// the `App` may be borrowed, since the queue is all this touches.
95    pub(crate) fn spawn<R: 'static>(&self, future: impl Future<Output = R> + 'static) -> Task<R> {
96        spawn_on(&self.ready, &self.dispatcher, future)
97    }
98
99    /// Hands `work` to the host to run off the main thread and answers its result as a task:
100    /// dart:isolate's `Isolate.run`.
101    pub(crate) fn run_in_background<T: Send + 'static>(
102        &self,
103        work: impl FnOnce() -> T + Send + 'static,
104    ) -> Task<T> {
105        self.spawn(background::run(&self.dispatcher, work))
106    }
107
108    /// Polls the ready tasks until none is left, including the tasks they wake, and returns
109    /// how many polls ran.
110    ///
111    /// # Panics
112    ///
113    /// After `POLL_BUDGET` polls, on the assumption that two tasks are waking each other in a
114    /// cycle.
115    pub(crate) fn poll_ready_tasks(&self) -> usize {
116        let mut polled = 0usize;
117        loop {
118            let next = lock(&self.ready).pop_front();
119            let Some(runnable) = next else {
120                break;
121            };
122            polled += 1;
123            assert!(
124                polled <= POLL_BUDGET,
125                "tasks did not converge after {POLL_BUDGET} polls; are two tasks waking each \
126                 other in a cycle?"
127            );
128            runnable.run();
129        }
130        polled
131    }
132}
133
134fn spawn_on<R: 'static>(
135    ready: &ReadyQueue,
136    dispatcher: &Arc<dyn Dispatcher>,
137    future: impl Future<Output = R> + 'static,
138) -> Task<R> {
139    let ready = Arc::clone(ready);
140    let host = Arc::clone(dispatcher);
141    // Runs on whichever thread woke the task: the task joins the queue, and the host is
142    // asked to wake the app, in case the main thread is asleep. The ask is not a timer
143    // wakeup and never changes the one a timer is waiting on. One wake covers a burst of
144    // them, so a task woken during a checkpoint costs at most one wake with nothing to poll.
145    let schedule = move |runnable| {
146        lock(&ready).push_back(runnable);
147        host.wake_now();
148    };
149    let (runnable, task) = async_task::spawn_local(future, schedule);
150    runnable.schedule();
151    Task::spawned(task)
152}
153
154fn lock(ready: &ReadyQueue) -> MutexGuard<'_, VecDeque<Runnable>> {
155    ready
156        .lock()
157        .expect("the ready queue is never poisoned: nothing runs under its lock")
158}
159
160/// A Dart `Future` as the framework ports it: a spawned task's result, or a value known now,
161/// and a `Future` itself.
162///
163/// Dart's futures always run to completion; dropping this handle likewise lets the task
164/// finish unobserved, where gpui's `Task` would cancel it.
165pub struct Task<T> {
166    inner: TaskInner<T>,
167}
168
169enum TaskInner<T> {
170    /// A value known now: what an `async` body that never awaits hands back.
171    Ready(Option<T>),
172    Spawned(Option<async_task::Task<T>>),
173}
174
175impl<T> Task<T> {
176    /// A task that resolves with `value` at once: Dart's `Future.value`, for an async callback
177    /// whose body had nothing to await.
178    pub fn ready(value: T) -> Task<T> {
179        Task {
180            inner: TaskInner::Ready(Some(value)),
181        }
182    }
183
184    pub(crate) fn spawned(task: async_task::Task<T>) -> Task<T> {
185        Task {
186            inner: TaskInner::Spawned(Some(task)),
187        }
188    }
189}
190
191// Sound: `poll` never projects a pin into the payload — `Ready` is taken by value and
192// `async_task::Task` is itself `Unpin`.
193impl<T> Unpin for Task<T> {}
194
195impl<T> Future for Task<T> {
196    type Output = T;
197
198    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
199        match &mut self.get_mut().inner {
200            TaskInner::Ready(value) => Poll::Ready(
201                value
202                    .take()
203                    .expect("a Task is polled until it completes, then never again"),
204            ),
205            TaskInner::Spawned(task) => {
206                let task = task
207                    .as_mut()
208                    .expect("a Task is polled until it completes, then never again");
209                Pin::new(task).poll(cx)
210            }
211        }
212    }
213}
214
215impl<T> Drop for Task<T> {
216    fn drop(&mut self) {
217        if let TaskInner::Spawned(task) = &mut self.inner
218            && let Some(task) = task.take()
219        {
220            task.detach();
221        }
222    }
223}
224
225#[cfg(test)]
226mod tests {
227    use std::cell::Cell;
228    use std::future::poll_fn;
229    use std::rc::Rc;
230    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
231    use std::task::Waker;
232
233    use inset_embedder::Instant;
234
235    use super::*;
236
237    /// A host that counts how often it was asked to wake the app, and runs work at once.
238    struct CountingDispatcher(AtomicUsize);
239
240    impl Dispatcher for CountingDispatcher {
241        fn wake_at(&self, _timer_wakeup: Option<Instant>) {}
242
243        fn wake_now(&self) {
244            self.0.fetch_add(1, Ordering::AcqRel);
245        }
246
247        fn dispatch(&self, work: Box<dyn FnOnce() + Send>) {
248            work();
249        }
250    }
251
252    fn executor() -> (ForegroundExecutor, Arc<CountingDispatcher>) {
253        let host = Arc::new(CountingDispatcher(AtomicUsize::new(0)));
254        let executor = ForegroundExecutor::new(Arc::clone(&host) as Arc<dyn Dispatcher>);
255        (executor, host)
256    }
257
258    fn turns_asked(host: &CountingDispatcher) -> usize {
259        host.0.load(Ordering::Acquire)
260    }
261
262    #[test]
263    fn a_spawn_asks_the_host_for_a_turn_and_the_task_runs_in_it() {
264        let (executor, host) = executor();
265        drop(executor.spawn(async {}));
266        assert_eq!(turns_asked(&host), 1);
267        assert_eq!(executor.handle().poll_ready_tasks(), 1);
268    }
269
270    #[test]
271    fn a_task_woken_from_another_thread_asks_for_a_turn_and_runs_in_it() {
272        let (executor, host) = executor();
273        let parked: Arc<Mutex<Option<Waker>>> = Arc::default();
274        let done = Arc::new(AtomicBool::new(false));
275        let ran = Rc::new(Cell::new(false));
276        let (slot, flag, seen) = (Arc::clone(&parked), Arc::clone(&done), Rc::clone(&ran));
277        drop(executor.spawn(async move {
278            poll_fn(|cx| {
279                if flag.load(Ordering::Acquire) {
280                    Poll::Ready(())
281                } else {
282                    *slot.lock().unwrap() = Some(cx.waker().clone());
283                    Poll::Pending
284                }
285            })
286            .await;
287            seen.set(true);
288        }));
289        let handle = executor.handle();
290        handle.poll_ready_tasks();
291        assert_eq!(
292            turns_asked(&host),
293            1,
294            "the spawn asked for a wake, which parked the task"
295        );
296
297        std::thread::spawn(move || {
298            done.store(true, Ordering::Release);
299            let waker = parked.lock().unwrap().take();
300            waker.expect("the wake parked the waker").wake();
301        })
302        .join()
303        .unwrap();
304        assert_eq!(
305            turns_asked(&host),
306            2,
307            "the worker's wake asks the host for another wake"
308        );
309        assert!(!ran.get(), "nothing runs outside the main thread's wake");
310        handle.poll_ready_tasks();
311        assert!(ran.get());
312    }
313}
inset_foundation/executor.rs

inset_foundation/
executor.rs
