From afe2fb8d28f2ef5c0757a8a07e91924d0bd9d938 Mon Sep 17 00:00:00 2001 From: sora <2075279110@qq.com> Date: Thu, 10 Sep 2026 10:47:21 +0000 Subject: [PATCH] Progress coverage for minute-scale work: overall bench bar (which benchmark of N, covers loading/scoring), byte-level download bars for dataset fetches (ModelScope/HF raw, tty-only), one shared reporter across benchmarks (caller-owned lifecycle) Co-Authored-By: Claude --- evalharness/cli.py | 17 +++++++- evalharness/data/loader.py | 60 ++++++++++++++++++++++----- evalharness/model/runner.py | 7 +++- evalharness/progress/rich_terminal.py | 40 ++++++++++++++++++ 4 files changed, 110 insertions(+), 14 deletions(-) diff --git a/evalharness/cli.py b/evalharness/cli.py index 92a88f1..91ed8fc 100644 --- a/evalharness/cli.py +++ b/evalharness/cli.py @@ -471,6 +471,7 @@ def _cmd_eval_run(args) -> int: Path(out_dir).mkdir(parents=True, exist_ok=True) console = _rich_console() _print_run_plan(console, args, model_spec) + _shared_reporter = None for i, name in enumerate(args.datasets): t0 = _time.time() try: @@ -490,8 +491,16 @@ def _cmd_eval_run(args) -> int: if RichTerminalProgress is not None: # share ONE console: phase lines printed by another - # writer during the live bar interleave incorrectly - progress_reporter = RichTerminalProgress(console=console) + # writer during the live bar interleave incorrectly. + # One reporter for the WHOLE run: overall bar (which + # benchmark) + sample bar (which sample), reused per + # benchmark via reset_samples(). + if _shared_reporter is None: + _shared_reporter = RichTerminalProgress(console=console) + _shared_reporter.owned_externally = True + progress_reporter = _shared_reporter + if total_runs > 1: + progress_reporter.set_overall(total_runs, i, 'benches') def status_callback(msg, _idx=i + 1, _name=name, _reporter=progress_reporter, @@ -569,6 +578,8 @@ def _cmd_eval_run(args) -> int: 'lat_p50': _pct(0.50), 'lat_p90': _pct(0.90), 'trunc': sum(1 for f in fins if f == 'length'), 'groups': groups, 'ok': True}) + if progress_reporter is not None: + progress_reporter.advance_overall() _print_benchmark_result(console, i + 1, total_runs, name, 'done', _time.time() - t0) except Exception as e: @@ -586,6 +597,8 @@ def _cmd_eval_run(args) -> int: _print_benchmark_result(console, i + 1, total_runs, name, 'failed', _time.time() - t0) + if _shared_reporter is not None: + _shared_reporter.close() if all_reports and out_dir: try: from evalharness.viz import render as _render diff --git a/evalharness/data/loader.py b/evalharness/data/loader.py index 773d9ff..066ca38 100644 --- a/evalharness/data/loader.py +++ b/evalharness/data/loader.py @@ -13,6 +13,7 @@ Supported sources: import csv import json import os +import sys import re import shutil import urllib.request @@ -212,6 +213,53 @@ def _parquet_ok(path: Path) -> bool: return False + +def _download_with_progress(resp, out, filename: str) -> None: + """Stream a download with a rich byte-level progress bar. + + Used for the minute-scale dataset fetches (ModelScope blobs, HF raw + files); silent (plain streaming) when rich is unavailable or output + is redirected.""" + total = 0 + try: + total = int(resp.headers.get('Content-Length') or 0) + except (TypeError, ValueError): + total = 0 + if not sys.stdout.isatty(): + while True: + chunk = resp.read(1 << 20) + if not chunk: + break + out.write(chunk) + return + try: + from rich.progress import (BarColumn, DownloadColumn, Progress, + SpinnerColumn, TextColumn, + TimeElapsedColumn, TransferSpeedColumn) + + pg = Progress(SpinnerColumn(), + TextColumn('[cyan]{task.fields[name]}[/cyan]'), + BarColumn(), DownloadColumn(), TransferSpeedColumn(), + TextColumn('•'), TimeElapsedColumn(), + transient=True) + with pg: + t = pg.add_task('download', total=total or None, name=filename) + while True: + chunk = resp.read(1 << 20) + if not chunk: + break + out.write(chunk) + pg.advance(t, len(chunk)) + return + except ImportError: + pass + while True: + chunk = resp.read(1 << 20) + if not chunk: + break + out.write(chunk) + + def _ms_download(repo: str, path: str, dest_dir: Path) -> Path: """Download one repo file into the raw cache (content-addressed, reused).""" dest = dest_dir / os.path.basename(path) @@ -225,11 +273,7 @@ def _ms_download(repo: str, path: str, dest_dir: Path) -> Path: tmp = dest.with_name(dest.name + f'.part-{os.getpid()}') req = urllib.request.Request(url, headers={'User-Agent': 'evalharness/0.1'}) with urllib.request.urlopen(req, timeout=600) as resp, open(tmp, 'wb') as out: - while True: - chunk = resp.read(1 << 20) - if not chunk: - break - out.write(chunk) + _download_with_progress(resp, out, dest.name) os.replace(tmp, dest) return dest @@ -369,11 +413,7 @@ def _hf_download(repo: str, path: str, dest_dir: Path) -> Path: tmp = dest.with_name(dest.name + f'.part-{os.getpid()}') req = urllib.request.Request(url, headers={'User-Agent': 'evalharness/0.1'}) with urllib.request.urlopen(req, timeout=1800) as resp, open(tmp, 'wb') as out: - while True: - chunk = resp.read(1 << 20) - if not chunk: - break - out.write(chunk) + _download_with_progress(resp, out, dest.name) os.replace(tmp, dest) return dest diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py index 9789728..e6b6b20 100644 --- a/evalharness/model/runner.py +++ b/evalharness/model/runner.py @@ -405,7 +405,7 @@ async def generate_predictions( status_callback(f'checkpoint restored: {len(restored)} ready, {len(pending)} pending') if progress_reporter is not None: - progress_reporter.start(len(work), dataset_name, completed=len(restored)) + progress_reporter.reset_samples(len(work), dataset_name, completed=len(restored)) async def run_one(i_s): i, s = i_s @@ -441,7 +441,10 @@ async def generate_predictions( status_callback(f'generation complete: {len(preds)} responses') return preds, usages, total_usage finally: - if progress_reporter is not None: + # reporter lifecycle belongs to the CALLER (CLI reuses one reporter + # across benchmarks and closes it after the whole run); only close + # here when nobody external passed it in + if progress_reporter is not None and not getattr(progress_reporter, 'owned_externally', False): progress_reporter.close() diff --git a/evalharness/progress/rich_terminal.py b/evalharness/progress/rich_terminal.py index e4f7147..006661d 100644 --- a/evalharness/progress/rich_terminal.py +++ b/evalharness/progress/rich_terminal.py @@ -20,6 +20,8 @@ class RichTerminalProgress: # accept an EXTERNAL console: CLI phase messages and the live bar must # share one console, or the two writers interleave and repaint wrongly self.console = console or Console() + self.overall_id = None + self.overall_total = 0 self.progress = Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), @@ -41,8 +43,29 @@ class RichTerminalProgress: self.inflight = 0 self.heartbeat_task = None + def set_overall(self, total: int, done: int, label: str = 'benches'): + """Multi-benchmark runs: a bar above the sample bar showing which + benchmark we are on (also covers loading/scoring phases).""" + self.overall_total = total + if self.overall_id is None: + self.progress.start() + # the shared column set reads fields from EVERY task: give the + # overall task the same fields or rendering raises KeyError + self.overall_id = self.progress.add_task( + f'[cyan]{label}[/cyan]', total=total, completed=min(done, total), + inflight=0, rate='0.00', success=done, failed=0, + waiting='00:00', last_result='') + else: + self.progress.update(self.overall_id, completed=min(done, total)) + + def advance_overall(self): + if self.overall_id is not None: + self.progress.advance(self.overall_id) + def start(self, total: int, description: str, completed: int = 0): self.started = time.monotonic() + if self.task_id is not None: + return # one live reporter at a time; reuse across benchmarks self.progress.start() self.task_id = self.progress.add_task( f"[green]{description}", @@ -57,6 +80,23 @@ class RichTerminalProgress: ) self.heartbeat_task = asyncio.create_task(self._heartbeat()) + def reset_samples(self, total: int, description: str, completed: int = 0): + """Start (or re-target) the per-sample task for the next benchmark.""" + self.started = time.monotonic() + self.inflight = 0 + if self.task_id is None: + self.progress.start() + self.task_id = self.progress.add_task( + f'[green]{description}', total=total, completed=min(completed, total), + success=completed, failed=0, rate='0.00', inflight=0, + waiting='00:00', last_result='restored') + self.heartbeat_task = asyncio.create_task(self._heartbeat()) + else: + self.progress.update(self.task_id, description=f'[green]{description}', + total=total, completed=min(completed, total), + success=completed, failed=0, rate='0.00', + inflight=0, last_result='restored') + def begin_sample(self, label: str): if self.task_id is None: return