gpui: Settle benchmark app state between task iterations (#62587)

Dropped entities are released only inside an update's effect flush, and
releases cascade: one flush drops the entities whose handles are gone,
their drops release further handles and can queue foreground work, and a
later flush collects those. `BenchAppContext` callers that only pump the
executor between iterations therefore saw torn-down state linger in the
entity map until some woken task happened to run an update — in a
downstream benchmark this looked like a per-iteration leak of the whole
app graph (~35 MB per iteration), releasing on an apparently timer-bound
schedule.

This adds `BenchAppContext::settle`, which alternates draining queued
work with GPUI update cycles until the dispatcher reports idle,
mirroring the update cadence production gets for free from frames and
input events. `bench_batched_task` now settles before each iteration's
setup (outside the timed interval), so the previous iteration's state is
fully released and cannot accumulate across a measurement. A new
`ThreadedDispatcher::is_idle` predicate backs the loop's termination and
is covered by a unit test.

Release Notes:

- N/A

crates/gpui/src/app/bench_context.rs

@@ -415,6 +415,30 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
             .run_until_idle();
     }
 
+    /// Alternates draining queued work with GPUI update cycles until neither
+    /// makes progress, so state dropped by benchmark code is fully released.
+    ///
+    /// Dropped entities are released only inside an update's effect flush, and
+    /// releases cascade: one flush drops the entities whose handles are gone,
+    /// their drops release further handles and can queue foreground work, and
+    /// a later flush collects those. Executor pumping alone never runs a
+    /// flush, so without this dropped state would linger in the entity map
+    /// until some woken task happened to run an update. Production gets this
+    /// cadence for free from frames and input events.
+    pub fn settle(&mut self) {
+        let dispatcher = self.background_executor.dispatcher().clone();
+        let dispatcher = dispatcher
+            .as_threaded()
+            .expect("validated in BenchAppContext::build");
+        loop {
+            self.run_until_idle();
+            self.update(|_| ());
+            if dispatcher.is_idle() {
+                return;
+            }
+        }
+    }
+
     /// Runs main-thread tasks until `ready` returns a value.
     ///
     /// Unlike [`Self::run_until_idle`], this returns as soon as `ready`
@@ -493,9 +517,16 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
         let report = self.report.clone();
 
         bencher.iter_batched_ref(
-            || MeasuredTaskInput {
-                input: setup(&mut setup_context),
-                frame_trace_scope: Some(FrameTraceScope::start()),
+            || {
+                // The previous iteration's input and output were just
+                // dropped; settling here releases their entities before the
+                // next setup, so per-iteration state cannot accumulate
+                // across a measurement.
+                setup_context.settle();
+                MeasuredTaskInput {
+                    input: setup(&mut setup_context),
+                    frame_trace_scope: Some(FrameTraceScope::start()),
+                }
             },
             |measured_input| {
                 let task = benchmark(&mut measured_input.input, &mut benchmark_context);

crates/gpui/src/platform/threaded_dispatcher.rs

@@ -370,6 +370,14 @@ impl ThreadedDispatcher {
         )
     }
 
+    /// Whether no main-thread work is queued, no background or timer
+    /// runnables are queued or running, and no armed timer is due. Timers
+    /// that aren't due yet are ignored, as in [`Self::run_until_idle`].
+    #[cfg(any(test, feature = "bench"))]
+    pub(crate) fn is_idle(&self) -> bool {
+        !self.main_queue_has_work() && !self.has_due_timer() && *self.idle.inflight.lock() == 0
+    }
+
     fn has_due_timer(&self) -> bool {
         let state = self.timers.state.lock();
         state
@@ -462,6 +470,32 @@ mod tests {
     use super::*;
     use crate::{BackgroundExecutor, ForegroundExecutor};
 
+    #[test]
+    fn is_idle_tracks_queued_work_but_ignores_undue_timers() {
+        let dispatcher = Arc::new(ThreadedDispatcher::new());
+        let foreground = ForegroundExecutor::new(dispatcher.clone());
+        assert!(dispatcher.is_idle());
+
+        foreground.spawn(async {}).detach();
+        assert!(!dispatcher.is_idle());
+        dispatcher.run_until_idle();
+        assert!(dispatcher.is_idle());
+
+        let background = BackgroundExecutor::new(dispatcher.clone());
+        let timer = background.timer(Duration::from_secs(60));
+        // The timer future's initial poll runs on a worker thread; wait for
+        // it so only the armed, not-yet-due timer remains.
+        dispatcher.run_until_idle();
+        assert!(
+            dispatcher.is_idle(),
+            "a timer that is not due yet should not count as pending work"
+        );
+        drop(timer);
+        dispatcher.cancel_pending_timers();
+        dispatcher.run_until_idle();
+        assert!(dispatcher.is_idle());
+    }
+
     #[test]
     fn run_ready_main_tasks_does_not_wait_for_background_handoffs() {
         let dispatcher = Arc::new(ThreadedDispatcher::new());