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'):