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 <noreply@anthropic.com>
This commit is contained in:
sora 2026-09-15 03:05:04 +00:00
parent 6b336a9e5d
commit ecb29309ef
2 changed files with 33 additions and 7 deletions

View File

@ -319,11 +319,21 @@ async def generate_predictions(
# assemble() tokenizes for the max_input_tokens truncation -- on # assemble() tokenizes for the max_input_tokens truncation -- on
# long-context benches that is SECONDS of CPU per sample (2M-token # long-context benches that is SECONDS of CPU per sample (2M-token
# docs), and running it inline FROZE the whole event loop: heartbeat, # docs). Two failure modes fixed here:
# gate probes and every other request serialized behind one encode. # - inline: froze the whole event loop behind one encode
# Thread it: the loop stays live and encodes parallelize (the Rust # - asyncio.to_thread (32-thread default pool): dozens of concurrent
# fast tokenizer releases the GIL). # tokenizers hogged the GIL and starved the progress renderer +
text = await asyncio.to_thread(assemble, sample) \ # 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 if isinstance(sample.input, str) else None
messages = ([ChatMessage(role='user', content=text)] messages = ([ChatMessage(role='user', content=text)]
if isinstance(sample.input, str) else list(sample.input)) 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() _PROBED_SPECS = set()
_ASSEMBLE_EXEC = None # bounded truncation pool (lazy)
async def _probe_model(adapter, model_spec: str) -> None: async def _probe_model(adapter, model_spec: str) -> None:

View File

@ -19,6 +19,21 @@ def _fmt(sec):
return f'{sec // 60}m{sec % 60:02d}s' if sec >= 60 else f'{sec}s' 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: class RichTerminalProgress:
def __init__(self, console=None): def __init__(self, console=None):
# accept an EXTERNAL console: CLI phase messages and the live bar must # 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='', total=total, completed=min(done, total), new='',
rate=f'{done / elapsed:.2f}', inflight=0, cur='0s', rate=f'{done / elapsed:.2f}', inflight=0, cur='0s',
elapsed=_fmt(elapsed), 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): def set_bench_tag(self, tag: str):
if self.disabled: if self.disabled:
@ -253,7 +268,7 @@ class RichTerminalProgress:
rate=f"{fresh / elapsed:.2f}", rate=f"{fresh / elapsed:.2f}",
inflight=self._inflight_txt(), cur='0s', inflight=self._inflight_txt(), cur='0s',
elapsed=_fmt(elapsed), 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 '-', if fresh and task.total and task.total > completed else '-',
) )