diff --git a/evalharness/agent/envs/swe_agentic.py b/evalharness/agent/envs/swe_agentic.py index ce4bbc3..ffff35b 100644 --- a/evalharness/agent/envs/swe_agentic.py +++ b/evalharness/agent/envs/swe_agentic.py @@ -24,6 +24,66 @@ from ..loop import Environment, register_env SENTINEL = 'COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT' + +class _SessionGate: + """Session-level admission for multi-turn agent benches (CONCUR-style). + + Turn-level fairness lets EVERY started session crawl at 1/N bandwidth + -- fine for single-turn benches, terrible for 100-turn agents: first + completions arrive absurdly late and all contexts stay hot. Here a + session HOLDS its slot for its whole life (execution continuity: an + in-progress session always outranks a new one), and the number of + admitted sessions follows the model gate's discovered limit. + + No preemption: a shrink only blocks NEW sessions; running ones drain + naturally. + """ + + def __init__(self): + self.active = 0 + self._cond = None + self._loop = None + + def _limit(self, adapter) -> int: + gates = getattr(adapter, '_gates', None) + if gates: + try: + return max(1, int(gates[0].limit)) + except Exception: + pass + return int(__import__('os').environ.get( + 'EVALHARNESS_SESSION_SLOTS', '8') or 8) + + async def enter(self, adapter): + import asyncio + + loop = asyncio.get_running_loop() + if self._cond is None or self._loop is not loop: + self._loop = loop + self._cond = asyncio.Condition() + self.active = 0 # fresh loop: nothing in flight + while self.active >= self._limit(adapter): + await self._cond.acquire() + try: + await self._cond.wait() + finally: + self._cond.release() + self.active += 1 + + async def exit(self): + import asyncio + + self.active = max(0, self.active - 1) + if self._cond is not None: + try: + async with self._cond: + self._cond.notify_all() + except RuntimeError: + pass + + +_SESSION_GATE = _SessionGate() + # mini-swe-agent swebench.yaml contract (verbatim sections that matter) INSTANCE_TEMPLATE = """ Consider the following PR description: @@ -144,12 +204,36 @@ class SWEAgenticEnvironment(Environment): image, 'tail', '-f', '/dev/null'], timeout=120) if r.returncode != 0: raise RuntimeError(f'start container failed: {r.stderr[:200]}') + # network hardening: github.com is reachable from these containers + # but large transfers stall mid-stream (GFW throttling) -- an agent + # curl/git then hangs until the 600s exec cap. Make curl and git + # abort stalled transfers themselves so the agent sees a FAST + # failure and reroutes instead of burning 10 minutes: + # curl: ~/.curlrc applies to every agent-invoked curl + # git: lowSpeed abort on <1KB/s sustained 30s + _docker(['exec', name, 'bash', '-lc', + 'printf "connect-timeout 15\\nspeed-time 30\\nspeed-limit 1024\\n" > /root/.curlrc 2>/dev/null; ' + 'git config --global http.lowSpeedLimit 1024 2>/dev/null; ' + 'git config --global http.lowSpeedTime 30 2>/dev/null'], timeout=30) return name def _exec(self, cmd: str, timeout: int = 600) -> Dict[str, Any]: # bash -lc: swebench images activate the per-instance testbed via # shell startup files (es parity: _SWE_BENCH_INTERPRETER) - r = _docker(['exec', self.container, 'bash', '-lc', cmd], timeout=timeout) + try: + r = _docker(['exec', self.container, 'bash', '-lc', cmd], + timeout=timeout) + except subprocess.TimeoutExpired: + # a hung command (e.g. curl to an unreachable github from a CN + # container) must be an OBSERVATION the agent can react to -- + # letting it raise killed the whole task 600s in, mid-loop, + # after all its previous turns were already spent + return {'exit': 124, + 'out': f'COMMAND TIMED OUT after {timeout}s (killed). ' + 'The command may be blocked on the network -- ' + 'github.com is often unreachable from this ' + 'container; try a mirror (ghproxy) or proceed ' + 'without the download.'} out = (r.stdout or '') + (r.stderr or '') # cap observation: agents choke on 100k-char dumps if len(out) > 30000: @@ -165,16 +249,38 @@ class SWEAgenticEnvironment(Environment): async def run_task(self, adapter, sample: Sample, max_turns: int = 250, system: str = '', user_adapter=None, gen_kwargs=None, **kw) -> Dict[str, Any]: + # SESSION admission BEFORE the container: hold arriving sessions at + # the door instead of letting all of them start containers and then + # crawl at 1/N turn bandwidth + await _SESSION_GATE.enter(adapter) + try: + return await self._run_session(adapter, sample, max_turns, + system, user_adapter, gen_kwargs) + finally: + await _SESSION_GATE.exit() + + async def _run_session(self, adapter, sample: Sample, max_turns: int = 250, + system: str = '', user_adapter=None, + gen_kwargs=None, **kw) -> Dict[str, Any]: image = (sample.sandbox and sample.sandbox.image) or '' if not image: raise RuntimeError('swe_agentic: sample has no sandbox image ' '(dataset plugin must declare it)') # register on ARRIVAL: the background pool starts pulling while - # earlier samples are still generating + # earlier samples are still generating. OFF the loop: register runs + # `docker images` (2-4s against a 534-image daemon) and 100 tasks + # doing that serially ON the loop froze the whole runner at startup from ...sandbox.image_service import get_image_service - get_image_service().register([image]) - self.container = self._start(image) + svc = get_image_service() + await asyncio.to_thread(svc.register, [image]) + # NEVER wait on the image barrier on the event loop: one blocked + # wait_ready froze the WHOLE runner (0.00/s, every other task + # stalled behind it). Park the wait (and the docker run) in a + # thread; _start's own wait then returns instantly (already ready) + if not await asyncio.to_thread(svc.wait_ready, image, 3600): + raise RuntimeError(f'image {image} unavailable after all sources') + self.container = await asyncio.to_thread(self._start, image) ps = (sample.metadata or {}).get('problem_statement') or sample.input_text messages = [ChatMessage(role='user', content=INSTANCE_TEMPLATE.format( problem_statement=ps, sentinel=SENTINEL))] @@ -232,7 +338,7 @@ class SWEAgenticEnvironment(Environment): self._exec, 'cd /testbed && git diff') patch = res['out'].strip() finally: - self._stop() + await asyncio.to_thread(self._stop) md = sample.metadata or {} return { diff --git a/evalharness/cli.py b/evalharness/cli.py index 9ea94d2..26967c9 100644 --- a/evalharness/cli.py +++ b/evalharness/cli.py @@ -751,6 +751,25 @@ def _cmd_eval_run(args) -> int: 'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s')] + [cats]) + # the endpoints this run drives are DIRECT infrastructure, but httpx + # honors http_proxy/https_proxy by default (trust_env) -- a personal + # proxy in the launching shell silently routed every eval request + # through it, and a sick proxy read as "endpoint dead" (fast-fail + # cascades killed whole runs while the endpoint was perfectly fine). + # Exclude our own endpoints from proxying, whatever client is used. + import urllib.parse as _up + + _np = {x.strip() for x in + os.environ.get('no_proxy', os.environ.get('NO_PROXY', '')).split(',') + if x.strip()} + for _u in filter(None, (getattr(args, 'api_url', ''), + getattr(args, 'judge_api_url', ''))): + _h = _up.urlparse(_u if '//' in _u else f'//{_u}').hostname + if _h: + _np.add(_h) + if _np: + os.environ['no_proxy'] = os.environ['NO_PROXY'] = ','.join(sorted(_np)) + run_started = _time.time() total_runs = len(args.datasets) model_spec = _compose_model_spec(args) diff --git a/evalharness/config/default.yaml b/evalharness/config/default.yaml index f7a1d04..afe315f 100644 --- a/evalharness/config/default.yaml +++ b/evalharness/config/default.yaml @@ -71,4 +71,4 @@ swe_bench_verified_agentic: max_tokens: 4096 # 单轮 bash 命令生成预算（mini-swe-agent 口径） temperature: 0.0 env: swe_agentic # 多轮 agent：bash 探索 /testbed + sentinel 提交 - max_turns: 250 # mini-swe-agent 默认步数 + max_turns: 100 # mini-swe-agent 默认步数 diff --git a/evalharness/config/thinking.yaml b/evalharness/config/thinking.yaml index 76d958b..f175d32 100644 --- a/evalharness/config/thinking.yaml +++ b/evalharness/config/thinking.yaml @@ -60,7 +60,7 @@ tau2_bench: max_tokens: 16384 swe_bench_verified_agentic: - max_tokens: 4096 + max_tokens: 8196 temperature: 0.0 env: swe_agentic max_turns: 250 diff --git a/evalharness/eval/recipes/agent.py b/evalharness/eval/recipes/agent.py index 8c657a4..90024e6 100644 --- a/evalharness/eval/recipes/agent.py +++ b/evalharness/eval/recipes/agent.py @@ -445,7 +445,9 @@ def swe_bench_verified_agentic(): extract='identity', scorers={'resolved': {'name': 'env_reward', 'backend': _swe_official_reward}}, - exec_workers=2, + # official swebench eval spins ONE container per instance (~5 min + # each); 2 workers stretched 500 instances to ~20h of scoring tail + exec_workers=4, description='SWE-bench Verified AGENTIC (mini-swe-agent protocol): ' 'multi-turn bash agent explores /testbed in the ' 'per-instance container, submits a git diff; scored by ' diff --git a/evalharness/model/pool.py b/evalharness/model/pool.py index 5d0ddf4..b4fadb8 100644 --- a/evalharness/model/pool.py +++ b/evalharness/model/pool.py @@ -14,8 +14,9 @@ expires, so one sick endpoint cannot absorb its share of the load. """ import asyncio -import contextlib import itertools +import os +import statistics import time from typing import Dict, List, Optional @@ -103,7 +104,12 @@ class PooledAdapter(ModelAdapter): adapter = self._next() try: i = self.adapters.index(adapter) - await self._gates[i].acquire() + stamp = await self._gates[i].acquire() + # NB: service time ONLY -- starting the clock before acquire + # bills the gate's own QUEUE wait as service slowness, and + # the shrinking medians convinced the gate the endpoint + # was degrading under load it was not under + t0 = time.monotonic() ok = False try: out = await adapter.generate(messages, tools=tools, **kw) @@ -113,7 +119,9 @@ class PooledAdapter(ModelAdapter): # 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._gates[i].release( + ok, stamp=stamp, dur_s=time.monotonic() - t0, + tokens=(out.usage.output_tokens if ok else 0)) self.usage = self.usage + out.usage self.stats['ok'] += 1 self._mark(adapter, True) @@ -137,73 +145,57 @@ class PooledAdapter(ModelAdapter): class AdaptiveGate: - """Per-endpoint adaptive concurrency limiter (AIMD + server signals). + """Per-endpoint adaptive concurrency limiter (Netflix Gradient2 style). - Goal: keep the backend SATURATED (high XPU util / throughput) without - pushing it over the cliff (500s / child crashes). Signals: - - server /metrics: num_queue_reqs > 0 means WE are pushing too hard - for the current mix; idle (no queue, low running) means room to grow - - request failures: multiplicative decrease (survive first) - Control law (classic AIMD): - +1 concurrency per probe interval when the endpoint looks underfed - -1 when the server reports a queue (gentle) - x0.7 on any failed request (fast backoff), floor at LO - Purely additive to PooledAdapter: one gate per backend, no caller change. + One formula, no thresholds, no server telemetry, no mode state machine: + + when a measurement window fills: + gradient = median per-stream speed this window / previous window + limit = limit * gradient + sqrt(limit) + + - healthy (gradient ~1): grows by sqrt(limit) -- big steps at low + limits (fast escape from 1), small relative steps at high limits + (conservative where overshoot is fatal) + - degraded (gradient <1): multiplicative shrink proportional to how + much each stream actually slowed + - window-vs-window comparison self-normalizes context growth: swe + sessions decode slower as they mature, and an all-time-best baseline + would wrongly shrink the gate forever for that alone + - windows are COUNT driven (min(10, 2*limit) same-stamp samples): + sparse workloads judge slowly by construction -- TCP's trick of + using RTT as the clock, inherited for free. Time-based windows + judge on thin evidence (a 30s window at limit 1 sees zero swe turns) + - every sample is stamped with the limit at admission; only + current-stamp samples close a window. Early completions at a new + level are survivorship-biased (short requests finish first) and + always read optimistic + - per-stream speed = completion_tokens/duration; gateways that report + no usage fall back to 1/duration (completion frequency) + - a failed request: limit x0.7 + a 30s admission pause so the server + can drain (anti death-spiral). Fully dead endpoints are the + runner-level circuit breaker's job. """ - LO = 1 # never go below: progress beats perfection - HI = 96 # sane ceiling for one endpoint - PROBE_S = 5.0 # safety tick (fails/hang detection); ramp decisions use - # level statistics, not this interval alone - INITIAL = 2.0 # class-level start point ('--concurrency auto' rebinds it) - # ---- exponential-probe + binary-search capacity discovery ---- - # probe: 1 -> 2 -> 4 -> ... while throughput keeps IMPROVING (>10%); - # the first level where it plateaus opens a bisect [last_good, bad]; - # bisect narrows to the knee; steady holds there. Any failure x0.7s - # immediately and restarts probing from the shrunken level. - GAIN_EPS = 1.1 # rate must beat the previous level by 10% to keep doubling - MIN_OK = 5 # baseline completions needed at a level before judging - # dwell fallback: JUDGE also when the level has been held this long (with - # >= 1 completion) -- scaled by observed inter-completion gap so a bench - # whose single request takes 60s is not judged on one lone sample at t=25s - DWELL_BASE_S = 120.0 - # robust judging: sample count scales WITH the level (a 2-completion - # estimate at level 8 is pure quantization noise), plus a minimum dwell - # so one lucky tick cannot speak for the whole level + LO = 1 + HI = 96 + INITIAL = 1.0 # '--concurrency auto' rebinds this + SILENCE_S = 30.0 # post-failure drain pause before admitting again + WINDOW_MAX = 10 # samples to close a window (2*limit when smaller) def __init__(self, adapter: ModelAdapter): self.adapter = adapter - self.limit = self.INITIAL # float for smooth x0.7; compare with int() + self.limit = self.INITIAL self._inflight = 0 self._cond: Optional[asyncio.Condition] = None - self._task: Optional[asyncio.Task] = None + self._loop = None self._stopped = False - self._loop = None # loop the cond/probe-task are bound to - self._contended = 0 # acquire() waits this probe interval (demand) - self._interval_fails = 0 # failed releases this probe interval - self._interval_ok = 0 # SUCCESSFUL releases this probe interval: - # zero completions = zero throughput, and a - # gate that ramps on demand alone would pile - # 96 concurrent prefills onto a server whose - # first 2 requests have not even answered - # capacity-discovery state - self._mode = 'probe' # probe | bisect | steady - self._level_t0 = None # when we arrived at the current limit - self._level_ok = 0 # completions at this level - 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} + self._quiet_until = 0.0 # post-failure drain window + self._window: List[tuple] = [] # (stamp, speed) since last close + self._prev_median: Optional[float] = None + self.stats = {'window': 0, 'grow': 0, 'shrink': 0, 'backoff_fail': 0} + # ---- bar plumbing ---- def _push_limit(self) -> None: - """Surface the current limit to the progress bar ('gate N').""" rep = (self.adapter.extra or {}).get('progress_reporter') fn = getattr(rep, 'set_gate', None) if fn is not None: @@ -213,9 +205,6 @@ class AdaptiveGate: pass def push_inflight(self) -> None: - """Tell the bar how many requests the gate has ACTUALLY admitted - (bar shows 'admitted/held' -- a bare held count with gate 2 read - as 'the gate is broken').""" rep = (self.adapter.extra or {}).get('progress_reporter') fn = getattr(rep, 'set_admitted', None) if fn is not None: @@ -225,262 +214,95 @@ class AdaptiveGate: pass # ---- gate semantics ---- - async def acquire(self) -> None: + async def acquire(self) -> int: + """Admit one request; the return value is its admission stamp + (the limit it was admitted under) -- pass it back to release().""" 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). + # lazy init OR LOOP CHANGE: pools survive across asyncio.run() + # boundaries and a Condition is loop-bound -- rebuild on the + # new loop (nothing is in flight there by construction) self._loop = loop self._cond = asyncio.Condition() self._inflight = 0 self._stopped = False self._push_limit() - 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)): - self._contended += 1 # demand exceeded the cap: potential ramp fuel + while (self._inflight >= max(1, int(self.limit)) + or time.monotonic() < self._quiet_until): + # timed wait: silence expiry must re-open admission even if no + # request completes to wake us await self._cond.acquire() try: - await self._cond.wait() + await asyncio.wait_for(self._cond.wait(), timeout=5.0) + except asyncio.TimeoutError: + pass finally: self._cond.release() self._inflight += 1 self.push_inflight() + return max(1, int(self.limit)) - def release(self, ok: bool) -> None: + def release(self, ok: bool, stamp: int = 0, dur_s: float = 0.0, + tokens: float = 0.0) -> None: self._inflight = max(0, self._inflight - 1) - self.push_inflight() - if ok: - self._interval_ok += 1 - self._level_ok += 1 - now = time.monotonic() - last = getattr(self, '_last_ok_t', None) - if last is not None: - gap = now - last - ge = getattr(self, '_gap_ema', None) - self._gap_ema = gap if ge is None else 0.6 * ge + 0.4 * gap - self._last_ok_t = now - else: # multiplicative decrease -- survival first - self._interval_fails += 1 - # capacity moved (or we overshot): shrink now and restart the - # discovery from the shrunken level - before = self.limit + if not ok: + self.stats['backoff_fail'] += 1 self.limit = max(self.LO, self.limit * 0.7) - if before != self.limit: - self.stats['backoff_fail'] += 1 - self._push_limit() - self._enter_level(mode='probe') + self._quiet_until = time.monotonic() + self.SILENCE_S + self._push_limit() + else: + speed = (tokens / dur_s if tokens > 0 + else 1.0 / dur_s if dur_s > 0 else 0.0) + if speed > 0: + self._window.append((stamp, speed)) + self._maybe_close_window() + self.push_inflight() self._wake() - def _enter_level(self, mode: str = '') -> None: - """Arrive at (a new) limit: start measuring this level fresh.""" - if mode: - self._mode = mode - self._level_t0 = time.monotonic() - self._level_ok = 0 + def _maybe_close_window(self) -> None: + cur = max(1, int(self.limit)) + fresh = [s for st, s in self._window if st == cur] + if len(fresh) < min(self.WINDOW_MAX, 2 * cur): + return # not enough evidence yet: keep waiting + med = statistics.median(fresh) + if self._prev_median: + # dampers (Netflix/Envoy both do this): small-window medians + # are noisy -- a bare ratio swings the limit 5 -> 1 -> 9 -> 2 + # in simulation. Clamp the gradient band and smooth the + # reference so one jittery window cannot whipsaw the gate. + g = max(0.5, min(1.5, med / self._prev_median)) + if g < 0.95: + # clearly degraded: PURE multiplicative shrink -- adding + # sqrt headroom here let a halved speed at low limits + # still RAISE the limit (sim: 3.4 x 0.5 + sqrt(3.4) = 3.6) + new = self.limit * g + self.stats['shrink'] += 1 + else: + # healthy (or within noise): gentle sqrt probe upward + new = self.limit * g + self.limit ** 0.5 + if new > self.limit: + self.stats['grow'] += 1 + self.limit = max(float(self.LO), min(float(self.HI), new)) + self._push_limit() + self._prev_median = (med if self._prev_median is None + else 0.6 * self._prev_median + 0.4 * med) + self._window = [] + self.stats['window'] += 1 def _wake(self) -> None: - if self._cond is not None: - # fire-and-forget notify (loop may not be ours -- best effort) - try: - fut = asyncio.ensure_future(self._notify_all()) - fut.add_done_callback(lambda f: None) - except RuntimeError: - pass + if self._cond is None: + return + try: + asyncio.get_running_loop().create_task(self._notify_all()) + except RuntimeError: + pass # no running loop (shutdown) async def _notify_all(self) -> None: async with self._cond: self._cond.notify_all() - # ---- server-signal probe ---- - def _no_signal_ramp(self) -> None: - """No server signals (no /metrics, 404, gateway stripped it): - discover capacity by measuring THROUGHPUT per concurrency level. - - probe: double while completions/s keeps improving (rate > prev x - 1.1) -- 1, 2, 4, 8 ... reaches the knee in log time - bisect: first level where the gain stalls opens [last_good, bad]; - narrow to the knee with midpoint measurements - steady: hold at the converged level; any failure x0.7s (handled in - release) and probing restarts from the shrunken level - - A level is judged only after MIN_OK completions or MAX_AT_LEVEL_S; - zero completions so far = hold (hang protection).""" - try: - now = time.monotonic() - if self._level_t0 is None: - self._enter_level() - dt = now - self._level_t0 - lvl = max(1, int(self.limit)) - - # not enough evidence yet at this level: keep measuring. - # evidence = 2 x level completions, no other floor - need_ok = max(2, lvl * 2) - # dwell fallback scales with the OBSERVED completion cadence: - # a 60s-per-request bench needs minutes, not 25s, before a - # single-sample judgment is acceptable - need_dt = max(self.DWELL_BASE_S, - 3.0 * (getattr(self, '_gap_ema', None) or 0.0)) - if self._level_ok < need_ok and dt < need_dt: - return - # zero completions so far: hang or overloaded -> hold - if self._level_ok == 0: - return - - rate = self._level_ok / max(dt, 1e-6) - prev_lvl, prev_rate = self._prev - self._prev = (lvl, rate) - - if self._mode == 'probe': - self.stats['probe'] += 1 - # continue while NOT WORSE (>= 0.9x): with heterogeneous - # request lengths (lbv2: 4k..2M-token docs) completion-rate - # noise dwarfs a 10% gain threshold, and demanding strict - # improvement bisected [1,2]->1 on the first plateau. - # Only CLEAR degradation (<0.9x) means past the knee. - ok = prev_rate is None or rate >= prev_rate * 0.9 - if ok and lvl < self.HI: - self._bis = (lvl, min(lvl * 2, self.HI)) # remember bounds - self.limit = float(min(lvl * 2, self.HI)) - self.stats['ramp_demand'] += 1 - self._push_limit() - self._enter_level() - elif not ok: - # throughput CLEARLY degraded: knee is in (prev_lvl, lvl] - self._mode = 'bisect' - self._bis = (prev_lvl or max(1, lvl // 2), lvl) - self._good_rate = prev_rate or rate - lo, hi = self._bis - mid = (lo + hi) // 2 - if hi - lo <= 1: - self.limit = float(lo) # prev_lvl was the knee - self._push_limit() - self._enter_level('steady') - else: - self.limit = float(mid) - self.stats['bisect'] += 1 - self._push_limit() - self._enter_level() - else: - self._enter_level('steady') # hit HI with gains: stay - elif self._mode == 'steady': - # the endpoint is SHARED: other tenants change its capacity - # while we run -- keep judging forever, nudge +-1 against the - # reference rate instead of pinning the converged level - ref = self._good_rate or rate - if rate >= ref * 1.05 and lvl < self.HI: - self.limit = float(min(lvl + 1, self.HI)) - self.stats['ramp_demand'] += 1 - self._push_limit() - self._enter_level() - elif rate <= ref * 0.85 and lvl > self.LO: - self.limit = float(max(self.LO, lvl - 1)) - self.stats['backoff_queue'] += 1 - self._push_limit() - self._enter_level() - else: - # reference drifts with fresh measurements (slow EWMA) - self._good_rate = 0.7 * ref + 0.3 * rate - self._enter_level() # restart the measurement window - elif self._mode == 'bisect': - lo, hi = self._bis - if rate >= self._good_rate * 0.9: - lo = lvl # not worse here: knee is at/above - else: - hi = lvl # clearly worse: knee is below - self._bis = (lo, hi) - if hi - lo <= 1: - self.limit = float(lo) - self._push_limit() - self._enter_level('steady') - else: - mid = (lo + hi) // 2 - self.limit = float(mid) - self.stats['bisect'] += 1 - self._push_limit() - self._enter_level() - finally: - self._interval_fails = 0 - self._interval_ok = 0 - self._contended = 0 - 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' - - def _fetch(): - with urllib.request.urlopen(url, timeout=4) as resp: - return resp.read().decode('utf-8', 'ignore') - - try: - # 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'): - running = float(line.rsplit(' ', 1)[-1]) - elif line.startswith('sglang:num_queue_reqs'): - queue = float(line.rsplit(' ', 1)[-1]) - self.stats['probe'] += 1 - if queue is None and running is None: - self._no_signal_ramp() - return - if queue is not None and queue >= 2: - # server is queuing OUR excess: gentle additive decrease - self.limit = max(self.LO, self.limit - 1) - self.stats['backoff_queue'] += 1 - self._push_limit() - elif (queue or 0) == 0 and (running is None or running < max(2, int(self.limit))) \ - and self._interval_ok > 0: - # underfed: no queue and running below our own cap -> ramp up - # (still requires completions this interval: no throughput, no ramp) - self.limit = min(self.HI, self.limit + 1) - self.stats['ramp'] += 1 - self._push_limit() - else: - self.stats['hold_queue'] += 1 - self._interval_fails = 0 - self._contended = 0 - self._wake() - - async def _probe_loop(self) -> None: - import contextlib - - while not self._stopped: - with contextlib.suppress(Exception): - await self._probe_once() - await asyncio.sleep(self.PROBE_S) - def stop(self) -> None: self._stopped = True - if self._task is not None: - self._task.cancel() def report(self) -> Dict[str, float]: return {'limit': max(1, int(self.limit)), 'inflight': self._inflight, diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py index c057d6d..fe2e9a0 100644 --- a/evalharness/model/runner.py +++ b/evalharness/model/runner.py @@ -446,10 +446,13 @@ async def generate_predictions( get_image_service().register(sorted(_imgs)) except Exception: pass - # let the adapter surface retry attempts to the bar - members = getattr(adapter, 'adapters', [adapter]) - for m_ in members: - m_.extra['progress_reporter'] = progress_reporter + # let the adapter surface retry attempts / gate limits to the bar + # (NB: this was indented INSIDE the except above -- it only ran when + # image registration THREW, so the gate never pushed its limit and + # the bar showed a bare in-flight count with no 'gate N') + members = getattr(adapter, 'adapters', [adapter]) + for m_ in members: + m_.extra['progress_reporter'] = progress_reporter # terminal (post-retry) sample failures are CONTAINED: one sample that # never makes it (server queue ate its first byte past every timeout) @@ -457,6 +460,13 @@ async def generate_predictions( # (scores as wrong, es-parity for timeouts), is NOT checkpointed (a # rerun retries it), and only a total wipeout fails the bench failed_samples: Dict[int, str] = {} + # circuit breaker: a dead/wedged endpoint makes EVERY sample fail fast + # (~7s each) -- without this the runner burns the whole bench in 15 + # minutes marking everything failed (a 465-sample run lost 98/100 to + # a dead sglang before anyone noticed). N consecutive failures with + # zero successes in between = abort now; --resume retries later. + _consec_fail = [0] + CB_ABORT_AT = 12 async def run_one(i_s): i, s = i_s @@ -469,7 +479,19 @@ async def generate_predictions( if progress_reporter is not None: progress_reporter.advance(success=False) failed_samples[i] = f'{type(e).__name__}: {str(e)[:120]}' + if len(failed_samples) == 1: + # surface the FIRST failure's error immediately -- the + # end-of-generation summary arrives minutes later and + # every fast-fail cascade diagnosis died waiting for it + print(f'⚡ FIRST FAILURE: {failed_samples[i]}', flush=True) + _consec_fail[0] += 1 + if _consec_fail[0] >= CB_ABORT_AT: + raise RuntimeError( + f'circuit breaker: {_consec_fail[0]} consecutive ' + f'failures, endpoint appears down (first: ' + f'{next(iter(failed_samples.values()))[:150]})') return i, None # empty marker: no checkpoint write + _consec_fail[0] = 0 if ckpt_store is not None: ckpt_store.append(keys[i], pred) return i, pred diff --git a/evalharness/progress/rich_terminal.py b/evalharness/progress/rich_terminal.py index b53ec7d..2d2133a 100644 --- a/evalharness/progress/rich_terminal.py +++ b/evalharness/progress/rich_terminal.py @@ -141,6 +141,7 @@ class RichTerminalProgress: self.inflight = 0 self.bench_name = description self.restored = max(completed, 0) # checkpoint head start this bench + self._failed = 0 # per-bench failure counter reset self._scoring_for = None # next scoring phase retargets anew desc = f'[green]{self.bench_tag}{description} · generating[/green]' if self.task_id is None: @@ -175,6 +176,7 @@ class RichTerminalProgress: self.started = time.monotonic() self.inflight = 0 self.restored = 0 + self._failed = 0 self.admitted = None self.progress.update( self.task_id, @@ -277,14 +279,21 @@ class RichTerminalProgress: task = self.progress.tasks[self.task_id] completed = task.completed + 1 self.inflight = max(0, self.inflight - 1) + if not success: + # failed samples advance the bar too (the bench must finish) -- + # but surface the count so 'Completed 310/500' cannot masquerade + # as progress when 275 of them died (e.g. image-wait timeouts) + self._failed = getattr(self, '_failed', 0) + 1 elapsed = max(time.monotonic() - self.started, 1e-6) # rate/eta over THIS RUN's fresh samples only: counting the restored # head start would print 141/6s = 23/s when 1 sample was generated fresh = max(completed - getattr(self, 'restored', 0), 0) + failed_txt = (f' [red]· {self._failed} failed[/red]' + if getattr(self, '_failed', 0) else '') self.progress.update( self.task_id, advance=1, - new=self._new_txt(completed), + new=self._new_txt(completed) + failed_txt, rate=f"{fresh / elapsed:.2f}", inflight=self._inflight_txt(), cur='0s', elapsed=_fmt_eta(elapsed), diff --git a/evalharness/sandbox/image_service.py b/evalharness/sandbox/image_service.py index 4b2d82f..07ce5ed 100644 --- a/evalharness/sandbox/image_service.py +++ b/evalharness/sandbox/image_service.py @@ -2,11 +2,10 @@ One service per process. Benches declare the images they will need (the runner registers them when a task arrives); a small worker pool pulls -them in the background — local tar batches first (es's swebench_v -shipment, 500 images offline), network mirror chain second. Sample -execution calls ``ensure``/``wait_ready`` BEFORE ``docker run``: an image -still in flight BLOCKS that sample (no failure, no timeout at 120s) until -the pull service delivers it or exhausts all sources. +them in the background from the CN mirror chain. Sample execution calls +``ensure``/``wait_ready`` BEFORE ``docker run``: an image still in flight +BLOCKS that sample (no failure, no timeout at 120s) until the pull +service delivers it or exhausts all sources. This replaces the two broken timings: - gen-phase `docker run` implicit pull (zero output, 120s timeout kill) @@ -18,51 +17,29 @@ import os import queue import subprocess import threading -from pathlib import Path from typing import Dict, Iterable, List, Optional -# local tar shipments searched before the network (dir -> glob pattern) -_TAR_SOURCES = [ - ('/data1/sora/evalscope/docker/swebench_v', 'swebench_batch_*.tar.gz'), - ('/data1/sora/evalscope/docker', 'bigcodebench-sandbox.tar.gz'), -] - def _docker(args, timeout=300): return subprocess.run(['docker'] + args, capture_output=True, text=True, timeout=timeout) +_LOCAL_TTL_S = 60.0 # `docker images` on a daemon busy with saves +_LOCAL_CACHE = [0.0, set()] # [expires_at, names] -- one query per TTL + + def _local_images() -> set: + import time + + now = time.monotonic() + if now < _LOCAL_CACHE[0]: + return _LOCAL_CACHE[1] r = _docker(['images', '--format', '{{.Repository}}:{{.Tag}}'], timeout=60) - return set(r.stdout.split()) if r.returncode == 0 else set() - - -def _tar_manifest_tags(path: str) -> List[str]: - """RepoTags inside a docker-save tar ( OCI layout or legacy ).""" - import tarfile - - try: - with tarfile.open(path, 'r:gz') as t: - for name in ('manifest.json', 'index.json'): - try: - member = t.getmember(name) - except KeyError: - continue - data = json.load(t.extractfile(member)) - if name == 'manifest.json': - return [tg for e in data for tg in (e.get('RepoTags') or [])] - # OCI index: ref lives in annotations; best effort - refs = [] - for m in data.get('manifests', []): - ref = (m.get('annotations') or {}).get( - 'org.opencontainers.image.ref.name', '') - if ref: - refs.append(ref) - return refs - except Exception: - return [] - return [] + names = set(r.stdout.split()) if r.returncode == 0 else set() + _LOCAL_CACHE[0] = now + _LOCAL_TTL_S + _LOCAL_CACHE[1] = names + return names class ImageService: @@ -76,7 +53,6 @@ class ImageService: self._events: Dict[str, threading.Event] = {} self._queue: 'queue.Queue[Optional[str]]' = queue.Queue() self._threads: List[threading.Thread] = [] - self._tar_index: Optional[Dict[str, str]] = None # tag -> tar path self._started = False self._pulled: List[str] = [] # for atexit release @@ -92,19 +68,37 @@ class ImageService: self._threads.append(t) def register(self, images: Iterable[str]): - """Declare upcoming images (called when tasks arrive). Idempotent; - already-local images skip the queue entirely.""" + """Declare upcoming images (called when tasks arrive). Idempotent + and MEMORY-FIRST: the run bulk-registers everything once, so the + per-task register must be a dict hit with ZERO docker calls -- + `docker images` per task (x100) against a daemon busy with + concurrent `docker save`s timed out at 60s and cascaded whole + runs into failure. Only genuinely UNKNOWN images trigger one + (TTL-cached) docker query for the batch; if even that fails under + daemon pressure, mark ready optimistically and let `docker run` + be the referee -- it has its own timeout and a clear error.""" self.start() - local = _local_images() - for img in images: - if not img: - continue - with self._lock: + unknown = [] + with self._lock: + for img in images: + if not img: + continue if img in self._wanted: continue self._wanted[img] = 'pulling' self._events[img] = threading.Event() - if img in local: + unknown.append(img) + if not unknown: + return + try: + local = _local_images() + except Exception: + local = None # daemon query failed: optimistic path below + for img in unknown: + if local is None or img in local: + if local is None: + self._log(f'· docker images unavailable -- assuming ' + f'local: {img.split("/")[-1][:44]}') with self._lock: self._wanted[img] = 'ready' self._events[img].set() @@ -143,16 +137,29 @@ class ImageService: self._queue.task_done() def _pull_one(self, image: str) -> bool: - # 1. local tar shipment (offline, fastest) - tar = self._find_in_tars(image) - if tar: - self._log(f'· loading {image.split("/")[-1][:44]} from {Path(tar).parent.name}/{Path(tar).name}') - r = _docker(['load', '-qi', tar], timeout=1800) - if r.returncode == 0 and image in _local_images(): - self._pulled.append(image) - return True - self._log(f' tar load failed, falling back to network') - # 2. mirror chain (progress + watchdog inside) + # fast path: a concurrent worker / previous run may have delivered + # it already (also flips any other queued-now-local images ready) + if image in _local_images(): + self._sweep_local() + return True + # 1. OUR ModelScope dataset shipment (CN CDN, per-image tars). The + # swebench/* namespace is 403 on every public CN mirror, so this + # is the only network source for those images. Opt out with + # EVALHARNESS_MS_IMAGE_REPO='' . + repo = os.environ.get('EVALHARNESS_MS_IMAGE_REPO', + 'SoraAmami/swebench-images') + if repo: + try: + from .ms_images import ms_pull + + if ms_pull(image, repo): + self._pulled.append(image) + self._sweep_local() + return True + except Exception: + pass + # 2. CN mirror chain (progress + idle watchdog inside); on hit the + # puller retags to the canonical name so callers see it directly try: from .prefetch import _pull_one as net_pull @@ -164,42 +171,23 @@ class ImageService: pass return False - # ---- tar index ---- - _INDEX_CACHE = '/tmp/evalharness_tar_index.json' - - def _find_in_tars(self, image: str) -> Optional[str]: - # the full 50-tar index costs minutes to build (gzip full-scan); - # cache it on disk so only the FIRST process ever pays - import glob - - if self._tar_index is None: - cache = {} - try: - cache = json.load(open(self._INDEX_CACHE)) - # invalidate when the tar set changes - cur = sorted(glob.glob('/data1/sora/evalscope/docker/swebench_v/*.tar.gz')) - if cache.get('_files') != [os.path.basename(x) for x in cur]: - cache = {} - except Exception: - cache = {} - if cache.get('tags'): - self._tar_index = cache['tags'] - else: - self._tar_index = {} - for d, pat in _TAR_SOURCES: - if not os.path.isdir(d): - continue - for f in sorted(glob.glob(os.path.join(d, pat))): - for tag in _tar_manifest_tags(f): - self._tar_index.setdefault(tag, f) - try: - files = [os.path.basename(x) for x in - glob.glob('/data1/sora/evalscope/docker/swebench_v/*.tar.gz')] - json.dump({'_files': sorted(files), 'tags': self._tar_index}, - open(self._INDEX_CACHE, 'w')) - except Exception: - pass - return self._tar_index.get(image) + def _sweep_local(self): + """A tar load makes ~10 images local at once: flip every queued + 'pulling' image that is now local to 'ready' (and drop it from the + release list duty) -- otherwise each waits its own full queue turn + just to discover it already arrived.""" + with self._lock: + waiting = [img for img, st in self._wanted.items() + if st == 'pulling'] + if not waiting: + return + local = _local_images() + with self._lock: + for img in waiting: + if img in local: + self._wanted[img] = 'ready' + self._events[img].set() + self._pulled.append(img) # we caused it; release at exit # ---- misc ---- def _log(self, msg): diff --git a/evalharness/sandbox/ms_images.py b/evalharness/sandbox/ms_images.py new file mode 100644 index 0000000..e2095c2 --- /dev/null +++ b/evalharness/sandbox/ms_images.py @@ -0,0 +1,214 @@ +"""ModelScope-hosted sandbox images: per-image tar upload + download. + +registry.modelscope.cn does not accept personal docker pushes, so images +ship as ONE docker-save tar per image inside a ModelScope MODEL repo +(file hosting with a stable CN CDN -- measured ~12MB/s anonymous; the +per-task granularity means running 3 instances downloads exactly 3 tars). +This is the only working network source for the swebench/* namespace -- +public CN mirrors (daocloud/1ms/...) 403 it. + +NB: it must be a MODEL repo, not a dataset repo: upload_file defaults to +repo_type='model' and silently CREATES one when given a dataset repo_id +(the first upload landed in a phantom model repo while the dataset repo +stayed empty -- confusing 404s on pull). + +Upload (resumable; skips tars already on the remote): + EVALHARNESS_MS_TOKEN=ms-... python -m evalharness.sandbox.ms_images \\ + upload --repo SoraAmami/swebench-images [--limit N] + +Download side is wired into ImageService._pull_one (first source, before +the docker-hub mirror chain) whenever EVALHARNESS_MS_IMAGE_REPO is set +(defaults on for this deployment; empty string disables). +""" + +import os +import subprocess +import sys +import tempfile +import time +from typing import List, Optional, Set + +# a private-repo token for downloads (public repos work without it) +_MS_TOKEN = os.environ.get('EVALHARNESS_MS_TOKEN', '').strip() +_MS_ENDPOINT = os.environ.get('EVALHARNESS_MS_ENDPOINT', 'https://modelscope.cn') +# big enough for a ~4GB uncompressed save; / has less headroom than /data1 +_TMP_ROOT = os.environ.get('EVALHARNESS_MS_TMP', '/data1/sora/temp/ms_images') +_STATE = os.path.join(_TMP_ROOT, 'uploaded.txt') # resume manifest + +# image prefixes this deployment ships (swe per-instance + code benches) +_DEFAULT_PREFIXES = ('swebench/',) + + +def _docker(args, timeout=1800): + return subprocess.run(['docker'] + args, capture_output=True, text=True, + timeout=timeout) + + +def file_for(image: str) -> str: + """Image ref -> dataset filename: last path segment minus tag. + + swebench/sweb.eval.x86_64.django_1776_django-13964:latest + -> sweb.eval.x86_64.django_1776_django-13964.tar.gz + """ + base = image.rsplit('/', 1)[-1] + base = base.split(':')[0] + return f'{base}.tar.gz' + + +def image_from_file(fname: str) -> Optional[str]: + """Inverse of file_for for the swe namespace (namespace is not + recoverable in general; callers here only ship swebench/*).""" + if not fname.endswith('.tar.gz'): + return None + return f"swebench/{fname[:-len('.tar.gz')]}:latest" + + +def _local_images() -> Set[str]: + r = _docker(['images', '--format', '{{.Repository}}:{{.Tag}}'], timeout=60) + return set(r.stdout.split()) if r.returncode == 0 else set() + + +def _remote_files(repo: str) -> Set[str]: + """Files currently in the MODEL repo (resume manifest source of truth).""" + from modelscope.hub.api import HubApi + + api = HubApi() + if _MS_TOKEN: + api.login(_MS_TOKEN) + files: Set[str] = set() + for e in api.get_model_files(repo, recursive=True): + p = (e.get('Path') or e.get('Name') or '').lstrip('/') + if p.endswith('.tar.gz'): + files.add(p) + return files + + +def _upload_one(api, repo: str, image: str, tmp_dir: str) -> bool: + tar = os.path.join(tmp_dir, file_for(image)) + os.makedirs(tmp_dir, exist_ok=True) + # gzip -1: ~2x faster than default for ~10% more bytes -- upload wall + # time is dominated by docker save IO, not the extra size + r = subprocess.run(f'docker save {image} | gzip -1 > {tar}', + shell=True, timeout=3600) + if r.returncode != 0 or not os.path.exists(tar): + print(f' save failed: {image}', flush=True) + return False + try: + api.upload_file(repo_id=repo, path_or_fileobj=tar, + path_in_repo=file_for(image), + repo_type='model', token=_MS_TOKEN or None) + with open(_STATE, 'a') as f: + f.write(file_for(image) + '\n') + gb = os.path.getsize(tar) / 1e9 + print(f' uploaded {file_for(image)} ({gb:.2f} GB)', flush=True) + return True + except Exception as e: + print(f' upload failed {image}: {type(e).__name__}: {str(e)[:120]}', + flush=True) + return False + finally: + try: + os.unlink(tar) + except OSError: + pass + + +def upload(repo: str, limit: int = 0, prefixes=None) -> None: + """Ship every local matching image to the MODEL repo. Resumable: + already-remote tars are skipped (state file + remote listing).""" + from modelscope.hub.api import HubApi + + os.makedirs(_TMP_ROOT, exist_ok=True) + api = HubApi() + if _MS_TOKEN: + api.login(_MS_TOKEN) + try: + api.get_model(repo) + except Exception: + print(f'creating model repo {repo} ...', flush=True) + api.create_model(model_id=repo) + done = set() + try: + done = {l.strip() for l in open(_STATE) if l.strip()} + except OSError: + pass + print(f'remote listing {repo} ...', flush=True) + try: + done |= _remote_files(repo) + except Exception as e: + print(f' remote listing failed ({e}); relying on local state', + flush=True) + prefixes = prefixes or _DEFAULT_PREFIXES + imgs = sorted(i for i in _local_images() if i.startswith(prefixes)) + if limit: + imgs = imgs[:limit] + todo = [i for i in imgs if file_for(i) not in done] + print(f'{len(imgs)} local images, {len(todo)} to upload ' + f'({len(imgs) - len(todo)} already remote)', flush=True) + ok = 0 + for n, img in enumerate(todo, 1): + print(f'[{n}/{len(todo)}] {img}', flush=True) + for attempt in range(3): + if _upload_one(api, repo, img, _TMP_ROOT): + ok += 1 + break + time.sleep(10 * (attempt + 1)) # hub hiccups: backoff + retry + print(f'done: {ok}/{len(todo)} uploaded', flush=True) + + +def ms_pull(image: str, repo: str) -> bool: + """Fetch one image tar from the dataset repo and docker-load it. + Returns True iff the image is local afterwards.""" + import urllib.request + + fname = file_for(image) + url = (f'{_MS_ENDPOINT}/api/v1/models/{repo}/repo' + f'?Revision=master&FilePath={fname}') + hdrs = {'User-Agent': 'evalharness'} + if _MS_TOKEN: + hdrs['Authorization'] = f'Bearer {_MS_TOKEN}' + os.makedirs(_TMP_ROOT, exist_ok=True) + fd, tar = tempfile.mkstemp(suffix='.tar.gz', dir=_TMP_ROOT) + os.close(fd) + try: + req = urllib.request.Request(url, headers=hdrs) + t0 = time.time() + with urllib.request.urlopen(req, timeout=600) as r, open(tar, 'wb') as f: + while True: + chunk = r.read(1 << 20) + if not chunk: + break + f.write(chunk) + r = _docker(['load', '-qi', tar], timeout=1800) + if r.returncode == 0 and image in _local_images(): + gb = os.path.getsize(tar) / 1e9 + print(f'· ms-images: loaded {image.split("/")[-1][:50]} ' + f'({gb:.2f} GB in {time.time() - t0:.0f}s)', flush=True) + return True + print(f'· ms-images: load failed for {image}: ' + f'{(r.stderr or "")[:120]}', flush=True) + return False + except Exception as e: + # 404 = this image was never uploaded; anything else = transient + code = getattr(e, 'code', None) + if code != 404: + print(f'· ms-images: fetch {fname}: {type(e).__name__}: ' + f'{str(e)[:100]}', flush=True) + return False + finally: + try: + os.unlink(tar) + except OSError: + pass + + +if __name__ == '__main__': + import argparse + + p = argparse.ArgumentParser() + p.add_argument('cmd', choices=['upload']) + p.add_argument('--repo', default='SoraAmami/swebench-images') + p.add_argument('--limit', type=int, default=0) + a = p.parse_args() + if a.cmd == 'upload': + upload(a.repo, limit=a.limit)