From 965783957e57d60d9928a25383e9dc440efb15e6 Mon Sep 17 00:00:00 2001 From: zhousha <736730048@qq.com> Date: Wed, 19 Aug 2026 13:59:54 +0800 Subject: [PATCH] add self-looping quota retry: retry quota-blocked models every 30min in-process, no redeploy needed --- main.py | 94 ++++++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 73 insertions(+), 21 deletions(-) diff --git a/main.py b/main.py index 9e438af..052fad7 100644 --- a/main.py +++ b/main.py @@ -6,7 +6,14 @@ xc_validation_strategy — 主入口 Kunlunxin_p-800 的 config_content 模板和模型列表仍保留在代码中，未列入本次 GPU_JOBS） （/adminApi/async/task/create-contest-task， Bearer Token 认证），之后保持 HTTP 服务存活。 -同时暴露 /health（K8s 探活）和 /status（运行状态）。 + +账号额度自动重试：如果某个模型提交时命中"当前等待中或运行中的异步模型验证 +任务数量已达上限"（账号额度已满），不算永久失败，会被留到下一轮；额度耗尽后 +本进程会原地等待 30 分钟，再自动重试所有因额度问题未提交成功的模型，如此循环， +直至全部提交成功或进程被平台关闭——不需要重新部署新策略，循环逻辑在本进程内完成。 +非额度原因的失败（如模型已在验证中等）不会重试。 + +同时暴露 /health（K8s 探活）和 /status（运行状态，含当前轮次/待重试数/下次重试时间）。 """ import json @@ -6086,13 +6093,16 @@ TOTAL_MODELS = sum(len(models) for _, models in GPU_JOBS) # ══════════════════════════════════════════════════════════ _state = { "strategy_id": STRATEGY_ID, - "phase": "starting", # starting | submitting | done | error + "phase": "starting", # starting | submitting | waiting_retry | done | error "total": TOTAL_MODELS, "submitted": 0, "failed": 0, "per_gpu": {gpu: 0 for gpu, _ in GPU_JOBS}, "started_at": None, "finished_at": None, + "round": 0, # 当前是第几轮提交 + "quota_blocked_remaining": 0, # 因额度上限暂未提交成功、等待下一轮重试的模型数 + "next_retry_at": None, # 下一轮重试的预计时间（额度耗尽等待期间） } _shutdown = threading.Event() @@ -6290,7 +6300,14 @@ ref_config: # ══════════════════════════════════════════════════════════ # 业务逻辑 # ══════════════════════════════════════════════════════════ -def _submit_task(token: str, gpu_type: str, model_id: str) -> Tuple[bool, str]: +# 账号"等待中/运行中"任务数已达上限时平台返回的业务错误信息（子串匹配）； +# 命中这个的模型不算永久失败，会在额度腾出空位后自动重试，不会被记作 failed +QUOTA_FULL_MSG = "当前等待中或运行中的异步模型验证任务数量已达上限" +# 额度耗尽后，隔多久自动重试一次剩余（因额度问题未提交成功）的模型 +RETRY_INTERVAL_SECONDS = 30 * 60 # 30 分钟 + + +def _submit_task(token: str, gpu_type: str, model_id: str) -> Tuple[bool, str, str]: headers = { "Content-Type": "application/json", "Authorization": f"Bearer {token}", @@ -6323,13 +6340,14 @@ def _submit_task(token: str, gpu_type: str, model_id: str) -> Tuple[bool, str]: if result.get("code") == 0: task_id = result.get("data", {}).get("id", "") print(f"[worker] OK {model_id} (GPU={gpu_type}) task_id={task_id}", flush=True) - return True, task_id + return True, task_id, "" else: - print(f"[worker] FAIL {model_id} (GPU={gpu_type}): {result.get('message')}", flush=True) - return False, "" + message = result.get("message") or "" + print(f"[worker] FAIL {model_id} (GPU={gpu_type}): {message}", flush=True) + return False, "", message except Exception as e: print(f"[worker] ERROR {model_id} (GPU={gpu_type}): {e}", flush=True) - return False, "" + return False, "", str(e) def _run_worker(): @@ -6340,35 +6358,69 @@ def _run_worker(): token = AUTH_TOKEN print("[worker] 使用预设 Token，跳过登录", flush=True) - for gpu_type, model_list in GPU_JOBS: - if _shutdown.is_set(): - break - print(f"\n{'='*60}\n🚀 开始处理 GPU={gpu_type}，共 {len(model_list)} 个模型\n{'='*60}", flush=True) + # 待提交队列：保持 GPU_JOBS 里原有的 (gpu_type, model_id) 顺序 + pending: List[Tuple[str, str]] = [ + (gpu_type, model_id) + for gpu_type, model_list in GPU_JOBS + for model_id in model_list + ] - for model_id in model_list: + round_num = 0 + while pending and not _shutdown.is_set(): + round_num += 1 + _state["round"] = round_num + _state["phase"] = "submitting" + _state["next_retry_at"] = None + print( + f"\n{'='*60}\n🚀 第 {round_num} 轮，待提交 {len(pending)} 个模型\n{'='*60}", + flush=True, + ) + + quota_blocked: List[Tuple[str, str]] = [] + for gpu_type, model_id in pending: if _shutdown.is_set(): break - ok, task_id = _submit_task(token, gpu_type, model_id) + ok, task_id, message = _submit_task(token, gpu_type, model_id) if ok: _state["submitted"] += 1 _state["per_gpu"][gpu_type] += 1 successful.append((task_id, gpu_type, model_id)) + elif QUOTA_FULL_MSG in message: + # 账号额度暂时满了，不算永久失败，留到下一轮重试 + quota_blocked.append((gpu_type, model_id)) else: + # 非额度原因失败（如重复提交等），不再重试 _state["failed"] += 1 - # 写入结果文件 - try: - with open("submitted_validation_tasks.txt", "w", encoding="utf-8") as f: - for tid, gpu, mid in successful: - f.write(f"{tid}\t{gpu}\t{mid}\n") - except Exception: - pass + pending = quota_blocked + _state["quota_blocked_remaining"] = len(pending) + + # 每轮结束都把已成功的结果落盘一次，避免中途重启丢失记录 + try: + with open("submitted_validation_tasks.txt", "w", encoding="utf-8") as f: + for tid, gpu, mid in successful: + f.write(f"{tid}\t{gpu}\t{mid}\n") + except Exception: + pass + + if pending and not _shutdown.is_set(): + next_retry = datetime.utcnow().timestamp() + RETRY_INTERVAL_SECONDS + _state["next_retry_at"] = datetime.utcfromtimestamp(next_retry).isoformat() + _state["phase"] = "waiting_retry" + print( + f"[worker] 第 {round_num} 轮结束：{len(pending)} 个模型因账号额度上限暂未提交，" + f"{RETRY_INTERVAL_SECONDS // 60} 分钟后自动重试（不部署新策略，本进程内循环）...", + flush=True, + ) + _shutdown.wait(RETRY_INTERVAL_SECONDS) _state["finished_at"] = datetime.utcnow().isoformat() _state["phase"] = "done" + _state["quota_blocked_remaining"] = len(pending) print( f"[worker] 完成 submitted={_state['submitted']} failed={_state['failed']} " - f"total={_state['total']} per_gpu={_state['per_gpu']}", + f"total={_state['total']} per_gpu={_state['per_gpu']} " + f"仍因额度未提交(如遇shutdown中断)={len(pending)}", flush=True, ) # 提交完成后继续保持进程存活，等待平台停止