From 6b336a9e5d24e8ca5367d623093a391ddffb64f7 Mon Sep 17 00:00:00 2001
From: sora <2075279110@qq.com>
Date: Tue, 15 Sep 2026 02:56:04 +0000
Subject: [PATCH] Gate probe: dedicated executor + /metrics negative cache

The 5s probe fetched /metrics via asyncio.to_thread, which shares the
DEFAULT executor with second-long truncation tokenizations -- 96 of
those queue-jumped the probe and the state machine never ticked (gate
frozen at 1 while results flowed, ETA 6h). Probes now run on a
dedicated single-thread executor, and after 3 consecutive fetch
failures the gate stops asking for /metrics entirely (this endpoint
404s; pure demand mode from then on).

Verified under a choked default executor: gate ticks 1->2 on schedule
and metrics_dead engages after 3 failures.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 evalharness/model/pool.py | 26 +++++++++++++++++++++++---
 1 file changed, 23 insertions(+), 3 deletions(-)

diff --git a/evalharness/model/pool.py b/evalharness/model/pool.py
index 8fcf5db..cbac67b 100644
--- a/evalharness/model/pool.py
+++ b/evalharness/model/pool.py
@@ -190,6 +190,12 @@ class AdaptiveGate:
         self._prev = (None, None)  # (level, rate) we came from / last-good
         self._bis = (None, None)   # bisect bounds (lo=good, hi=bad)
         self._good_rate = 0.0      # throughput at the good bound (baseline)
+        self._fetch_exec = None    # DEDICATED executor for /metrics probes:
+                                   # the shared default pool is occupied by
+                                   # second-long tokenization jobs, and the
+                                   # probe queued behind them never ticked
+                                   # (gate frozen at 1 while results flowed)
+        self._metrics_dead = False # 3 consecutive fetch failures -> stop asking
         self.stats = {'probe': 0, 'ramp': 0, 'hold_queue': 0, 'backoff_fail': 0,
                       'backoff_queue': 0, 'ramp_demand': 0, 'bisect': 0}
 
@@ -369,6 +375,9 @@ class AdaptiveGate:
             self._wake()
 
     async def _probe_once(self) -> None:
+        if self._metrics_dead:
+            self._no_signal_ramp()   # endpoint said no thrice: stop asking
+            return
         import urllib.request
 
         url = f'{self.adapter.api_base.rstrip("/")}/metrics'
@@ -378,12 +387,23 @@ class AdaptiveGate:
                 return resp.read().decode('utf-8', 'ignore')
 
         try:
-            # thread: the blocking fetch must never stall the event loop
-            # (an unreachable host parks urlopen for the full 4s timeout)
-            text = await asyncio.to_thread(_fetch)
+            # DEDICATED single-thread executor: the shared asyncio pool is
+            # full of second-long truncation tokenizations, and a probe
+            # queued behind them never ran (gate appeared frozen)
+            if self._fetch_exec is None:
+                import concurrent.futures
+
+                self._fetch_exec = concurrent.futures.ThreadPoolExecutor(
+                    max_workers=1, thread_name_prefix='gate-probe')
+            loop = asyncio.get_running_loop()
+            text = await loop.run_in_executor(self._fetch_exec, _fetch)
         except Exception:
+            self._fetch_fails = getattr(self, '_fetch_fails', 0) + 1
+            if self._fetch_fails >= 3:
+                self._metrics_dead = True   # 404/unreachable: pure demand mode
             self._no_signal_ramp()  # no metrics: demand-driven fallback
             return
+        self._fetch_fails = 0
         running = queue = None
         for line in text.splitlines():
             if line.startswith('sglang:num_running_reqs'):