From e6a131b8353d5a009da7f158752ee9cef5147e83 Mon Sep 17 00:00:00 2001 From: BANADDA MUBARAKA <83466862+BANADDA@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:14:29 +0300 Subject: [PATCH 1/3] validator: report each miner's evaluation stage to the server as it runs --- microtensor_subnet/validator/client.py | 19 ++- microtensor_subnet/validator/epoch.py | 42 ++++++- .../validator/pools/escalation.py | 5 +- microtensor_subnet/validator/pools/harness.py | 19 +-- microtensor_subnet/validator/pools/slm.py | 5 +- microtensor_subnet/validator/pools/system.py | 5 +- microtensor_subnet/validator/progress.py | 111 ++++++++++++++++++ 7 files changed, 192 insertions(+), 14 deletions(-) create mode 100644 microtensor_subnet/validator/progress.py diff --git a/microtensor_subnet/validator/client.py b/microtensor_subnet/validator/client.py index b6cfc58..5779fe2 100644 --- a/microtensor_subnet/validator/client.py +++ b/microtensor_subnet/validator/client.py @@ -14,6 +14,7 @@ SIGNATURE_HEADER: Final[str] = "x-mt-signature" TIMEOUT_SECONDS: Final[float] = 60.0 BUNDLE_TIMEOUT_SECONDS: Final[float] = 600.0 +PROGRESS_TIMEOUT_SECONDS: Final[float] = 5.0 ARENA_NETWORKS: dict[str, str] = {"test": "testnet", "finney": "mainnet"} @@ -46,7 +47,13 @@ def _headers(self, method: str, path: str, body: bytes) -> dict[str, str]: return headers def _request( - self, method: str, path: str, body: dict[str, Any] | None = None, *, signed: bool = False + self, + method: str, + path: str, + body: dict[str, Any] | None = None, + *, + signed: bool = False, + timeout: float | None = None, ) -> Any: raw = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() if body else b"" headers = ( @@ -55,7 +62,9 @@ def _request( else {"accept": "application/json", "content-type": "application/json"} ) try: - with httpx.Client(base_url=self.base_url.rstrip("/"), timeout=self.timeout) as http: + with httpx.Client( + base_url=self.base_url.rstrip("/"), timeout=timeout or self.timeout + ) as http: response = http.request(method, path, content=raw or None, headers=headers) except httpx.HTTPError as exc: raise ServerError(f"{method} {path}: {exc}") from exc @@ -133,3 +142,9 @@ def submit_artifact_result(self, body: dict[str, Any]) -> dict[str, Any]: if not isinstance(found, dict) or not found.get("accepted"): raise ServerError(f"POST /v2/reports/artifact was not accepted: {found!r}"[:300]) return dict(found) + + def report_progress(self, body: dict[str, Any]) -> dict[str, Any]: + found = self._request( + "POST", "/v2/reports/progress", body, signed=True, timeout=PROGRESS_TIMEOUT_SECONDS + ) + return dict(found or {}) diff --git a/microtensor_subnet/validator/epoch.py b/microtensor_subnet/validator/epoch.py index e7b9525..0b4cc75 100644 --- a/microtensor_subnet/validator/epoch.py +++ b/microtensor_subnet/validator/epoch.py @@ -19,13 +19,23 @@ from microtensor_subnet.core.reports import PoolScore from microtensor_subnet.scoring import pools from microtensor_subnet.serving.probes import sets as probe_sets -from microtensor_subnet.validator import collect, compress, items, provenance, report, score +from microtensor_subnet.validator import ( + collect, + compress, + items, + progress, + provenance, + report, + score, +) from microtensor_subnet.validator.client import ServerClient, ServerError from microtensor_subnet.validator.config import EpochConfig from microtensor_subnet.validator.context import ( ESCALATION, + HARNESS, ROUTER, SLM, + SYSTEM, Artifact, EpochContext, Reference, @@ -228,6 +238,7 @@ class Runner: reference_router: RouterRun | None = None reference_escalation: Any = None lock: threading.Lock = field(default_factory=threading.Lock) + reporter: progress.Reporter = field(default_factory=progress.Reporter) def _memory(self, model_dir: Path, config: Mapping[str, Any]) -> int: slm = config.get("slm") or {} @@ -405,6 +416,7 @@ def slm_phase(self, artifact: Artifact, slot: int, gpu: str) -> None: SlmClient(sandbox.slm_url), memory_bytes=self._memory(artifact.slm_dir, artifact.config), parameters=artifact.slm_parameters, + progress=self.reporter.tracker(artifact.hotkey, SLM), ) outcome.slm.vram_peak_bytes = self._measured(sandbox, outcome.slm.vram_peak_bytes) finally: @@ -438,7 +450,11 @@ def full_phase(self, artifact: Artifact, slot: int, gpu: str) -> None: try: if "harness" in artifact.has: outcome.harness = harness.run( - self.context, self.host, sandbox, timeout=self.config.request_seconds + self.context, + self.host, + sandbox, + timeout=self.config.request_seconds, + progress=self.reporter.tracker(artifact.hotkey, HARNESS), ) self.host.restart_main(sandbox, artifact_dir) if "escalation" in artifact.has and self.reference_run is not None: @@ -454,6 +470,7 @@ def full_phase(self, artifact: Artifact, slot: int, gpu: str) -> None: reference=self.reference_run, timeout=self.config.request_seconds, token_price=token_price, + progress=self.reporter.tracker(artifact.hotkey, ESCALATION), ) if self.reference_run is not None: outcome.system = system.run( @@ -464,6 +481,7 @@ def full_phase(self, artifact: Artifact, slot: int, gpu: str) -> None: reference=self.reference_run, timeout=self.config.request_seconds, token_price=token_price, + progress=self.reporter.tracker(artifact.hotkey, SYSTEM), ) finally: outcome.usage = self.host.close(sandbox) @@ -494,6 +512,7 @@ def router_run(self, record: MinerRecord, sources: Mapping[str, Any]) -> RouterR self.gpus.give(slot, gpu) def close(self) -> None: + self.reporter.close() self.host.services.stop() if self.agent is not None: self.agent.stop() @@ -566,6 +585,9 @@ def set_up_baselines( base_dirs=base_dirs, known_public=provenance.public_checkpoint_hashes(config.home / "public_checkpoints.json"), reference_slm_dir=reference_dir, + reporter=progress.Reporter( + client, config.arena_id, context.epoch_index, enabled=not config.dry_run + ), protected={ context.reference.slm_dir, *([reference_dir] if reference_dir is not None else []), @@ -734,6 +756,16 @@ def build_queue( return queue, failures +def announce( + runner: Runner, queue: Sequence[collect.Candidate], failures: Mapping[str, str] +) -> None: + for position, candidate in enumerate(queue): + runner.reporter.post(candidate.hotkey, progress.QUEUED, position + 1, len(queue)) + for hotkey, reason in failures.items(): + if hotkey in runner.context.roster: + runner.reporter.post(hotkey, progress.REFUSED, note=reason) + + def miner_loop(runner: Runner, queue: Sequence[collect.Candidate]) -> None: workers = ( 1 if runner.limits.on_cpu else max(1, min(runner.config.concurrency, runner.slot_count)) @@ -759,6 +791,7 @@ def process_miner(runner: Runner, position: int, candidate: collect.Candidate) - record.drafts = outcome.slm.drafts() if outcome.slm is not None else None record.failure = outcome.failure if not outcome.failure: + runner.reporter.post(candidate.hotkey, progress.SCORING) record.scores = score_miner(runner, artifact, outcome) record.provenance = dict(outcome.provenance) except Exception as exc: @@ -781,6 +814,7 @@ def hold(path: Path) -> None: runner.holding[candidate.hotkey] = (*runner.holding.get(candidate.hotkey, ()), path) try: + runner.reporter.post(candidate.hotkey, progress.FETCHING) artifact = collect.materialise( candidate, cache_dir=runner.config.cache_dir, @@ -791,6 +825,7 @@ def hold(path: Path) -> None: formats=[str(f) for f in definition.get("allowed_formats") or []], hold=hold, ) + runner.reporter.post(candidate.hotkey, progress.CHECKING) with runner.lock: runner.holding[candidate.hotkey] = _paths_of(artifact) record.artifact = artifact @@ -955,6 +990,8 @@ def post_result(runner: Runner, record: MinerRecord) -> bool: log.warning("%s result not accepted, kept for the retry at close: %s", record.hotkey, exc) return False record.posted = True + stage = progress.REFUSED if record.failure else progress.SENT + runner.reporter.post(record.hotkey, stage, note=record.failure) return True @@ -1113,6 +1150,7 @@ def run_epoch( try: queue, failures = build_queue(config, chain, context) log.info("epoch %d: %d queued, %d refused", epoch_index, len(queue), len(failures)) + announce(runner, queue, failures) miner_loop(runner, queue) router_pass(runner) retry_unposted(runner) diff --git a/microtensor_subnet/validator/pools/escalation.py b/microtensor_subnet/validator/pools/escalation.py index d6b896c..f17b0e6 100644 --- a/microtensor_subnet/validator/pools/escalation.py +++ b/microtensor_subnet/validator/pools/escalation.py @@ -7,6 +7,7 @@ from microtensor_subnet.artifact import contract from microtensor_subnet.core.items import Item from microtensor_subnet.validator.context import ESCALATION, EpochContext +from microtensor_subnet.validator.progress import Progress, Tally from microtensor_subnet.validator.relay import REPAIR, RelayError, TicketDesk from microtensor_subnet.validator.results import EscalationRun, ItemRun, ReferenceRun from microtensor_subnet.validator.sandbox import Sandbox, call_main @@ -39,10 +40,12 @@ def run( reference: ReferenceRun, timeout: float, token_price: float, + progress: Progress | None = None, ) -> EscalationRun: repairs: list[ItemRun] = [] session = sandbox.session - for item, draft in drafts: + tally = Tally(progress, len(drafts)) + for item, draft in tally.over(drafts): redo = reference.redo_answers.get(item.item_id) redo_cost = redo.cost_micros if redo is not None else 0 cap = max(1, int(redo_cost * TICKET_MULTIPLE)) diff --git a/microtensor_subnet/validator/pools/harness.py b/microtensor_subnet/validator/pools/harness.py index 0df070f..dc49bcc 100644 --- a/microtensor_subnet/validator/pools/harness.py +++ b/microtensor_subnet/validator/pools/harness.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging -from collections.abc import Sequence +from collections.abc import Iterable, Sequence from pathlib import Path from typing import Any @@ -9,6 +9,7 @@ from microtensor_subnet.core.items import Item from microtensor_subnet.scoring.stats import percentile from microtensor_subnet.validator.context import HARNESS, EpochContext, render_prompt +from microtensor_subnet.validator.progress import Progress, Tally from microtensor_subnet.validator.results import Condition, HarnessRun, ItemRun from microtensor_subnet.validator.sandbox import Sandbox, SandboxHost, call_main from microtensor_subnet.validator.sandbox.slm import SlmClient, SlmError @@ -47,7 +48,7 @@ def run_condition( context: EpochContext, sandbox: Sandbox, name: str, - items: Sequence[Item], + items: Iterable[Item], *, timeout: float, ) -> Condition: @@ -76,7 +77,7 @@ def run_condition( def run_bare( context: EpochContext, client: SlmClient, - items: Sequence[Item], + items: Iterable[Item], *, max_tokens: int | None = None, ) -> Condition: @@ -102,20 +103,24 @@ def run( *, items: Sequence[Item] | None = None, timeout: float, + progress: Progress | None = None, ) -> HarnessRun: chosen = list(items) if items is not None else context.items.items(HARNESS) - miner = run_condition(context, sandbox, "H", chosen, timeout=timeout) - baselines = [run_bare(context, SlmClient(sandbox.slm_url), chosen)] + variants: list[Path] = [context.reference.harness_dir, *context.reference.variant_dirs] + tally = Tally(progress, len(chosen) * (len(variants) + 2)) + miner = run_condition(context, sandbox, "H", tally.over(chosen), timeout=timeout) + baselines = [run_bare(context, SlmClient(sandbox.slm_url), tally.over(chosen))] limit = float(context.harness_limits.get("added_wall_p95_ms") or 0) added = percentile(miner.added_ms, 0.95) limits_ok = not limit or added <= limit reason = "" if limits_ok else f"added wall time p95 {added:.0f} ms is over {limit:.0f} ms" - variants: list[Path] = [context.reference.harness_dir, *context.reference.variant_dirs] for index, variant in enumerate(variants, start=1): try: host.restart_main(sandbox, variant) except Exception as exc: log.warning("reference harness %d could not start: %s", index, exc) continue - baselines.append(run_condition(context, sandbox, f"B{index}", chosen, timeout=timeout)) + baselines.append( + run_condition(context, sandbox, f"B{index}", tally.over(chosen), timeout=timeout) + ) return HarnessRun(miner=miner, baselines=baselines, limits_ok=limits_ok, limits_reason=reason) diff --git a/microtensor_subnet/validator/pools/slm.py b/microtensor_subnet/validator/pools/slm.py index 7c69151..74f35fc 100644 --- a/microtensor_subnet/validator/pools/slm.py +++ b/microtensor_subnet/validator/pools/slm.py @@ -6,6 +6,7 @@ from microtensor_subnet.core.items import Item from microtensor_subnet.scoring.stats import percentile from microtensor_subnet.validator.context import SLM, EpochContext, render_prompt +from microtensor_subnet.validator.progress import Progress, Tally from microtensor_subnet.validator.results import ItemRun, SlmRun from microtensor_subnet.validator.sandbox.slm import SlmClient, SlmError @@ -42,13 +43,15 @@ def run( parameters: int, max_tokens: int | None = None, items: Sequence[Item] | None = None, + progress: Progress | None = None, ) -> SlmRun: budget = max_tokens or context.max_tokens chosen = list(items) if items is not None else context.items.originals(SLM) twins = context.items.twins_of(SLM) runs: list[ItemRun] = [] twin_runs: list[ItemRun] = [] - for item in chosen: + tally = Tally(progress, len(chosen)) + for item in tally.over(chosen): runs.append(_run_item(context, client, item, budget)) twin = twins.get(item.item_id) if twin is not None: diff --git a/microtensor_subnet/validator/pools/system.py b/microtensor_subnet/validator/pools/system.py index 956976a..99842b6 100644 --- a/microtensor_subnet/validator/pools/system.py +++ b/microtensor_subnet/validator/pools/system.py @@ -6,6 +6,7 @@ from microtensor_subnet.artifact import contract from microtensor_subnet.core.items import Item from microtensor_subnet.validator.context import SYSTEM, EpochContext +from microtensor_subnet.validator.progress import Progress, Tally from microtensor_subnet.validator.relay import SYSTEM as SYSTEM_PURPOSE from microtensor_subnet.validator.relay import RelayError, TicketDesk from microtensor_subnet.validator.results import FloorRun, ItemRun, ReferenceRun, SystemRun @@ -25,13 +26,15 @@ def run( timeout: float, token_price: float, items: Sequence[Item] | None = None, + progress: Progress | None = None, ) -> SystemRun: chosen = list(items) if items is not None else context.items.items(SYSTEM) session = sandbox.session floors: list[FloorRun] = [] + tally = Tally(progress, len(context.floors) * len(chosen)) for floor in context.floors: runs: list[ItemRun] = [] - for item in chosen: + for item in tally.over(chosen): redo = reference.system_answers.get(item.item_id) cap = max(1, int((redo.cost_micros if redo else 0) * TICKET_MULTIPLE)) try: diff --git a/microtensor_subnet/validator/progress.py b/microtensor_subnet/validator/progress.py new file mode 100644 index 0000000..8f0f6ee --- /dev/null +++ b/microtensor_subnet/validator/progress.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import logging +import threading +import time +from collections import deque +from collections.abc import Callable, Iterable, Iterator +from dataclasses import dataclass +from typing import Any, Final, TypeVar + +log = logging.getLogger("microtensor_subnet.validator.progress") + +QUEUED: Final[str] = "queued" +FETCHING: Final[str] = "fetching" +CHECKING: Final[str] = "checking" +SCORING: Final[str] = "scoring" +SENT: Final[str] = "sent" +REFUSED: Final[str] = "refused" +THROTTLE_SECONDS: Final[float] = 15.0 +QUEUE_LIMIT: Final[int] = 256 +NOTE_LIMIT: Final[int] = 300 + +Progress = Callable[[int, int], None] +T = TypeVar("T") + + +@dataclass(slots=True) +class Tally: + progress: Progress | None + total: int + done: int = 0 + + def __post_init__(self) -> None: + self._emit() + + def over(self, rows: Iterable[T]) -> Iterator[T]: + for row in rows: + yield row + self.done += 1 + self._emit() + + def _emit(self) -> None: + if self.progress is not None: + self.progress(self.done, self.total) + + +class Reporter: + def __init__( + self, client: Any = None, arena_id: str = "", epoch_index: int = 0, *, enabled: bool = True + ) -> None: + self._client = client if enabled else None + self._arena_id = arena_id + self._epoch_index = epoch_index + self._pending: deque[dict[str, Any]] = deque(maxlen=QUEUE_LIMIT) + self._last: dict[str, tuple[str, float]] = {} + self._wake = threading.Condition() + self._worker: threading.Thread | None = None + self._closed = False + + def post(self, miner: str, stage: str, done: int = 0, total: int = 0, note: str = "") -> None: + if self._client is None: + return + try: + self._enqueue(miner, stage, max(0, done), max(0, done, total), note) + except Exception as exc: + log.debug("progress for %s not queued: %s", miner, exc) + + def tracker(self, miner: str, stage: str) -> Progress: + return lambda done, total: self.post(miner, stage, done, total) + + def close(self) -> None: + with self._wake: + self._closed = True + self._wake.notify_all() + + def _enqueue(self, miner: str, stage: str, done: int, total: int, note: str) -> None: + now = time.monotonic() + with self._wake: + if self._closed: + return + last = self._last.get(miner) + if last and last[0] == stage and done < total and now - last[1] < THROTTLE_SECONDS: + return + self._last[miner] = (stage, now) + self._pending.append( + { + "arena_id": self._arena_id, + "epoch_index": self._epoch_index, + "miner": miner, + "stage": stage, + "done": done, + "total": total, + "note": note[:NOTE_LIMIT], + } + ) + if self._worker is None: + self._worker = threading.Thread(target=self._drain, name="progress", daemon=True) + self._worker.start() + self._wake.notify_all() + + def _drain(self) -> None: + while True: + with self._wake: + self._wake.wait_for(lambda: bool(self._pending) or self._closed) + if not self._pending: + return + body = self._pending.popleft() + try: + self._client.report_progress(body) + except Exception as exc: + log.debug("progress for %s not sent: %s", body["miner"], exc) From e89a0dbfbcf84df06a95878192d791a2bf150f52 Mon Sep 17 00:00:00 2001 From: BANADDA MUBARAKA <83466862+BANADDA@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:27:14 +0300 Subject: [PATCH 2/3] validator: type the progress reporter's client as Any --- microtensor_subnet/validator/progress.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/microtensor_subnet/validator/progress.py b/microtensor_subnet/validator/progress.py index 8f0f6ee..4294f47 100644 --- a/microtensor_subnet/validator/progress.py +++ b/microtensor_subnet/validator/progress.py @@ -48,7 +48,7 @@ class Reporter: def __init__( self, client: Any = None, arena_id: str = "", epoch_index: int = 0, *, enabled: bool = True ) -> None: - self._client = client if enabled else None + self._client: Any = client if enabled else None self._arena_id = arena_id self._epoch_index = epoch_index self._pending: deque[dict[str, Any]] = deque(maxlen=QUEUE_LIMIT) From 3c7e7209053bb594558041bef80d5848d94c55bc Mon Sep 17 00:00:00 2001 From: BANADDA MUBARAKA <83466862+BANADDA@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:27:37 +0300 Subject: [PATCH 3/3] validator: announce the queue before setup and report every pool against its published size --- microtensor_subnet/validator/collect.py | 3 ++ microtensor_subnet/validator/context.py | 5 +++ microtensor_subnet/validator/epoch.py | 38 +++++++++++++------ .../validator/pools/escalation.py | 2 +- microtensor_subnet/validator/pools/harness.py | 3 +- microtensor_subnet/validator/pools/slm.py | 2 +- microtensor_subnet/validator/pools/system.py | 2 +- microtensor_subnet/validator/progress.py | 18 +++++++-- 8 files changed, 54 insertions(+), 19 deletions(-) diff --git a/microtensor_subnet/validator/collect.py b/microtensor_subnet/validator/collect.py index bf6e13c..96ffc3b 100644 --- a/microtensor_subnet/validator/collect.py +++ b/microtensor_subnet/validator/collect.py @@ -178,6 +178,7 @@ def materialise( bounds: tuple[int, int] | None = None, formats: list[str] | None = None, hold: Callable[[Path], None] | None = None, + on_downloaded: Callable[[], None] | None = None, ) -> Artifact: commitment = candidate.commitment if hold is not None: @@ -186,6 +187,8 @@ def materialise( artifact_dir = download(commitment.artifact_repo, commitment.artifact_sha, cache_dir) except Exception as exc: raise CollectError(f"artifact repository could not be fetched: {exc}") from exc + if on_downloaded is not None: + on_downloaded() try: manifest = load_manifest(artifact_dir) except ManifestError as exc: diff --git a/microtensor_subnet/validator/context.py b/microtensor_subnet/validator/context.py index 3222f2c..1eb49e5 100644 --- a/microtensor_subnet/validator/context.py +++ b/microtensor_subnet/validator/context.py @@ -110,6 +110,11 @@ class EpochContext: def grader(self) -> str: return str(self.definition.get("grader") or "sqlite_execution") + def pool_size(self, test: str) -> int: + counts = self.definition.get("item_counts") + found = counts.get(test) if isinstance(counts, Mapping) else None + return int(found or 0) or len(self.items.originals(test)) + @property def stop(self) -> tuple[str, ...]: block = self.definition.get("generation") or {} diff --git a/microtensor_subnet/validator/epoch.py b/microtensor_subnet/validator/epoch.py index 0b4cc75..0e971b6 100644 --- a/microtensor_subnet/validator/epoch.py +++ b/microtensor_subnet/validator/epoch.py @@ -401,6 +401,7 @@ def slm_phase(self, artifact: Artifact, slot: int, gpu: str) -> None: outcome = self.outcomes.setdefault(artifact.hotkey, ArtifactOutcome(hotkey=artifact.hotkey)) if artifact.slm_dir is None: return + self.reporter.post(artifact.hotkey, SLM, 0, self.context.pool_size(SLM)) try: sandbox = self._open( self.context.reference.harness_dir, @@ -527,6 +528,7 @@ def set_up_baselines( *, wallet: Any, hotkey: str, + reporter: progress.Reporter | None = None, ) -> Runner: limits = Limits.from_definition(context.definition, context.max_model_len) placement: probe_judge.Placement | None = None @@ -585,9 +587,7 @@ def set_up_baselines( base_dirs=base_dirs, known_public=provenance.public_checkpoint_hashes(config.home / "public_checkpoints.json"), reference_slm_dir=reference_dir, - reporter=progress.Reporter( - client, config.arena_id, context.epoch_index, enabled=not config.dry_run - ), + reporter=reporter or progress.Reporter(), protected={ context.reference.slm_dir, *([reference_dir] if reference_dir is not None else []), @@ -757,13 +757,16 @@ def build_queue( def announce( - runner: Runner, queue: Sequence[collect.Candidate], failures: Mapping[str, str] + reporter: progress.Reporter, + context: EpochContext, + queue: Sequence[collect.Candidate], + failures: Mapping[str, str], ) -> None: for position, candidate in enumerate(queue): - runner.reporter.post(candidate.hotkey, progress.QUEUED, position + 1, len(queue)) + reporter.post(candidate.hotkey, progress.QUEUED, position + 1, len(queue)) for hotkey, reason in failures.items(): - if hotkey in runner.context.roster: - runner.reporter.post(hotkey, progress.REFUSED, note=reason) + if hotkey in context.roster: + reporter.post(hotkey, progress.REFUSED, note=reason) def miner_loop(runner: Runner, queue: Sequence[collect.Candidate]) -> None: @@ -824,8 +827,8 @@ def hold(path: Path) -> None: bounds=_bounds(definition), formats=[str(f) for f in definition.get("allowed_formats") or []], hold=hold, + on_downloaded=lambda: runner.reporter.post(candidate.hotkey, progress.CHECKING), ) - runner.reporter.post(candidate.hotkey, progress.CHECKING) with runner.lock: runner.holding[candidate.hotkey] = _paths_of(artifact) record.artifact = artifact @@ -1144,13 +1147,26 @@ def run_epoch( hotkey = hotkey_address(wallet) client = ServerClient(config.server_url, hotkey, wallet) context = load_context(config, client, chain, epoch_index=epoch_index, hotkey=hotkey) - runner = set_up_baselines( - config, client, context, docker or Docker(), wallet=wallet, hotkey=hotkey + reporter = progress.Reporter( + client, config.arena_id, context.epoch_index, enabled=not config.dry_run ) try: queue, failures = build_queue(config, chain, context) log.info("epoch %d: %d queued, %d refused", epoch_index, len(queue), len(failures)) - announce(runner, queue, failures) + announce(reporter, context, queue, failures) + runner = set_up_baselines( + config, + client, + context, + docker or Docker(), + wallet=wallet, + hotkey=hotkey, + reporter=reporter, + ) + except BaseException: + reporter.close() + raise + try: miner_loop(runner, queue) router_pass(runner) retry_unposted(runner) diff --git a/microtensor_subnet/validator/pools/escalation.py b/microtensor_subnet/validator/pools/escalation.py index f17b0e6..55bfc9f 100644 --- a/microtensor_subnet/validator/pools/escalation.py +++ b/microtensor_subnet/validator/pools/escalation.py @@ -44,7 +44,7 @@ def run( ) -> EscalationRun: repairs: list[ItemRun] = [] session = sandbox.session - tally = Tally(progress, len(drafts)) + tally = Tally(progress, context.pool_size(ESCALATION), len(drafts)) for item, draft in tally.over(drafts): redo = reference.redo_answers.get(item.item_id) redo_cost = redo.cost_micros if redo is not None else 0 diff --git a/microtensor_subnet/validator/pools/harness.py b/microtensor_subnet/validator/pools/harness.py index dc49bcc..e6fd269 100644 --- a/microtensor_subnet/validator/pools/harness.py +++ b/microtensor_subnet/validator/pools/harness.py @@ -107,7 +107,7 @@ def run( ) -> HarnessRun: chosen = list(items) if items is not None else context.items.items(HARNESS) variants: list[Path] = [context.reference.harness_dir, *context.reference.variant_dirs] - tally = Tally(progress, len(chosen) * (len(variants) + 2)) + tally = Tally(progress, context.pool_size(HARNESS), len(chosen) * (len(variants) + 2)) miner = run_condition(context, sandbox, "H", tally.over(chosen), timeout=timeout) baselines = [run_bare(context, SlmClient(sandbox.slm_url), tally.over(chosen))] limit = float(context.harness_limits.get("added_wall_p95_ms") or 0) @@ -119,6 +119,7 @@ def run( host.restart_main(sandbox, variant) except Exception as exc: log.warning("reference harness %d could not start: %s", index, exc) + tally.advance(len(chosen)) continue baselines.append( run_condition(context, sandbox, f"B{index}", tally.over(chosen), timeout=timeout) diff --git a/microtensor_subnet/validator/pools/slm.py b/microtensor_subnet/validator/pools/slm.py index 74f35fc..4c3fa3f 100644 --- a/microtensor_subnet/validator/pools/slm.py +++ b/microtensor_subnet/validator/pools/slm.py @@ -50,7 +50,7 @@ def run( twins = context.items.twins_of(SLM) runs: list[ItemRun] = [] twin_runs: list[ItemRun] = [] - tally = Tally(progress, len(chosen)) + tally = Tally(progress, context.pool_size(SLM), len(chosen)) for item in tally.over(chosen): runs.append(_run_item(context, client, item, budget)) twin = twins.get(item.item_id) diff --git a/microtensor_subnet/validator/pools/system.py b/microtensor_subnet/validator/pools/system.py index 99842b6..01c799b 100644 --- a/microtensor_subnet/validator/pools/system.py +++ b/microtensor_subnet/validator/pools/system.py @@ -31,7 +31,7 @@ def run( chosen = list(items) if items is not None else context.items.items(SYSTEM) session = sandbox.session floors: list[FloorRun] = [] - tally = Tally(progress, len(context.floors) * len(chosen)) + tally = Tally(progress, context.pool_size(SYSTEM), len(context.floors) * len(chosen)) for floor in context.floors: runs: list[ItemRun] = [] for item in tally.over(chosen): diff --git a/microtensor_subnet/validator/progress.py b/microtensor_subnet/validator/progress.py index 4294f47..9f55dbc 100644 --- a/microtensor_subnet/validator/progress.py +++ b/microtensor_subnet/validator/progress.py @@ -28,7 +28,9 @@ class Tally: progress: Progress | None total: int + steps: int done: int = 0 + shown: int = -1 def __post_init__(self) -> None: self._emit() @@ -36,12 +38,20 @@ def __post_init__(self) -> None: def over(self, rows: Iterable[T]) -> Iterator[T]: for row in rows: yield row - self.done += 1 - self._emit() + self.advance(1) + + def advance(self, count: int) -> None: + self.done += count + self._emit() def _emit(self) -> None: - if self.progress is not None: - self.progress(self.done, self.total) + if self.progress is None: + return + total = max(0, self.total) + shown = total if self.done >= self.steps else self.done * total // self.steps + if shown != self.shown: + self.shown = shown + self.progress(shown, total) class Reporter: