fix: prioritize preferred GPU routes with fallback
This commit is contained in:
CoolBoy
@@ -65,6 +65,7 @@ DEFAULT_RUNS_DIR = Path("runs")
|
||||
DEFAULT_LEDGER_PATH = Path("ledger/submissions.jsonl")
|
||||
ADAPTIVE_SCAN_MAX_MODELS = 3000
|
||||
ADAPTIVE_SCAN_MIN_FALLBACK_MODELS = 500
|
||||
MAX_ROUTES_PER_MODEL = 6
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
@@ -571,54 +572,74 @@ def process_model_for_candidates(
|
||||
skipped: list[dict[str, Any]] = []
|
||||
failed: list[dict[str, Any]] = []
|
||||
allowed_task_types_set = [spec.task_type for spec in specs]
|
||||
pending_task_types_by_gpu: list[tuple[str, list[str]]] = []
|
||||
# Preserve the routing order: the first six GPUs are the preferred
|
||||
# tranche. Lower-ranked GPUs are a fallback only when the preferred
|
||||
# tranche produces no candidate for this model. This keeps queue/quality
|
||||
# prioritization deterministic while still recovering from a model being
|
||||
# already processed, too large, or unsupported on every preferred GPU.
|
||||
preferred_target_gpus = target_gpus[:MAX_ROUTES_PER_MODEL]
|
||||
fallback_target_gpus = target_gpus[MAX_ROUTES_PER_MODEL:]
|
||||
|
||||
route_target_gpus = target_gpus[:6] if official_registry is not None else target_gpus
|
||||
for target_gpu in route_target_gpus:
|
||||
if submission_exclusion_store is not None and submission_exclusion_store.is_blocked(model.repo_id, target_gpu):
|
||||
skipped.append(
|
||||
{"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "model_gpu_uniqueness_blocklist"}
|
||||
)
|
||||
continue
|
||||
if target_gpu in processed_gpus:
|
||||
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "already_processed_for_gpu"})
|
||||
continue
|
||||
if outcome_tracker and outcome_tracker.is_model_gpu_failed(model.repo_id, target_gpu):
|
||||
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "local_failure_cooldown_24h"})
|
||||
continue
|
||||
|
||||
exact_task_types = list(allowed_task_types_set)
|
||||
if official_registry is not None:
|
||||
exact_task_types = official_registry.task_types_for(
|
||||
modelhub_client,
|
||||
model_address=model.model_address,
|
||||
model_last_modified=model.last_modified.isoformat() if model.last_modified else None,
|
||||
gpu=target_gpu,
|
||||
)
|
||||
route_task_types = (
|
||||
list(exact_task_types)
|
||||
if allow_dynamic_tasks
|
||||
else [task_type for task_type in allowed_task_types_set if task_type in exact_task_types]
|
||||
)
|
||||
compatible_task_types = []
|
||||
for task_type in route_task_types:
|
||||
register_dynamic_task_route(task_type, model.pipeline_tag or task_type)
|
||||
if market_intelligence is not None and not market_intelligence.has_framework_route(task_type, target_gpu):
|
||||
try:
|
||||
market_intelligence.ensure_framework_route(modelhub_client, task_type, target_gpu)
|
||||
except Exception as exc:
|
||||
raise OfficialCapabilityUnavailable(
|
||||
f"framework capability unavailable for {task_type}/{target_gpu}: {type(exc).__name__}: {exc}"
|
||||
) from exc
|
||||
if template_selector.supported_frameworks_for_auto(task_type, target_gpu):
|
||||
compatible_task_types.append(task_type)
|
||||
def prepare_pending_routes(route_target_gpus: list[str]) -> list[tuple[str, list[str]]]:
|
||||
pending: list[tuple[str, list[str]]] = []
|
||||
for target_gpu in route_target_gpus:
|
||||
if submission_exclusion_store is not None and submission_exclusion_store.is_blocked(model.repo_id, target_gpu):
|
||||
skipped.append(
|
||||
{"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "model_gpu_uniqueness_blocklist"}
|
||||
)
|
||||
continue
|
||||
if market_intelligence is not None and market_intelligence.has_framework_route(task_type, target_gpu):
|
||||
compatible_task_types.append(task_type)
|
||||
if not compatible_task_types:
|
||||
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "no_compatible_auto_template_or_framework"})
|
||||
continue
|
||||
pending_task_types_by_gpu.append((target_gpu, compatible_task_types))
|
||||
if target_gpu in processed_gpus:
|
||||
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "already_processed_for_gpu"})
|
||||
continue
|
||||
if outcome_tracker and outcome_tracker.is_model_gpu_failed(model.repo_id, target_gpu):
|
||||
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "local_failure_cooldown_24h"})
|
||||
continue
|
||||
|
||||
exact_task_types = list(allowed_task_types_set)
|
||||
if official_registry is not None:
|
||||
exact_task_types = official_registry.task_types_for(
|
||||
modelhub_client,
|
||||
model_address=model.model_address,
|
||||
model_last_modified=model.last_modified.isoformat() if model.last_modified else None,
|
||||
gpu=target_gpu,
|
||||
)
|
||||
route_task_types = (
|
||||
list(exact_task_types)
|
||||
if allow_dynamic_tasks
|
||||
else [task_type for task_type in allowed_task_types_set if task_type in exact_task_types]
|
||||
)
|
||||
compatible_task_types = []
|
||||
for task_type in route_task_types:
|
||||
register_dynamic_task_route(task_type, model.pipeline_tag or task_type)
|
||||
if market_intelligence is not None and not market_intelligence.has_framework_route(task_type, target_gpu):
|
||||
try:
|
||||
market_intelligence.ensure_framework_route(modelhub_client, task_type, target_gpu)
|
||||
except Exception as exc:
|
||||
raise OfficialCapabilityUnavailable(
|
||||
f"framework capability unavailable for {task_type}/{target_gpu}: {type(exc).__name__}: {exc}"
|
||||
) from exc
|
||||
if template_selector.supported_frameworks_for_auto(task_type, target_gpu):
|
||||
compatible_task_types.append(task_type)
|
||||
continue
|
||||
if market_intelligence is not None and market_intelligence.has_framework_route(task_type, target_gpu):
|
||||
compatible_task_types.append(task_type)
|
||||
if not compatible_task_types:
|
||||
skipped.append(
|
||||
{
|
||||
"repoId": model.repo_id,
|
||||
"targetGpu": target_gpu,
|
||||
"reason": "no_compatible_auto_template_or_framework",
|
||||
}
|
||||
)
|
||||
continue
|
||||
pending.append((target_gpu, compatible_task_types))
|
||||
return pending
|
||||
|
||||
pending_task_types_by_gpu = prepare_pending_routes(preferred_target_gpus)
|
||||
used_fallback_routes = False
|
||||
if not pending_task_types_by_gpu and fallback_target_gpus:
|
||||
pending_task_types_by_gpu = prepare_pending_routes(fallback_target_gpus)
|
||||
used_fallback_routes = True
|
||||
|
||||
if not pending_task_types_by_gpu:
|
||||
return candidates, skipped, failed
|
||||
@@ -628,41 +649,56 @@ def process_model_for_candidates(
|
||||
except Exception as exc:
|
||||
return [], skipped, [{"repoId": model.repo_id, "reason": str(exc)}]
|
||||
|
||||
for target_gpu, task_types in pending_task_types_by_gpu:
|
||||
best, preflight_reason = choose_candidate_for_gpu_detailed(
|
||||
model=model,
|
||||
inspection=inspection,
|
||||
template_selector=template_selector,
|
||||
task_types=task_types,
|
||||
target_gpu=target_gpu,
|
||||
market_intelligence=market_intelligence,
|
||||
preflight_advisor=preflight_advisor,
|
||||
require_official_config=official_registry is not None,
|
||||
config_optimizer=config_optimizer,
|
||||
)
|
||||
if best is None:
|
||||
reason = (
|
||||
preflight_reason
|
||||
or (
|
||||
"no_publicly_vetted_compatible_framework"
|
||||
if market_intelligence is not None
|
||||
else "no_compatible_auto_template_or_framework"
|
||||
def evaluate_pending_routes(pending_routes: list[tuple[str, list[str]]]) -> int:
|
||||
before_count = len(candidates)
|
||||
for target_gpu, task_types in pending_routes:
|
||||
if len(candidates) - before_count >= MAX_ROUTES_PER_MODEL:
|
||||
break
|
||||
best, preflight_reason = choose_candidate_for_gpu_detailed(
|
||||
model=model,
|
||||
inspection=inspection,
|
||||
template_selector=template_selector,
|
||||
task_types=task_types,
|
||||
target_gpu=target_gpu,
|
||||
market_intelligence=market_intelligence,
|
||||
preflight_advisor=preflight_advisor,
|
||||
require_official_config=official_registry is not None,
|
||||
config_optimizer=config_optimizer,
|
||||
)
|
||||
if best is None:
|
||||
reason = (
|
||||
preflight_reason
|
||||
or (
|
||||
"no_publicly_vetted_compatible_framework"
|
||||
if market_intelligence is not None
|
||||
else "no_compatible_auto_template_or_framework"
|
||||
)
|
||||
)
|
||||
)
|
||||
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason})
|
||||
continue
|
||||
record = candidate_to_record(best)
|
||||
if outcome_tracker is not None:
|
||||
record["transformersPrerequisiteSatisfied"] = (
|
||||
outcome_tracker.has_non_transformers_failure(model.repo_id)
|
||||
)
|
||||
if market_intelligence is not None:
|
||||
record.update(market_intelligence.gpu_metadata(target_gpu))
|
||||
record.update(market_intelligence.framework_metadata(best.task_type, target_gpu, best.framework))
|
||||
optimization = (record.get("preflightMetadata") or {}).get("configOptimization") or {}
|
||||
if optimization.get("applied"):
|
||||
record["frameworkConfigSource"] = "official_plus_learned_patch"
|
||||
candidates.append(record)
|
||||
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason})
|
||||
continue
|
||||
record = candidate_to_record(best)
|
||||
if outcome_tracker is not None:
|
||||
record["transformersPrerequisiteSatisfied"] = (
|
||||
outcome_tracker.has_non_transformers_failure(model.repo_id)
|
||||
)
|
||||
if market_intelligence is not None:
|
||||
record.update(market_intelligence.gpu_metadata(target_gpu))
|
||||
record.update(market_intelligence.framework_metadata(best.task_type, target_gpu, best.framework))
|
||||
optimization = (record.get("preflightMetadata") or {}).get("configOptimization") or {}
|
||||
if optimization.get("applied"):
|
||||
record["frameworkConfigSource"] = "official_plus_learned_patch"
|
||||
candidates.append(record)
|
||||
return len(candidates) - before_count
|
||||
|
||||
evaluate_pending_routes(pending_task_types_by_gpu)
|
||||
# If preferred GPUs had routes but every one failed candidate preflight,
|
||||
# give the remaining eligible GPUs one chance. Once a
|
||||
# preferred candidate exists, lower-ranked GPUs stay untouched for this
|
||||
# model and remain available for later scans.
|
||||
if not candidates and not used_fallback_routes and fallback_target_gpus:
|
||||
fallback_pending = prepare_pending_routes(fallback_target_gpus)
|
||||
if fallback_pending:
|
||||
evaluate_pending_routes(fallback_pending)
|
||||
|
||||
return candidates, skipped, failed
|
||||
|
||||
@@ -968,7 +1004,7 @@ def one_candidate_per_model(candidates: list[dict[str, Any]]) -> list[dict[str,
|
||||
def limit_routes_per_model(
|
||||
candidates: list[dict[str, Any]],
|
||||
*,
|
||||
max_routes: int = 3,
|
||||
max_routes: int = MAX_ROUTES_PER_MODEL,
|
||||
) -> list[dict[str, Any]]:
|
||||
selected: list[dict[str, Any]] = []
|
||||
counts: Counter[str] = Counter()
|
||||
@@ -1453,7 +1489,7 @@ def run_submission(
|
||||
config_optimizer=config_optimizer,
|
||||
allow_dynamic_tasks=not bool(getattr(args, "task_types", None)),
|
||||
)
|
||||
stage_candidates = limit_routes_per_model(stage_candidates, max_routes=3)
|
||||
stage_candidates = limit_routes_per_model(stage_candidates, max_routes=MAX_ROUTES_PER_MODEL)
|
||||
remaining_candidate_capacity = max(0, candidate_goal - current_candidate_models)
|
||||
stage_candidates = limit_candidate_models(
|
||||
stage_candidates,
|
||||
|
||||
Reference in New Issue
Block a user