From 52547de9a3f3a733f7c91193b81c0384d874b3e2 Mon Sep 17 00:00:00 2001 From: sora <2075279110@qq.com> Date: Tue, 15 Sep 2026 02:42:53 +0000 Subject: [PATCH] Gate: exponential probe + binary search for capacity discovery Replaces the +1/5s linear ramp: start at 1, double while measured completions/s keeps improving (>10% over the previous level); the first plateau opens a bisect [last_good, bad] that narrows to the knee, then holds steady. Failures still cut x0.7 instantly and restart probing from the shrunken level; zero completions = hold. Judging a level needs max(MIN_OK, level) completions -- a 2-completion rate estimate at level 8 is quantization noise (caught by simulation converging to 1 on a capacity-8 endpoint). Simulated against throughput curves min(level, capacity): capacity 8 -> 1,2,4,8,16 | bisect 12,10,9 -> steady 8 capacity 16 -> 1,2,4,8,16,32 | bisect ... -> steady 16 capacity 4 -> 1,2,4,8 | bisect 6,5 -> steady 4 Co-Authored-By: Claude --- evalharness/cli.py | 4 +- evalharness/model/pool.py | 135 +++++++++++++++++++++++++++++++++----- 2 files changed, 119 insertions(+), 20 deletions(-) diff --git a/evalharness/cli.py b/evalharness/cli.py index 7081be5..ffe782f 100644 --- a/evalharness/cli.py +++ b/evalharness/cli.py @@ -696,7 +696,7 @@ def _cmd_eval_run(args) -> int: # adapts on its own signals (see pool.AdaptiveGate) from evalharness.model.pool import AdaptiveGate - AdaptiveGate.INITIAL = float(max(2, getattr(args, 'concurrency', 8))) + AdaptiveGate.INITIAL = float(max(1, getattr(args, 'concurrency', 1))) if not out_dir and model_spec and not args.out: # always persist results: default dir = evalharness-results/-/ import re as _re @@ -1252,7 +1252,7 @@ def main(argv=None) -> int: # an int + the flag if str(getattr(args, 'concurrency', '32')).strip().lower() == 'auto': args.auto_concurrency = True - args.concurrency = 2 # gate start; it ramps on its own signals + args.concurrency = 1 # gate starts at 1: probe x2, bisect to capacity else: args.concurrency = int(args.concurrency) return args.func(args) diff --git a/evalharness/model/pool.py b/evalharness/model/pool.py index 9b95c2a..4b1a586 100644 --- a/evalharness/model/pool.py +++ b/evalharness/model/pool.py @@ -151,10 +151,22 @@ class AdaptiveGate: Purely additive to PooledAdapter: one gate per backend, no caller change. """ - LO = 2 # never go below: progress beats perfection + LO = 1 # never go below: progress beats perfection HI = 96 # sane ceiling for one endpoint - PROBE_S = 5.0 # metrics probe interval - INITIAL = 8.0 # class-level start point (--auto-concurrency rebinds it) + 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 = 3 # baseline completions needed at a level before judging + MAX_AT_LEVEL_S = 25 # ... or this many seconds, whichever comes first + # 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 def __init__(self, adapter: ModelAdapter): self.adapter = adapter @@ -171,8 +183,15 @@ class AdaptiveGate: # 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.stats = {'probe': 0, 'ramp': 0, 'hold_queue': 0, 'backoff_fail': 0, - 'backoff_queue': 0, 'ramp_demand': 0} + 'backoff_queue': 0, 'ramp_demand': 0, 'bisect': 0} def _push_limit(self) -> None: """Surface the current limit to the progress bar ('gate N').""" @@ -216,15 +235,26 @@ class AdaptiveGate: self._inflight = max(0, self._inflight - 1) if ok: self._interval_ok += 1 + self._level_ok += 1 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 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._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 _wake(self) -> None: if self._cond is not None: # fire-and-forget notify (loop may not be ours -- best effort) @@ -240,20 +270,89 @@ class AdaptiveGate: # ---- server-signal probe ---- def _no_signal_ramp(self) -> None: - """No server signals available (no /metrics, 404/HTTPError, gateway - stripped it, non-sglang backend): fall back to demand-driven AIMD -- - ramp while the cap is the binding constraint (callers had to WAIT on - acquire) and the interval was failure-free. Failures still cut x0.7 - per release, so a drowning backend shrinks the gate immediately.""" - if self._interval_fails == 0 and self._interval_ok > 0 \ - and self._contended > 0 and int(self.limit) < self.HI: - self.limit = min(self.HI, self.limit + 1) - self.stats['ramp_demand'] += 1 - self._push_limit() - self._interval_fails = 0 - self._interval_ok = 0 - self._contended = 0 - self._wake() + """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. + # need = max(MIN_OK, level): rate noise shrinks only with + # samples proportional to the concurrency being judged + need_ok = max(self.MIN_OK, lvl) + if self._level_ok < need_ok and dt < self.MAX_AT_LEVEL_S: + return + # zero completions in MAX_AT_LEVEL_S: 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 + improved = prev_rate is None or rate > prev_rate * self.GAIN_EPS + if improved 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 improved: + # throughput plateaued: knee is between prev_lvl and 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 == 'bisect': + lo, hi = self._bis + if rate > self._good_rate * self.GAIN_EPS: + lo = lvl # still improving: knee is higher + self._good_rate = rate + else: + hi = lvl # no gain: knee is lower + 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: import urllib.request