fix: 算力节点任务/GPU 统计纳入评测与推理占用,补充测试依赖
- compute_nodes() 的 current_running_jobs 纳入运行中的评测任务(eval_tasks) 与已加载的推理模型(compare_tasks 持久化状态 + 内存标记兜底), 多节点时 GPU 被评测/推理占用不再显示 0/1 - gpus() 直接按 eval_tasks / compare_tasks 派生 GPU busy/reserved 状态, 修复 gpu_id=0 时 or -1 导致匹配失败;删除评测后 GPU 不再残留 busy - 评测不再复用 mark_inference_loaded 内存标记,GPU 占用由任务数据驱动 - requirements.txt 补充 pytest / ruff(此前仅声明在 pyproject dev 可选依赖) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -1537,8 +1537,8 @@ async def model_eval_start(payload: dict[str, Any] = Body(...)) -> dict[str, Any
|
||||
"compute_node_id": node["id"],
|
||||
"output_dir": output_dir,
|
||||
})
|
||||
if job.get("status") in {"queued", "running"}:
|
||||
store.mark_inference_loaded(node["id"])
|
||||
# 评测占用 GPU 由 eval_tasks 派生(gpus()/compute_nodes() 直接统计),
|
||||
# 不再复用 mark_inference_loaded 内存标记,避免删除评测后 GPU 状态残留 busy
|
||||
return ok({"task_id": task["id"], "status": "running", "job": job})
|
||||
except Exception as exc:
|
||||
store.update_eval_task(task["id"], {"status": "failed", "error": str(exc)})
|
||||
|
||||
@@ -2837,6 +2837,28 @@ class PlatformStore:
|
||||
"SELECT compute_node_id, COUNT(*) AS cnt FROM fine_tune_tasks WHERE status IN ('syncing','queued','running') GROUP BY compute_node_id"
|
||||
).fetchall()
|
||||
running_map = {r["compute_node_id"]: r["cnt"] for r in running}
|
||||
# 评测任务同样占用算力节点,纳入运行任务统计
|
||||
for row in conn.execute(
|
||||
"SELECT payload FROM eval_tasks WHERE status IN ('syncing','queued','running')"
|
||||
).fetchall():
|
||||
node_id = json_loads(row["payload"], {}).get("compute_node_id")
|
||||
if node_id:
|
||||
running_map[node_id] = running_map.get(node_id, 0) + 1
|
||||
# 推理模型占用算力节点同样计入:优先从 compare_tasks 持久化状态派生
|
||||
# (重启后仍准确),并用内存标记兜底(直接 preload 的模型无 compare 记录)
|
||||
inference_node_ids = set(self._inference_nodes)
|
||||
for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall():
|
||||
ls = json_loads(ctr["payload"], {}).get("load_status") or {}
|
||||
if isinstance(ls, str):
|
||||
try:
|
||||
ls = json.loads(ls)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
ls = {}
|
||||
for m in ls.get("loaded_models") or []:
|
||||
if m.get("status") in {"ready", "running"} and m.get("node_id"):
|
||||
inference_node_ids.add(m["node_id"])
|
||||
for nid in inference_node_ids:
|
||||
running_map[nid] = running_map.get(nid, 0) + 1
|
||||
rows = conn.execute("SELECT * FROM compute_nodes ORDER BY scheduler_weight DESC, code").fetchall()
|
||||
return [
|
||||
{
|
||||
@@ -3054,6 +3076,26 @@ class PlatformStore:
|
||||
"SELECT * FROM fine_tune_tasks WHERE status IN ('syncing','queued','running')"
|
||||
).fetchall()
|
||||
]
|
||||
# 评测任务同样占用节点 GPU
|
||||
eval_running = [
|
||||
json_loads(row["payload"], {})
|
||||
for row in conn.execute(
|
||||
"SELECT payload FROM eval_tasks WHERE status IN ('syncing','queued','running')"
|
||||
).fetchall()
|
||||
]
|
||||
# 推理模型占用的节点:优先从 compare_tasks 持久化状态派生(重启后仍准确),
|
||||
# 内存标记兜底(直接 preload 的模型无 compare 记录)
|
||||
inference_node_ids = set(self._inference_nodes)
|
||||
for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall():
|
||||
ls = json_loads(ctr["payload"], {}).get("load_status") or {}
|
||||
if isinstance(ls, str):
|
||||
try:
|
||||
ls = json.loads(ls)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
ls = {}
|
||||
for m in ls.get("loaded_models") or []:
|
||||
if m.get("status") in {"ready", "running"} and m.get("node_id"):
|
||||
inference_node_ids.add(m["node_id"])
|
||||
items = []
|
||||
for row in rows:
|
||||
task = next(
|
||||
@@ -3064,11 +3106,21 @@ class PlatformStore:
|
||||
),
|
||||
None,
|
||||
)
|
||||
busy = task is not None and task.get("status") == "running"
|
||||
reserved = task is not None and task.get("status") in {"syncing", "queued"}
|
||||
eval_task = next(
|
||||
(
|
||||
t
|
||||
for t in eval_running
|
||||
if t.get("compute_node_id") == row["node_id"]
|
||||
and row["gpu_index"] == (int(t["gpu_id"]) if t.get("gpu_id") is not None else -1)
|
||||
),
|
||||
None,
|
||||
)
|
||||
busy = (task is not None and task.get("status") == "running") or eval_task is not None
|
||||
reserved = (task is not None and task.get("status") in {"syncing", "queued"}) or (
|
||||
eval_task is not None and eval_task.get("status") in {"syncing", "queued"}
|
||||
)
|
||||
# Also mark GPU as busy if an inference model is loaded on this node
|
||||
inference_busy = self.is_inference_loaded(row["node_id"])
|
||||
if inference_busy and not busy:
|
||||
if row["node_id"] in inference_node_ids and not busy:
|
||||
busy = True
|
||||
reserved = False
|
||||
memory_used = round(row["memory_total_gb"] * (0.72 if busy else 0.18 if reserved else 0.04), 1)
|
||||
@@ -3100,6 +3152,16 @@ class PlatformStore:
|
||||
}
|
||||
]
|
||||
if task
|
||||
else [
|
||||
{
|
||||
"pid": int(eval_task.get("process_id") or 0),
|
||||
"name": "eval_runner",
|
||||
"memory_used_gb": memory_used,
|
||||
"task_name": eval_task.get("eval_task_name") or eval_task.get("name") or "评测任务",
|
||||
"user": "admin",
|
||||
}
|
||||
]
|
||||
if eval_task
|
||||
else [],
|
||||
}
|
||||
)
|
||||
|
||||
@@ -174,9 +174,7 @@ async def poll_compute_jobs_once() -> dict[str, Any]:
|
||||
except Exception:
|
||||
pass
|
||||
store.apply_eval_job_result(eval_task["id"], job, result_content)
|
||||
# If job completed, un-mark inference loaded
|
||||
if job.get("status") in {"completed", "failed", "stopped"}:
|
||||
store.mark_inference_unloaded(node["id"])
|
||||
# 评测 GPU 占用由 eval_tasks 状态派生,无需维护推理内存标记
|
||||
eval_synced += 1
|
||||
except Exception as exc: # noqa: BLE001
|
||||
failed.append({"eval_task_id": eval_task["id"], "error": str(exc)})
|
||||
|
||||
@@ -19,3 +19,7 @@ llama-index-core==0.14.23
|
||||
llama-index-embeddings-huggingface==0.6.1
|
||||
docling==2.115.0
|
||||
tiktoken>=0.7.0
|
||||
|
||||
# 测试与代码检查
|
||||
pytest>=8.2.0
|
||||
ruff>=0.5.0
|
||||
|
||||
Reference in New Issue
Block a user