diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py index 69a4f3a..dcbdac4 100644 --- a/backend/app/api/v1/endpoints/platform.py +++ b/backend/app/api/v1/endpoints/platform.py @@ -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)}) diff --git a/backend/app/db/platform_store.py b/backend/app/db/platform_store.py index 02835d4..c08b877 100644 --- a/backend/app/db/platform_store.py +++ b/backend/app/db/platform_store.py @@ -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 [], } ) diff --git a/backend/app/modules/compute_gateway/sync.py b/backend/app/modules/compute_gateway/sync.py index 4019ba2..1cac998 100644 --- a/backend/app/modules/compute_gateway/sync.py +++ b/backend/app/modules/compute_gateway/sync.py @@ -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)}) diff --git a/backend/requirements.txt b/backend/requirements.txt index 04e346a..2065541 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -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