feat: cancel-all mode - cancel all waiting/running tasks on startup
This commit is contained in:
77
main.py
77
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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user