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 <noreply@anthropic.com>
This commit is contained in:
sora
parent
f5c2e4c5af
commit
a4d4592864
@ -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)
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user