From 2222e1545b889ec5cce413adc5052b234b5d6cf2 Mon Sep 17 00:00:00 2001 From: z3st Date: Fri, 24 Jul 2026 18:51:51 +0800 Subject: [PATCH] feat: cancel-all mode - cancel all waiting/running tasks on startup --- main.py | 77 +++++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 67 insertions(+), 10 deletions(-) diff --git a/main.py b/main.py index bee1e13..2dd944c 100644 --- a/main.py +++ b/main.py @@ -234,6 +234,69 @@ def check_queue_available() -> int: return -1 +def cancel_all_tasks(): + """取消所有 waiting 和 running 状态的任务""" + headers = { + 'Xc-Token': TARGET_TOKEN, + 'Accept': 'application/json', + 'Content-Type': 'application/json', + } + + # 1. 查询所有任务 + all_tasks = [] + page = 1 + while True: + try: + resp = requests.get( + f"{MODELHUB_API}/adapt/task/page", + headers={'Xc-Token': TARGET_TOKEN, 'Accept': 'application/json'}, + params={'current': page, 'pageSize': 100, 'onlyMine': 'true'}, + timeout=15 + ) + data = resp.json() + if data.get('code') != 0: + break + records = data['data'].get('records', []) + if not records: + break + all_tasks.extend(records) + log(f" 查询第{page}页: {len(records)} 条, 累计 {len(all_tasks)}/{data['data'].get('total')}") + if len(all_tasks) >= int(data['data'].get('total', 0)): + break + page += 1 + except Exception as e: + log(f" 查询失败: {e}") + return + + # 2. 筛选 waiting/running 状态 + cancellable = [t for t in all_tasks if t.get('status') in ('waiting', 'running')] + if not cancellable: + log(f" 没有可取消的任务 (总任务 {len(all_tasks)} 个)") + return + + task_ids = [int(t.get('taskId')) for t in cancellable if t.get('taskId')] + log(f" 总任务 {len(all_tasks)} 个, 可取消 {len(task_ids)} 个 (waiting/running)") + + # 3. 分批取消(每批最多50个) + url = f"{MODELHUB_API}/async/task/stop-create-contest-task" + cancelled = 0 + for i in range(0, len(task_ids), 50): + batch = task_ids[i:i+50] + try: + resp = requests.put(url, headers=headers, json={'taskIds': batch}, timeout=15) + data = resp.json() + if data.get('code') == 0: + cancelled += len(batch) + log(f" 批次 {i//50+1}: 取消 {len(batch)} 个 ✅") + else: + log(f" 批次 {i//50+1}: 失败 code={data.get('code')} msg={data.get('message','')[:80]}") + except Exception as e: + log(f" 批次 {i//50+1}: 异常 {e}") + time.sleep(1) + + log(f" 取消完成: {cancelled}/{len(task_ids)}") + + def build_config_params() -> str: """构建 YAML 配置 - 完全匹配平台自动生成的格式(只支持vllm)""" params = { @@ -682,18 +745,12 @@ def main(): log("连通性测试完成") log("=" * 50) - # 开始正式提交流程 - log("\n自动触发提交流程...") + # 取消所有 waiting/running 任务 + log("\n取消所有等待/运行中任务...") try: - state['running'] = True - state['last_run'] = datetime.now().isoformat() - count = run_pipeline(submit_limit=30) - state['last_result'] = {'submitted': count, 'success': True} + cancel_all_tasks() except Exception as e: - log(f"流程异常: {traceback.format_exc()}") - state['last_result'] = {'error': str(e), 'success': False} - finally: - state['running'] = False + log(f"取消任务异常: {e}") threading.Thread(target=_startup_test, daemon=True).start()