From 336fa8b38e4af4e62072fe4c1a6f49063c2aa620 Mon Sep 17 00:00:00 2001 From: Anthony Eid <56899983+Anthony-Eid@users.noreply.github.com> Date: Wed, 5 Aug 2026 10:23:02 -0600 Subject: [PATCH] gpui: Check run_until readiness between main-thread runnables (#62220) `ThreadedDispatcher::run_until` - the completion mechanism under `BenchAppContext` async task benchmarks (`bench_task` / `bench_batched_task`, added in #62180) - drained the entire main queue before checking its readiness predicate. Main-thread work that re-queues itself (an idle-time sweep, a polling loop) keeps the queue non-empty until it finishes every iteration, so a task benchmark's measured interval silently extended past the awaited task's completion until all such deferred work settled. Benchmarks of UI workloads that schedule follow-up idle work could report several times their true completion latency. This changes `run_until` to step main-thread runnables one at a time and check readiness before each, so it returns at the completion it awaits rather than at queue quiescence, making async task benchmarks more accurate. The regression test spawns a task that re-queues itself 10,000 times plus a one-shot completion task, and asserts `run_until` returns without draining the re-queued work; it fails against the previous drain-first implementation. Release Notes: - N/A --- .../gpui/src/platform/threaded_dispatcher.rs | 91 ++++++++++++++++++- 1 file changed, 89 insertions(+), 2 deletions(-) diff --git a/crates/gpui/src/platform/threaded_dispatcher.rs b/crates/gpui/src/platform/threaded_dispatcher.rs index db11f3542f..d0e2476c18 100644 --- a/crates/gpui/src/platform/threaded_dispatcher.rs +++ b/crates/gpui/src/platform/threaded_dispatcher.rs @@ -262,6 +262,12 @@ impl ThreadedDispatcher { /// Unlike [`Self::run_until_idle`], this waits across temporary quiescence. /// This is required when completion can arrive from an external worker that /// is not represented in the dispatcher's in-flight count. + /// + /// Readiness is checked before every main-thread runnable, so this returns + /// as soon as `ready` observes completion rather than after the queue + /// drains — deferred work that re-queues itself (idle sweeps, pollers) + /// must not extend a benchmark's measured interval past the completion it + /// awaits. #[cfg(any(test, feature = "bench"))] pub(crate) fn run_until(&self, mut ready: impl FnMut() -> Option) -> R { assert!( @@ -269,10 +275,12 @@ impl ThreadedDispatcher { "run_until must be called on the threaded dispatcher's main thread" ); loop { - self.drain_main_queue(); if let Some(result) = ready() { return result; } + if self.run_one_main_task() { + continue; + } let mut inflight = self.idle.inflight.lock(); if self.main_queue_has_work() { @@ -282,6 +290,28 @@ impl ThreadedDispatcher { } } + /// Runs at most one queued main-thread task, returning whether one ran. + /// + /// [`Self::run_until`] steps tasks one at a time so it can observe + /// readiness between them: a task that perpetually re-queues itself (like + /// an idle-time sweep) would otherwise keep [`Self::drain_main_queue`] + /// looping past the completion the caller is waiting for. + #[cfg(any(test, feature = "bench"))] + fn run_one_main_task(&self) -> bool { + let runnable = self.main_receiver.lock().try_pop(); + match runnable { + Ok(Some(runnable)) => { + let location = runnable.metadata().location; + let spawned = runnable.metadata().spawned; + profiler::update_running_task(spawned, location); + runnable.run(); + profiler::save_task_timing(); + true + } + Ok(None) | Err(_) => false, + } + } + /// Runs all main-thread tasks that are queued right now, without waiting for /// background work or timers to finish. pub fn run_ready_main_tasks(&self) -> bool { @@ -408,7 +438,8 @@ impl PlatformDispatcher for ThreadedDispatcher { #[cfg(test)] mod tests { - use std::sync::atomic::{AtomicBool, Ordering}; + use std::future::Future; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use super::*; use crate::{BackgroundExecutor, ForegroundExecutor}; @@ -504,6 +535,62 @@ mod tests { assert!(completed.load(Ordering::SeqCst)); } + #[test] + fn run_until_returns_at_readiness_despite_requeuing_main_work() { + const REQUEUE_LIMIT: usize = 10_000; + + let dispatcher = Arc::new(ThreadedDispatcher::new()); + let foreground = ForegroundExecutor::new(dispatcher.clone()); + + // Mirrors main-thread work that yields and immediately re-queues + // itself (e.g. an idle-time sweep): the main queue never drains until + // such work finishes every iteration, so readiness must be observed + // between runnables rather than only at quiescence. + let iterations = Arc::new(AtomicUsize::new(0)); + foreground + .spawn({ + let iterations = iterations.clone(); + async move { + for _ in 0..REQUEUE_LIMIT { + iterations.fetch_add(1, Ordering::SeqCst); + yield_once().await; + } + } + }) + .detach(); + + let completed = Arc::new(AtomicBool::new(false)); + foreground + .spawn({ + let completed = completed.clone(); + async move { + completed.store(true, Ordering::SeqCst); + } + }) + .detach(); + + dispatcher.run_until(|| completed.load(Ordering::SeqCst).then_some(())); + assert!( + iterations.load(Ordering::SeqCst) < REQUEUE_LIMIT, + "run_until should return at readiness instead of draining re-queued main work" + ); + } + + /// Completes after one re-schedule: the poll returns `Pending` and wakes + /// immediately, so the runnable re-enters the main queue. + fn yield_once() -> impl Future { + let mut yielded = false; + std::future::poll_fn(move |poll_context| { + if yielded { + std::task::Poll::Ready(()) + } else { + yielded = true; + poll_context.waker().wake_by_ref(); + std::task::Poll::Pending + } + }) + } + #[test] fn timers_fire_in_real_time() { let dispatcher = Arc::new(ThreadedDispatcher::new());