Meta-Eval/EvalHarness
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Gate: exponential probe + binary search for capacity discovery

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
sora 2026-09-15 02:42:53 +00:00
parent 4eb853c8fc
commit 52547de9a3
2 changed files with 119 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
evalharness/cli.py
View File

@ -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/<stamp>-<model>/
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)

123
evalharness/model/pool.py
View File

@ -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,16 +270,85 @@ 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)
"""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