CoolBoy/submmit
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

perf: complete history bootstrap in one streaming phase

Browse Source
This commit is contained in:
CoolBoy
2026-09-04 11:35:43 +08:00
parent 54ed4351a0
commit 2c748dcec2
6 changed files with 183 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files

326
modelhub_submmit_api/poll_runner.py
View File

@@ -6,8 +6,6 @@ import os
import sys
import time
from collections import deque
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import timedelta
from pathlib import Path
from typing import Any, Callable
@@ -334,200 +332,6 @@ def _load_task_compatibility_contexts(
return contexts
def _load_bounded_owned_history(
modelhub_client: ModelHubClient | ModelHubClientPool,
*,
max_records_per_account: int,
) -> tuple[list[dict[str, Any]], list[int]]:
clients = (
list(modelhub_client.clients)
if isinstance(modelhub_client, ModelHubClientPool)
else [modelhub_client]
)
by_account: dict[int, list[dict[str, Any]]] = {}
errors: list[int] = []
with ThreadPoolExecutor(max_workers=min(12, max(1, len(clients)))) as executor:
futures = {
executor.submit(
client.list_tasks,
page_size=100,
only_mine=True,
max_records=max_records_per_account,
): index
for index, client in enumerate(clients, start=1)
}
for future in as_completed(futures):
account_index = futures[future]
try:
by_account[account_index] = future.result()
except Exception:
errors.append(account_index)
deduped: dict[str, dict[str, Any]] = {}
anonymous: list[dict[str, Any]] = []
for account_index in sorted(by_account):
for task in by_account[account_index]:
if not isinstance(task, dict):
continue
task_id = task.get("taskId")
if task_id is None:
anonymous.append(task)
continue
deduped[str(task_id)] = task
return [*deduped.values(), *anonymous], sorted(errors)
def _bootstrap_architecture_history(
*,
modelhub_client: ModelHubClient | ModelHubClientPool,
outcome_tracker: OutcomeTracker,
ledger_path: Path,
now,
) -> dict[str, Any]:
"""Prefer bounded recent public evidence, then bounded owned history."""
probe_size = _env_int(
"MODELHUB_ARCHITECTURE_COMMUNITY_PROBE_SIZE",
50,
minimum=1,
maximum=100,
)
community_records: list[dict[str, Any]] = []
community_probe_error: str | None = None
try:
probe_payload = modelhub_client.list_tasks_page(
current=1,
page_size=probe_size,
only_mine=False,
status="success",
verify_result=-1,
)
probe_records = (probe_payload.get("data") or {}).get("records") or []
community_records = [record for record in probe_records if isinstance(record, dict)]
except Exception as exc:
community_probe_error = f"{type(exc).__name__}: {exc}"
community_usable = [
record
for record in community_records
if record.get("logCosUrl")
and record.get("modelId")
and record.get("gpuType")
]
log(
f"[architecture-bootstrap] community_probe={len(community_records)} "
f"usable_failure_details={len(community_usable)} "
f"error={'none' if community_probe_error is None else community_probe_error}"
)
source = "community_latest"
listing_errors: list[int] = []
if community_usable:
lookback_days = _env_int(
"MODELHUB_ARCHITECTURE_COMMUNITY_LOOKBACK_DAYS",
30,
minimum=1,
maximum=365,
)
latest_limit = _env_int(
"MODELHUB_ARCHITECTURE_COMMUNITY_LATEST_LIMIT",
100,
minimum=1,
maximum=50_000,
)
try:
history_tasks = modelhub_client.list_tasks(
page_size=100,
only_mine=False,
begin_time=now - timedelta(days=lookback_days),
end_time=now,
status="success",
verify_result=-1,
max_records=latest_limit,
)
history_tasks = [task for task in history_tasks if task.get("logCosUrl")]
log(
f"[architecture-bootstrap] source=community_latest "
f"lookback_days={lookback_days} records={len(history_tasks)} limit={latest_limit}"
)
except Exception as exc:
source = "owned_recent_bounded"
log(
f"[architecture-bootstrap] community_history_error={type(exc).__name__}: {exc} "
"fallback=owned_recent_bounded"
)
history_tasks, listing_errors = _load_bounded_owned_history(
modelhub_client,
max_records_per_account=_env_int(
"MODELHUB_ARCHITECTURE_BOOTSTRAP_TASKS_PER_ACCOUNT",
10,
minimum=10,
maximum=500,
),
)
else:
source = "owned_recent_bounded"
per_account_limit = _env_int(
"MODELHUB_ARCHITECTURE_BOOTSTRAP_TASKS_PER_ACCOUNT",
10,
minimum=10,
maximum=500,
)
history_tasks, listing_errors = _load_bounded_owned_history(
modelhub_client,
max_records_per_account=per_account_limit,
)
account_count = (
len(modelhub_client.clients)
if isinstance(modelhub_client, ModelHubClientPool)
else 1
)
log(
f"[architecture-bootstrap] source=owned_recent_bounded records={len(history_tasks)} "
f"accounts={account_count} per_account_limit={per_account_limit} "
f"listing_errors={','.join(map(str, listing_errors)) or 'none'}"
)
contexts = _load_task_compatibility_contexts(
outcome_tracker,
ledger_path=ledger_path,
)
summary = outcome_tracker.bootstrap_from_history_tasks(
history_tasks,
task_contexts=contexts,
enrichment_limit=_env_int(
"MODELHUB_ARCHITECTURE_BOOTSTRAP_MAX_LOGS",
120,
minimum=0,
maximum=100_000,
),
enrichment_workers=_env_int(
"MODELHUB_ARCHITECTURE_BOOTSTRAP_WORKERS",
8,
minimum=1,
maximum=16,
),
log=log,
)
feedback = outcome_tracker.get_stats_report()
block_count = len(feedback.get("architectureCompatibilityBlocks") or {})
summary.update(
{
"source": source,
"communityProbeRecords": len(community_records),
"communityUsableFailureDetails": len(community_usable),
"listingErrorAccounts": listing_errors,
"architectureBlocks": block_count,
}
)
log(
f"[architecture-bootstrap] finished source={source} "
f"terminal={summary['terminalRecords']} failure_logs={summary['eligibleFailureLogs']} "
f"explicit_architecture_failures={summary['explicitArchitectureFailures']} "
f"recovered_frameworks={summary['recoveredFrameworks']} blocks={block_count}"
)
return summary
def _architecture_backfill_clients(
modelhub_client: ModelHubClient | ModelHubClientPool,
) -> list[ModelHubClient]:
@@ -620,20 +424,20 @@ def _advance_architecture_history_backfill(
page_size = _env_int(
"MODELHUB_ARCHITECTURE_BACKFILL_PAGE_SIZE",
50,
100,
minimum=10,
maximum=100,
)
pages_per_cycle = _env_int(
"MODELHUB_ARCHITECTURE_BACKFILL_PAGES_PER_CYCLE",
2,
pages_per_batch = _env_int(
"MODELHUB_ARCHITECTURE_BACKFILL_PAGES_PER_BATCH",
10,
minimum=1,
maximum=12,
maximum=20,
)
max_logs = _env_int(
"MODELHUB_ARCHITECTURE_BACKFILL_LOGS_PER_CYCLE",
page_size * pages_per_cycle,
minimum=1,
log_batch_size = _env_int(
"MODELHUB_ARCHITECTURE_BACKFILL_LOG_BATCH_SIZE",
200,
minimum=10,
maximum=500,
)
accounts = progress.get("accounts") or {}
@@ -642,7 +446,7 @@ def _advance_architecture_history_backfill(
page_calls = 0
page_errors = 0
for _ in range(pages_per_cycle):
for _ in range(pages_per_batch):
incomplete = [
index
for index in range(len(clients))
@@ -669,6 +473,7 @@ def _advance_architecture_history_backfill(
data = payload.get("data") or {}
records = [item for item in (data.get("records") or []) if isinstance(item, dict)]
pages = max(0, int(data.get("pages") or 0))
state.pop("lastError", None)
state["recordsScanned"] = int(state.get("recordsScanned") or 0) + len(records)
progress["recordsScanned"] = int(progress.get("recordsScanned") or 0) + len(records)
new_count = 0
@@ -695,6 +500,7 @@ def _advance_architecture_history_backfill(
page_errors += 1
state["listingErrors"] = int(state.get("listingErrors") or 0) + 1
state["lastError"] = f"{type(exc).__name__}: {exc}"[:500]
break
batch_summary: dict[str, Any] = {
"terminalRecords": 0,
@@ -708,13 +514,17 @@ def _advance_architecture_history_backfill(
outcome_tracker,
ledger_path=ledger_path,
),
enrichment_limit=max_logs,
# Every usable failure in this metadata batch is inspected. The
# batch boundary bounds memory; enrichment_batch_size bounds the
# number of retained log jobs at one time.
enrichment_limit=0,
enrichment_workers=_env_int(
"MODELHUB_ARCHITECTURE_BOOTSTRAP_WORKERS",
8,
minimum=1,
maximum=16,
),
enrichment_batch_size=log_batch_size,
log=log,
)
# Each slice is immediately reduced to cumulative counters, routing
@@ -753,6 +563,66 @@ def _advance_architecture_history_backfill(
return progress
def _complete_architecture_history_backfill(
*,
modelhub_client: ModelHubClient | ModelHubClientPool,
outcome_tracker: OutcomeTracker,
ledger_path: Path,
progress_path: Path = DEFAULT_ARCHITECTURE_BACKFILL_PATH,
sync_callback: Callable[[str], bool] | None = None,
) -> dict[str, Any]:
"""Finish the one-time scan in bounded batches after the first submit pass."""
checkpoint_interval = _env_int(
"MODELHUB_ARCHITECTURE_BACKFILL_CHECKPOINT_RECORDS",
2000,
minimum=500,
maximum=10_000,
)
scanned_since_sync = 0
try:
progress = read_json(progress_path)
except (FileNotFoundError, ValueError, TypeError):
progress = _new_architecture_backfill_progress(
modelhub_client,
now=utc_now(),
)
write_json(progress_path, progress)
while not bool(progress.get("complete")):
before_scanned = int(progress.get("recordsScanned") or 0)
progress = _advance_architecture_history_backfill(
modelhub_client=modelhub_client,
outcome_tracker=outcome_tracker,
ledger_path=ledger_path,
progress_path=progress_path,
)
scanned_now = int(progress.get("recordsScanned") or 0)
scanned_delta = max(0, scanned_now - before_scanned)
scanned_since_sync += scanned_delta
should_sync = bool(progress.get("complete")) or scanned_since_sync >= checkpoint_interval
if should_sync and sync_callback is not None:
if not sync_callback("history_backfill"):
log(
f"[architecture-backfill] status=paused reason=state_sync_failed "
f"records_since_checkpoint={scanned_since_sync}"
)
break
log(
f"[architecture-backfill] checkpoint=durable "
f"records_total={scanned_now} complete={str(bool(progress.get('complete'))).lower()}"
)
scanned_since_sync = 0
# A page/API error must not create a hot loop. The normal poll loop will
# retry the same durable cursor on its next cycle.
if scanned_delta <= 0 and not bool(progress.get("complete")):
log("[architecture-backfill] status=deferred reason=no_scan_progress retry_next_cycle=true")
break
return progress
def run_poll_loop(
*,
base_args: argparse.Namespace,
@@ -868,17 +738,15 @@ def run_poll_loop(
f"history_backfill={architecture_bootstrap_summary['historyBackfill']}"
)
else:
architecture_bootstrap_summary = _bootstrap_architecture_history(
modelhub_client=modelhub_client,
outcome_tracker=outcome_tracker,
ledger_path=Path(base_args.ledger_path),
now=now,
)
outcome_tracker.compact_decision_state()
# A missing decision checkpoint means the prior aggregate was
# unavailable. Start a resumable full owned-history backfill;
# the small synchronous seed above only makes the first routing
# decisions useful without delaying submissions.
architecture_bootstrap_summary = {
"source": "cold_start_streaming_backfill",
"terminalRecords": 0,
"architectureBlocks": 0,
"historyBackfill": "pending_after_first_submission_pass",
}
# Progress without its aggregate checkpoint is not usable:
# skipping those pages would under-count history. Reset both
# sides of the resumable scan whenever the checkpoint is gone.
write_json(
architecture_backfill_path,
_new_architecture_backfill_progress(
@@ -886,6 +754,10 @@ def run_poll_loop(
now=now,
),
)
log(
"[architecture-bootstrap] source=cold_start_streaming_backfill "
"startup_history_scan=disabled backfill=after_first_submission_pass"
)
architecture_bootstrap_summary["enabled"] = True
_persist_architecture_blacklist(
outcome_tracker.get_stats_report(),
@@ -927,7 +799,7 @@ def run_poll_loop(
pending_architecture_cleanup = False
last_cleaned_architecture_blocks: set[str] = set()
def advance_architecture_backfill() -> None:
def complete_architecture_backfill() -> None:
nonlocal pending_architecture_cleanup
if getattr(base_args, "skip_outcome_sync", False) or not architecture_backfill_path.is_file():
return
@@ -938,11 +810,16 @@ def run_poll_loop(
previous_blocks = set(
(outcome_tracker.get_stats_report().get("architectureCompatibilityBlocks") or {}).keys()
)
_advance_architecture_history_backfill(
_complete_architecture_history_backfill(
modelhub_client=modelhub_client,
outcome_tracker=outcome_tracker,
ledger_path=Path(base_args.ledger_path),
progress_path=architecture_backfill_path,
sync_callback=(
(lambda phase: state_sync.sync(phase))
if state_sync is not None
else None
),
)
current_feedback = outcome_tracker.get_stats_report()
current_blocks = _persist_architecture_blacklist(
@@ -1165,7 +1042,7 @@ def run_poll_loop(
# Historical learning is deliberately behind queue maintenance and
# capacity checks. It advances even while full, but never blocks
# the worker's initial recovery or first submission attempt.
advance_architecture_backfill()
complete_architecture_backfill()
if state_sync is not None:
try:
if cycles % 3 == 0 and hasattr(modelhub_client, "list_active_tasks_by_account"):
@@ -1205,9 +1082,10 @@ def run_poll_loop(
f"stop={cycle_summary['stoppedReason']}"
)
# Do this only after the submission pass. Each cycle consumes at most a
# small configured slice and immediately reduces it to decision state.
advance_architecture_backfill()
# The first submission pass gets priority. Then the one-time cold-start
# scan runs continuously in bounded memory and disappears permanently
# once its compact decision checkpoint is complete.
complete_architecture_backfill()
if state_sync is not None:
try: