From a4d45928645543aedb4e4069fe64889eeaa6872b Mon Sep 17 00:00:00 2001 From: sora <2075279110@qq.com> Date: Mon, 14 Sep 2026 03:14:08 +0000 Subject: [PATCH] Scoring-phase progress; pool: fix double-release + cross-loop reuse Scoring progress: docker-exec benches (humaneval etc.) score for minutes with zero feedback -- the bar sat at 'generating 100%' and looked hung. evaluate() now takes on_scored(i, n) (atomic counter, fires from worker threads), run_eval passes it through, and the CLI shows 'scoring 42/164' on the bar + milestone log lines every 10% (also fixes the phase match: 'scoring' never matched the capitalized 'Scoring predictions...' status message, so the bar never even switched its label). PooledAdapter: - one release per acquire: the exception path released True (inner finally) AND False (except handler), double-decrementing _inflight (over-admission) and applying the x0.7 backoff twice - AdaptiveGate: rebuild the Condition + probe task when the event loop changes -- pools are cached across benchmarks and the CLI runs asyncio.run() per bench/repeat; a loop-bound Condition from a closed loop raises 'bound to a different event loop' under contention Co-Authored-By: Claude --- evalharness/cli.py | 20 +++++++++++++++++--- evalharness/eval/runner.py | 21 +++++++++++++++++++-- evalharness/model/pool.py | 28 +++++++++++++++++++++++----- evalharness/model/runner.py | 2 ++ 4 files changed, 61 insertions(+), 10 deletions(-) diff --git a/evalharness/cli.py b/evalharness/cli.py index 346e7eb..1024fab 100644 --- a/evalharness/cli.py +++ b/evalharness/cli.py @@ -704,14 +704,27 @@ def _cmd_eval_run(args) -> int: _tag = f'[{_idx}/{total_runs}] ' if total_runs > 1 else '' if _reporter is not None: _reporter.log(f'{_tag}{_name}: {_narration(msg)}') - if 'scoring' in msg: + _m = msg.lower() + if 'scoring' in _m: _reporter.set_phase('scoring') - elif 'generating model responses' in msg: + elif 'generating model responses' in _m: _reporter.set_phase('generating') - elif 'writing' in msg: + elif 'writing' in _m: _reporter.set_phase('writing') else: _print_phase(_console, _idx, total_runs, _name, msg) + + def on_scored(done, total_s, _reporter=progress_reporter, + _cb=status_callback): + """Per-sample scoring progress: docker-exec benches score + for minutes with zero feedback otherwise (bar sits at + 'generating 100%' and looks hung).""" + if _reporter is not None: + _reporter.set_phase(f'scoring {done}/{total_s}') + # milestone lines: pipes/logs without a live bar see movement + if _cb and total_s and (done % max(1, total_s // 10) == 0 + or done == total_s): + _cb(f'Scoring {done}/{total_s} samples') _gen_kw = {**bench_cfg, **(getattr(args, '_gen_override', {}) or {})} # max_input_tokens must be a SEPARATE run_eval param (it drives # truncation in assemble(), not a gen_kwarg the adapter sees) -- @@ -733,6 +746,7 @@ def _cmd_eval_run(args) -> int: gen_profile=getattr(args, 'profile', ''), progress_reporter=progress_reporter, status_callback=status_callback, + on_scored=on_scored, repeat=_rep + 1)) _m = next((v for k, v in report.metrics.items() if k != 'extraction_failure_rate'), None) diff --git a/evalharness/eval/runner.py b/evalharness/eval/runner.py index f250551..fe7e871 100644 --- a/evalharness/eval/runner.py +++ b/evalharness/eval/runner.py @@ -27,6 +27,7 @@ def evaluate( model: str = '', judge: Optional[Callable] = None, extra_metadata: Optional[Dict] = None, + on_scored: Optional[Callable[[int, int], None]] = None, ) -> EvalReport: """Score a dataset against raw predictions. @@ -115,6 +116,22 @@ def evaluate( return result workers = getattr(recipe, 'exec_workers', 1) + n_total = len(samples) + import itertools + + _scored = itertools.count(1) # next() is atomic: safe from worker threads + + def _counted(sample, pred): + # docker/subprocess scoring is minute-scale per sample; surface + # per-sample progress or the run looks frozen at 'generating 100%' + r = judge_one(sample, pred) + if on_scored is not None: + try: + on_scored(next(_scored), n_total) + except Exception: + pass + return r + if workers > 1 and len(samples) > 1: # parallel judging: docker/subprocess execution is embarrassingly # parallel (one container per sample); text scorers are cheap and @@ -122,10 +139,10 @@ def evaluate( import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool: - results = list(pool.map(judge_one, samples, predictions)) + results = list(pool.map(_counted, samples, predictions)) else: for sample, pred in zip(samples, predictions): - results.append(judge_one(sample, pred)) + results.append(_counted(sample, pred)) report = EvalReport( dataset=ds_name, diff --git a/evalharness/model/pool.py b/evalharness/model/pool.py index 09af0c8..1a5f6d5 100644 --- a/evalharness/model/pool.py +++ b/evalharness/model/pool.py @@ -104,10 +104,16 @@ class PooledAdapter(ModelAdapter): try: i = self.adapters.index(adapter) await self._gates[i].acquire() + ok = False try: out = await adapter.generate(messages, tools=tools, **kw) + ok = True finally: - self._gates[i].release(True) + # ONE release per acquire: the old code released True in + # this finally AND False in the except handler, double- + # decrementing _inflight (gate over-admits) and applying + # the x0.7 backoff twice per failure + self._gates[i].release(ok) self.usage = self.usage + out.usage self.stats['ok'] += 1 self._mark(adapter, True) @@ -117,8 +123,6 @@ class PooledAdapter(ModelAdapter): except Exception as e: # dead/overloaded instance -> next last_exc = e self._mark(adapter, False) - with contextlib.suppress(ValueError): - self._gates[self.adapters.index(adapter)].release(False) # 4xx (e.g. 400 overloaded) still worth trying ANOTHER instance: # one backend's state can differ from the rest continue @@ -158,14 +162,28 @@ class AdaptiveGate: self._cond: Optional[asyncio.Condition] = None self._task: Optional[asyncio.Task] = None self._stopped = False + self._loop = None # loop the cond/probe-task are bound to self.stats = {'probe': 0, 'ramp': 0, 'hold_queue': 0, 'backoff_fail': 0, 'backoff_queue': 0} # ---- gate semantics ---- async def acquire(self) -> None: - if self._cond is None: # lazy init in the running loop + loop = asyncio.get_running_loop() + if self._cond is None or self._loop is not loop or self._stopped: + # lazy init OR LOOP CHANGE: pools are cached across benchmarks, + # and the CLI runs asyncio.run() per bench (per repeat!) -- a new + # run means a new event loop while this gate survives. A Condition + # is loop-bound: reusing the old one raises "bound to a different + # event loop" under contention, and the old probe task died with + # the closed loop. Rebuild both; inflight resets to 0 (nothing is + # in flight on a fresh loop by construction). + self._loop = loop self._cond = asyncio.Condition() - self._task = asyncio.get_event_loop().create_task(self._probe_loop()) + self._inflight = 0 + self._stopped = False + if self._task is not None: + self._task.cancel() # dead task from the closed loop; no-op + self._task = loop.create_task(self._probe_loop()) while self._inflight >= max(1, int(self.limit)): await self._cond.acquire() try: diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py index fe46c7e..1288fd4 100644 --- a/evalharness/model/runner.py +++ b/evalharness/model/runner.py @@ -629,6 +629,7 @@ async def run_eval( prompt_style: str = 'strict_letter', gen_profile: str = '', repeat: int = 1, + on_scored=None, ) -> EvalReport: """Generate + score in one call. Model spec examples: 'mock', 'mock:boxed', 'openai/http://gpu03:8000/v1?qwen3-8b', 'deploy:vllm/qwen3-8b'. @@ -780,6 +781,7 @@ async def run_eval( extra_metadata={'gen_input_tokens': usage.input_tokens, 'gen_output_tokens': usage.output_tokens, 'gen_total_tokens': usage.total_tokens}, + on_scored=on_scored, ) report.model = model_spec report.dataset = name