From ecb29309efa74e699a6772838630bfb1a37dea40 Mon Sep 17 00:00:00 2001 From: sora <2075279110@qq.com> Date: Tue, 15 Sep 2026 03:05:04 +0000 Subject: [PATCH] Bounded truncation pool (8 threads); human eta up to days 95 waiters all tokenizing 2M-token docs through the 32-thread default executor saturated the GIL: the rich render thread and the event loop starved, so the bar froze and jumped (and the gate probe went blind). Truncation now runs on a dedicated 8-thread pool; the remaining workers queue and the loop/renderer stay responsive. eta formats as 45s / 7m15s / 2h35m40s / 6d03h12m as it grows. Co-Authored-By: Claude --- evalharness/model/runner.py | 21 ++++++++++++++++----- evalharness/progress/rich_terminal.py | 19 +++++++++++++++++-- 2 files changed, 33 insertions(+), 7 deletions(-) diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py index c81df9f..bdbc4b8 100644 --- a/evalharness/model/runner.py +++ b/evalharness/model/runner.py @@ -319,11 +319,21 @@ async def generate_predictions( # assemble() tokenizes for the max_input_tokens truncation -- on # long-context benches that is SECONDS of CPU per sample (2M-token - # docs), and running it inline FROZE the whole event loop: heartbeat, - # gate probes and every other request serialized behind one encode. - # Thread it: the loop stays live and encodes parallelize (the Rust - # fast tokenizer releases the GIL). - text = await asyncio.to_thread(assemble, sample) \ + # docs). Two failure modes fixed here: + # - inline: froze the whole event loop behind one encode + # - asyncio.to_thread (32-thread default pool): dozens of concurrent + # tokenizers hogged the GIL and starved the progress renderer + + # loop itself (bar froze, then jumped) + # A DEDICATED BOUNDED pool: 8 encodes at a time, remaining workers + # queue -- GIL pressure capped, everything stays responsive. + global _ASSEMBLE_EXEC + if _ASSEMBLE_EXEC is None: + import concurrent.futures + + _ASSEMBLE_EXEC = concurrent.futures.ThreadPoolExecutor( + max_workers=8, thread_name_prefix='assemble') + text = await asyncio.get_running_loop().run_in_executor( + _ASSEMBLE_EXEC, assemble, sample) \ if isinstance(sample.input, str) else None messages = ([ChatMessage(role='user', content=text)] if isinstance(sample.input, str) else list(sample.input)) @@ -540,6 +550,7 @@ def _progress(progress: bool, done: int, total: int, t0: float, usage: Usage) -> _PROBED_SPECS = set() +_ASSEMBLE_EXEC = None # bounded truncation pool (lazy) async def _probe_model(adapter, model_spec: str) -> None: diff --git a/evalharness/progress/rich_terminal.py b/evalharness/progress/rich_terminal.py index 05ce268..174ddb9 100644 --- a/evalharness/progress/rich_terminal.py +++ b/evalharness/progress/rich_terminal.py @@ -19,6 +19,21 @@ def _fmt(sec): return f'{sec // 60}m{sec % 60:02d}s' if sec >= 60 else f'{sec}s' +def _fmt_eta(sec): + """eta: s -> m s -> h m s -> d h m (grows with the unit that matters).""" + sec = int(sec) + if sec < 60: + return f'{sec}s' + m, s = divmod(sec, 60) + if m < 60: + return f'{m}m{s:02d}s' + h, m = divmod(m, 60) + if h < 24: + return f'{h}h{m:02d}m{s:02d}s' + d, h = divmod(h, 24) + return f'{d}d{h:02d}h{m:02d}m' + + class RichTerminalProgress: def __init__(self, console=None): # accept an EXTERNAL console: CLI phase messages and the live bar must @@ -168,7 +183,7 @@ class RichTerminalProgress: total=total, completed=min(done, total), new='', rate=f'{done / elapsed:.2f}', inflight=0, cur='0s', elapsed=_fmt(elapsed), - eta=_fmt((total - done) * elapsed / done) if done and total > done else '-') + eta=_fmt_eta((total - done) * elapsed / done) if done and total > done else '-') def set_bench_tag(self, tag: str): if self.disabled: @@ -253,7 +268,7 @@ class RichTerminalProgress: rate=f"{fresh / elapsed:.2f}", inflight=self._inflight_txt(), cur='0s', elapsed=_fmt(elapsed), - eta=_fmt((task.total - completed) * elapsed / fresh) + eta=_fmt_eta((task.total - completed) * elapsed / fresh) if fresh and task.total and task.total > completed else '-', )