Retry chain sanity: adapter 3 retries for long budgets / 6 for normal (was 6 flat); run_one 3 retries (was 6) with shorter backoff; worst case now 12 attempts ~5min (was 42+ attempts ~20min); progress bar shows live retry count; heartbeat recreated on every reset_samples
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
d00f330430
commit
55b2beab54
@ -192,7 +192,10 @@ class OpenAICompatible(ModelAdapter):
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
if self.api_key:
|
||||
headers['Authorization'] = f'Bearer {self.api_key}'
|
||||
retries = self.extra.get('retries', 6)
|
||||
# long generations cost minutes per attempt -- retry far less
|
||||
# (3 for normal requests, 1 for >8k-token budgets)
|
||||
_mt = int((kw.get('max_tokens') or payload.get('max_tokens') or 4096))
|
||||
retries = self.extra.get('retries', 3 if _mt > 8192 else 6)
|
||||
last_exc: Exception = None
|
||||
stream = bool(self.extra.get('collect_perf') and not kw.get('no_stream'))
|
||||
if stream:
|
||||
@ -255,6 +258,9 @@ class OpenAICompatible(ModelAdapter):
|
||||
raise
|
||||
import asyncio
|
||||
|
||||
# surface the retry to the progress bar (user visibility)
|
||||
if self.extra.get('progress_reporter') is not None:
|
||||
self.extra['progress_reporter'].set_retries(attempt + 1)
|
||||
await asyncio.sleep(min(2 ** attempt * 3, 120))
|
||||
raise last_exc # unreachable
|
||||
|
||||
|
||||
@ -408,6 +408,10 @@ async def generate_predictions(
|
||||
|
||||
if progress_reporter is not None:
|
||||
progress_reporter.reset_samples(len(work), dataset_name, completed=len(restored))
|
||||
# let the adapter surface retry attempts to the bar
|
||||
members = getattr(adapter, 'adapters', [adapter])
|
||||
for m_ in members:
|
||||
m_.extra['progress_reporter'] = progress_reporter
|
||||
|
||||
async def run_one(i_s):
|
||||
i, s = i_s
|
||||
@ -415,18 +419,18 @@ async def generate_predictions(
|
||||
# backoff so one ConnectError burst cannot kill the whole batch --
|
||||
# the adapter already retries 5xx/429 and the pool fails over per
|
||||
# instance; this is the last line of defense around asyncio.gather
|
||||
for attempt in range(6):
|
||||
for attempt in range(3):
|
||||
try:
|
||||
pred = await one(s)
|
||||
break
|
||||
except Exception:
|
||||
if attempt == 5:
|
||||
if attempt == 2:
|
||||
if progress_reporter is not None:
|
||||
progress_reporter.advance(success=False)
|
||||
raise
|
||||
# minute-scale backoff: cluster routes flap in multi-minute
|
||||
# bursts; short retries exhaust inside one dead window
|
||||
await asyncio.sleep((10, 20, 40, 60, 90)[attempt])
|
||||
await asyncio.sleep((10, 20, 40)[attempt])
|
||||
if ckpt_store is not None:
|
||||
ckpt_store.append(keys[i], pred)
|
||||
return i, pred
|
||||
|
||||
@ -39,6 +39,7 @@ class RichTerminalProgress:
|
||||
TaskProgressColumn(),
|
||||
TextColumn("• Completed {task.completed}/{task.total}"),
|
||||
TextColumn("• in-flight {task.fields[inflight]} ({task.fields[cur]})"),
|
||||
TextColumn("• retries {task.fields[retries]}"),
|
||||
TextColumn("• {task.fields[rate]}/s"),
|
||||
TextColumn("• {task.fields[elapsed]}"),
|
||||
TextColumn("• eta {task.fields[eta]}"),
|
||||
@ -113,14 +114,14 @@ class RichTerminalProgress:
|
||||
self.task_id = self.progress.add_task(
|
||||
desc, total=total, completed=min(completed, total),
|
||||
success=completed, failed=0, rate='0.00', inflight=0,
|
||||
cur='0s', elapsed='0s', eta='-',
|
||||
cur='0s', elapsed='0s', eta='-', retries=0,
|
||||
waiting='00:00', last_result='restored')
|
||||
else:
|
||||
self.progress.update(self.task_id, description=desc,
|
||||
total=total, completed=min(completed, total),
|
||||
success=completed, failed=0, rate='0.00',
|
||||
inflight=0, cur='0s', elapsed='0s', eta='-',
|
||||
last_result='restored')
|
||||
inflight=0, cur='0s', elapsed='0s', eta='-', retries=0,
|
||||
last_result='redone')
|
||||
# ALWAYS recreate the heartbeat: the previous one may have died during
|
||||
# pause/resume cycles between benchmarks (stale reference -> silent
|
||||
# death -> frozen clock while the spinner still animates)
|
||||
@ -157,6 +158,11 @@ class RichTerminalProgress:
|
||||
self.progress.update(self.task_id, inflight=self.inflight, cur='0s',
|
||||
waiting="00:00", last_result=f"waiting {label}")
|
||||
|
||||
def set_retries(self, n: int):
|
||||
"""Show the retry count on the bar (from the adapter's attempt)."""
|
||||
if self.task_id is not None:
|
||||
self.progress.update(self.task_id, retries=n)
|
||||
|
||||
def rollback(self):
|
||||
if self.disabled:
|
||||
return
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user