gpui: Bound benchmark frame servicing to the tasks queued at frame start (#62235)

The renderer benchmarks drain the main queue to quiescence before each
measured frame, so a task that yields and re-queues itself (an idle
sweep, a poller) runs to completion inside a single measured frame — or
keeps the benchmark spinning forever. Real platforms don't work that
way: macOS's main-queue drain only runs the items present when the drain
starts, Linux runs one runnable per calloop idle callback, and Windows
drains under a 10ms budget before servicing paint messages — frames
always get a chance to preempt. So `run_ready_main_tasks` now only runs
the tasks that were queued when servicing began, letting self-re-queuing
work advance one batch per frame like it would under a real run loop,
and `BenchAppContext` grows a `run_until` (wrapping the dispatcher
primitive task benchmarks already use) so setup code can pump the app
until a condition holds without settling to idle first.

Release Notes:

- N/A

crates/gpui/src/app/bench_context.rs

@@ -415,6 +415,18 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
             .run_until_idle();
     }
 
+    /// Runs main-thread tasks until `ready` returns a value.
+    ///
+    /// Unlike [`Self::run_until_idle`], this returns as soon as `ready`
+    /// reports completion, leaving any remaining queued work pending.
+    pub fn run_until<R>(&self, ready: impl FnMut() -> Option<R>) -> R {
+        self.background_executor
+            .dispatcher()
+            .as_threaded()
+            .expect("validated in BenchAppContext::build")
+            .run_until(ready)
+    }
+
     /// Measures a generic benchmark workload using Criterion's iteration loop.
     ///
     /// The closure is invoked once per Criterion iteration with this
@@ -527,6 +539,8 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
         let collector = FrameTraceScope::start();
 
         let mut benchmark = || {
+            // Work already queued at frame start delays the frame in
+            // production too, so run it inside the measured interval.
             dispatcher
                 .as_threaded()
                 .expect("validated in BenchAppContext::build")

crates/gpui/src/platform/threaded_dispatcher.rs

@@ -312,14 +312,32 @@ impl ThreadedDispatcher {
         }
     }
 
-    /// Runs all main-thread tasks that are queued right now, without waiting for
-    /// background work or timers to finish.
+    /// Runs the main-thread tasks that were queued when the call began,
+    /// returning whether any ran. Tasks dispatched while running (e.g. a task
+    /// re-queuing itself after yielding) are left for the next call, as on
+    /// the platform run loops.
     pub fn run_ready_main_tasks(&self) -> bool {
         assert!(
             self.is_main_thread(),
             "run_ready_main_tasks must be called on the threaded dispatcher's main thread"
         );
-        self.drain_main_queue()
+        let pending = self.main_receiver.lock().len();
+        let mut ran_any = false;
+        for _ in 0..pending {
+            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();
+                    ran_any = true;
+                }
+                Ok(None) | Err(_) => break,
+            }
+        }
+        ran_any
     }
 
     /// Cancels all pending timers so timers armed by one workload can't fire
@@ -591,6 +609,32 @@ mod tests {
         })
     }
 
+    #[test]
+    fn run_ready_main_tasks_advances_requeuing_work_one_batch_per_call() {
+        const REQUEUE_LIMIT: usize = 10_000;
+
+        let dispatcher = Arc::new(ThreadedDispatcher::new());
+        let foreground = ForegroundExecutor::new(dispatcher.clone());
+
+        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();
+
+        assert!(dispatcher.run_ready_main_tasks());
+        assert_eq!(iterations.load(Ordering::SeqCst), 1);
+        assert!(dispatcher.run_ready_main_tasks());
+        assert_eq!(iterations.load(Ordering::SeqCst), 2);
+    }
+
     #[test]
     fn timers_fire_in_real_time() {
         let dispatcher = Arc::new(ThreadedDispatcher::new());

crates/gpui/src/queue.rs

@@ -225,6 +225,12 @@ impl<T> PriorityQueueReceiver<T> {
         self.state.queues.lock().is_empty()
     }
 
+    /// Returns the number of queued elements across all priorities.
+    pub(crate) fn len(&self) -> usize {
+        let queues = self.state.queues.lock();
+        queues.high_priority.len() + queues.medium_priority.len() + queues.low_priority.len()
+    }
+
     /// Tries to pop one element from the priority queue without blocking.
     ///
     /// This will early return if there are no elements in the queue.