From 9379f93633ce6f642ac7c2f7cd15c5e066f9b50d Mon Sep 17 00:00:00 2001 From: wuyongtao Date: Wed, 19 Aug 2026 10:44:56 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A8=E7=90=86=E4=B8=8E=E8=AF=84?= =?UTF-8?q?=E6=B5=8B=E6=94=AF=E6=8C=81=E5=A4=9A=E5=8D=A1=20GPU=20=E9=80=89?= =?UTF-8?q?=E6=8B=A9=EF=BC=8C=E8=AE=AD=E7=BB=83=E4=BB=BB=E5=8A=A1=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=20GPU=20=E7=9B=91=E6=8E=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 GPU 选择归一化 helper，统一前端各形态的选择（gpu_indices/gpus/gpu_id） - 评测与推理支持同一节点内多卡选择，校验所选 GPU 空闲后再派发 - 新增 /fine-tune/{id}/gpu-status 接口，训练日志页展示实时 GPU 指标 - 算力节点推理加载支持 CUDA_VISIBLE_DEVICES 多卡可见，nvidia-smi 进程级监控 - GPU 占用跟踪细化为按卡记录，覆盖评测任务、推理模型与对比任务 Co-Authored-By: Claude --- backend/app/api/v1/endpoints/platform.py | 121 ++++++++++++++++-- backend/app/db/platform_store.py | 68 +++++++++- backend/app/modules/compute_gateway/client.py | 4 + backend/app/modules/compute_gateway/sync.py | 3 +- compute/api/main.py | 49 ++++++- compute/engines/llama_factory/inference.py | 25 ++++ frontend/src/api/modules/fineTune.ts | 11 ++ frontend/src/types/index.ts | 8 +- frontend/src/views/eval/EvalCreateView.vue | 18 ++- .../views/eval/create/EvalTaskSetupStep.vue | 20 ++- .../src/views/eval/create/StartEvalStep.vue | 2 +- .../views/inference/InferenceCreateView.vue | 57 ++++++--- frontend/src/views/system/TrainingLogView.vue | 22 ++-- 13 files changed, 360 insertions(+), 48 deletions(-) diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py index eb3ab1d..7b9ac29 100644 --- a/backend/app/api/v1/endpoints/platform.py +++ b/backend/app/api/v1/endpoints/platform.py @@ -57,6 +57,33 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None: return None +def _normalize_gpu_indices(payload: dict[str, Any], *, allow_primary: bool = True) -> list[int]: + """Normalize all frontend GPU selection shapes to sorted integer indexes.""" + raw = payload.get("gpu_indices") + if raw is None: + raw = payload.get("gpus") + if raw is None and allow_primary and payload.get("gpu_id") is not None: + raw = [payload.get("gpu_id")] + if raw is None or raw == "": + return [] + if isinstance(raw, str): + raw = [item.strip() for item in raw.split(",") if item.strip()] + if not isinstance(raw, (list, tuple, set)): + raw = [raw] + result: set[int] = set() + for item in raw: + if isinstance(item, str) and ":" in item: + item = item.rsplit(":", 1)[-1] + try: + index = int(item) + except (TypeError, ValueError) as exc: + raise ValueError(f"invalid GPU index: {item}") from exc + if index < 0: + raise ValueError("GPU index must be non-negative") + result.add(index) + return sorted(result) + + async def _wait_for_object_storage() -> None: """Wait for MinIO before starting a resource task.""" settings = get_settings() @@ -1527,10 +1554,10 @@ async def start_fine_tune( if node_id and gpu_indices: if not store.check_gpu_access(current_user["id"], node_id, gpu_indices): raise fail(403, "无权使用所选 GPU，请联系管理员分配") - # 记录创建者 if node_id and not gpu_indices: payload["allowed_gpu_indices"] = store.assigned_gpu_indexes(current_user["id"], node_id) - payload["strict_node_selection"] = bool(node_id) + # 页面明确选择节点时，调度器必须保持节点约束；否则可能落到其它节点。 + payload["strict_node_selection"] = bool(payload.get("compute_node_id") or payload.get("node_id")) payload.setdefault("created_by", current_user.get("id")) try: return ok(await _submit_fine_tune_task(store, payload)) @@ -1661,6 +1688,42 @@ async def fine_tune_diagnostics(task_id: str) -> dict[str, Any]: ) +@router.get("/fine-tune/{task_id}/gpu-status") +async def fine_tune_gpu_status(task_id: str, current_user: dict[str, Any] = Depends(get_current_user)) -> dict[str, Any]: + """Return live GPU metrics for the task's selected node and cards.""" + store = get_platform_store() + try: + task = store.task(task_id) + except KeyError: + raise fail(404, "fine tune task not found") + if not has_resource_access("fine-tune", task_id, current_user, "read"): + raise fail(403, "no permission to access this task") + node = _node_for_task(task) + selected = set(_normalize_gpu_indices({"gpus": task.get("gpus") or []}, allow_primary=False)) + if not node: + return ok({"source": "unavailable", "items": [], "selected_gpus": sorted(selected)}) + try: + if get_settings().compute_mode == "simulator": + live_items = store.gpus() + else: + live_items = await ComputeNodeClient(node["api_base_url"]).gpu_resources() + items = [] + for item in live_items: + index = int(item.get("gpu_index", item.get("id", -1))) + if selected and index not in selected: + continue + items.append({ + **item, + "id": index, + "node_id": node["id"], + "node_code": node.get("code"), + "node_name": node.get("name"), + }) + return ok({"source": "compute", "items": items, "selected_gpus": sorted(selected)}) + except Exception as exc: # noqa: BLE001 - let UI retain last good snapshot + return ok({"source": "unavailable", "items": [], "selected_gpus": sorted(selected), "error": str(exc)}) + + @router.put("/fine-tune/{task_id}") async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]: if not has_resource_access("fine-tune", task_id, current_user, "write"): @@ -1808,6 +1871,12 @@ async def model_eval_detail(task_id: str, current_user: dict = Depends(get_curre async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]: """Start an evaluation task: submit eval job to compute node.""" store = get_platform_store() + try: + gpu_indices = _normalize_gpu_indices(payload) + except ValueError as exc: + raise fail(400, str(exc)) + if not gpu_indices: + raise fail(400, "请选择至少一张 GPU") # 1. Create eval task record payload.setdefault("created_by", current_user.get("id")) task = store.create_eval_task({**payload, "status": "pending"}) @@ -1910,6 +1979,17 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di store.update_eval_task(task["id"], {"status": "failed", "error": message}) return ok({"task_id": task["id"], "status": "failed", "error": message}) + node_gpus = { + int(item.get("id", item.get("gpu_index", -1))): item + for item in store.gpus() + if item.get("node_id") == node["id"] + } + unavailable = [index for index in gpu_indices if node_gpus.get(index, {}).get("status") != "idle"] + if unavailable: + message = f"selected GPU is not idle on compute node {node.get('code')}: {unavailable}" + store.update_eval_task(task["id"], {"status": "failed", "error": message}) + return ok({"task_id": task["id"], "status": "failed", "error": message}) + # 6. Build eval job payload output_dir = f"/data/yg-ft/outputs/{task['id']}" job_payload = { @@ -1923,7 +2003,9 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di "output_dir": output_dir, "basic_metrics": payload.get("basic_metrics", {}), "dimension": dimension_cfg, - "gpus": [int(payload.get("gpu_id", 0))], + "gpu_id": gpu_indices[0], + "gpu_indices": gpu_indices, + "gpus": gpu_indices, "temperature": payload.get("temperature", 0.1), "max_new_tokens": payload.get("max_new_tokens", 512), "compute_node_id": node["id"], @@ -2222,6 +2304,15 @@ async def model_compare_load(task_id: str, current_user: dict = Depends(get_curr "model_name_or_path": model_path, "template": item.get("template", "qwen"), } + try: + item_gpu_indices = _normalize_gpu_indices(item) + except ValueError as exc: + loaded_models.append({**item, "status": "error", "error": str(exc)}) + continue + if not item_gpu_indices: + loaded_models.append({**item, "status": "error", "error": "no GPU selected"}) + continue + load_payload["gpu_indices"] = item_gpu_indices if item.get("adapter_path"): load_payload["adapter_name_or_path"] = item["adapter_path"] if get_settings().compute_mode == "simulator": @@ -2230,14 +2321,28 @@ async def model_compare_load(task_id: str, current_user: dict = Depends(get_curr # 只派发：HTTP 响应成功即视为已接受（节点会异步加载），loaded 字段忽略 item_dispatched = False errors = [] - for node in _candidate_online_nodes(store, preferred_node_id): + candidate_nodes = _candidate_online_nodes(store, preferred_node_id) + if preferred_node_id: + candidate_nodes = candidate_nodes[:1] + for node in candidate_nodes: try: + node_gpu_map = { + int(gpu.get("id", gpu.get("gpu_index", -1))): gpu + for gpu in store.gpus() + if gpu.get("node_id") == node["id"] + } + unavailable = [ + index for index in item_gpu_indices + if node_gpu_map.get(index, {}).get("status") != "idle" + ] + if unavailable: + raise RuntimeError(f"selected GPU is not idle on compute node {node.get('code')}: {unavailable}") if get_settings().minio_enabled: await _wait_for_object_storage() client = ComputeNodeClient(node["api_base_url"]) await client.inference_load(load_payload) - store.mark_inference_loaded(node["id"]) - loaded_models.append({**item, "status": "starting", "node_id": node["id"], "node_name": node.get("name")}) + store.mark_inference_loaded(node["id"], item_gpu_indices) + loaded_models.append({**item, "gpu_indices": item_gpu_indices, "gpus": item_gpu_indices, "status": "starting", "node_id": node["id"], "node_name": node.get("name")}) item_dispatched = True break except Exception as exc: # noqa: BLE001 - try next candidate node @@ -2340,7 +2445,7 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[ # 计算节点现在异步加载：HTTP 接受（loading/ready）即视为派发成功 result = await client.inference_load(payload) if result.get("loaded") or result.get("status") in {"loading", "ready"}: - store.mark_inference_loaded(node["id"]) + store.mark_inference_loaded(node["id"], _normalize_gpu_indices(payload)) return ok(result) except Exception as exc: return ok({"loaded": False, "error": str(exc)}) @@ -2420,7 +2525,7 @@ async def model_chat_trained_preload(payload: dict[str, Any] = Body(...), curren # 计算节点现在异步加载：HTTP 接受（loading/ready）即视为派发成功 result = await client.inference_load({**payload, "compute_node_id": node["id"]}) if result.get("loaded") or result.get("status") in {"loading", "ready"}: - store.mark_inference_loaded(node["id"]) + store.mark_inference_loaded(node["id"], _normalize_gpu_indices(payload)) return ok(result) except Exception as exc: return ok({"loaded": False, "error": str(exc)}) diff --git a/backend/app/db/platform_store.py b/backend/app/db/platform_store.py index 2a2f5dd..cea1469 100644 --- a/backend/app/db/platform_store.py +++ b/backend/app/db/platform_store.py @@ -454,15 +454,21 @@ class PlatformStore: self.ensure_seed_data() # Track which compute nodes have an active inference model loaded self._inference_nodes: set[str] = set() + self._inference_gpu_indexes: dict[str, set[int]] = {} self._last_runtime_refresh = 0.0 # ── inference node tracking ──────────────────────────────────── - def mark_inference_loaded(self, node_id: str) -> None: + def mark_inference_loaded(self, node_id: str, gpu_indexes: list[int] | None = None) -> None: self._inference_nodes.add(node_id) + if gpu_indexes is not None: + self._inference_gpu_indexes[node_id] = {int(item) for item in gpu_indexes} + else: + self._inference_gpu_indexes.pop(node_id, None) def mark_inference_unloaded(self, node_id: str) -> None: self._inference_nodes.discard(node_id) + self._inference_gpu_indexes.pop(node_id, None) def is_inference_loaded(self, node_id: str) -> bool: return node_id in self._inference_nodes @@ -2766,7 +2772,35 @@ class PlatformStore: "SELECT gpu_index FROM gpu_allocations WHERE node_id=? AND status IN ('allocated','running')", (node_id,), ).fetchall() - return {int(row["gpu_index"]) for row in rows} + active = {int(row["gpu_index"]) for row in rows} + # Evaluation jobs use the same Compute ProcessManager GPU lock but do + # not have fine-tune allocation rows; derive their selected cards here + # so a training task cannot race onto an evaluation GPU. + for row in conn.execute( + "SELECT payload FROM eval_tasks WHERE status IN ('syncing','queued','running')" + ).fetchall(): + payload = json_loads(row["payload"], {}) + if payload.get("compute_node_id") != node_id: + continue + selected = payload.get("gpu_indices") or payload.get("gpus") + if selected is None and payload.get("gpu_id") is not None: + selected = [payload.get("gpu_id")] + active.update(int(item) for item in selected or []) + # Loaded inference models also reserve only their selected cards. + active.update(self._inference_gpu_indexes.get(node_id, set())) + for row in conn.execute("SELECT payload FROM compare_tasks").fetchall(): + payload = json_loads(row["payload"], {}) + load_status = payload.get("load_status") or {} + if isinstance(load_status, str): + load_status = json_loads(load_status, {}) + for item in load_status.get("loaded_models") or []: + if item.get("node_id") != node_id or item.get("status") not in {"starting", "ready", "running"}: + continue + selected = item.get("gpu_indices") or item.get("gpus") + if selected is None and item.get("gpu_id") is not None: + selected = [item.get("gpu_id")] + active.update(int(gpu) for gpu in selected or []) + return active def _node_gpu_indexes(self, conn: PgConnection, node: dict[str, Any]) -> set[int]: rows = conn.execute("SELECT gpu_index FROM gpus WHERE node_id=?", (node["id"],)).fetchall() @@ -3195,6 +3229,9 @@ class PlatformStore: # 推理模型占用算力节点同样计入：优先从 compare_tasks 持久化状态派生 # （重启后仍准确），并用内存标记兜底（直接 preload 的模型无 compare 记录） inference_node_ids = set(self._inference_nodes) + inference_gpu_indexes: dict[str, set[int]] = { + node_id: set(indexes) for node_id, indexes in self._inference_gpu_indexes.items() + } for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall(): ls = json_loads(ctr["payload"], {}).get("load_status") or {} if isinstance(ls, str): @@ -3204,8 +3241,13 @@ class PlatformStore: 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: + node_id = m["node_id"] + selected = m.get("gpu_indices") or m.get("gpus") + if selected: + inference_gpu_indexes.setdefault(node_id, set()).update(int(item) for item in selected) + else: + inference_node_ids.add(node_id) + for nid in set(inference_node_ids) | set(inference_gpu_indexes): running_map[nid] = running_map.get(nid, 0) + 1 rows = conn.execute("SELECT * FROM compute_nodes ORDER BY scheduler_weight DESC, code").fetchall() return [ @@ -3434,6 +3476,9 @@ class PlatformStore: # 推理模型占用的节点：优先从 compare_tasks 持久化状态派生（重启后仍准确）， # 内存标记兜底（直接 preload 的模型无 compare 记录） inference_node_ids = set(self._inference_nodes) + inference_gpu_indexes: dict[str, set[int]] = { + node_id: set(indexes) for node_id, indexes in self._inference_gpu_indexes.items() + } for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall(): ls = json_loads(ctr["payload"], {}).get("load_status") or {} if isinstance(ls, str): @@ -3443,7 +3488,12 @@ class PlatformStore: 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"]) + node_id = m["node_id"] + selected = m.get("gpu_indices") or m.get("gpus") + if selected: + inference_gpu_indexes.setdefault(node_id, set()).update(int(item) for item in selected) + else: + inference_node_ids.add(node_id) items = [] for row in rows: task = next( @@ -3459,7 +3509,10 @@ class PlatformStore: 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) + and row["gpu_index"] in { + int(item) + for item in (t.get("gpu_indices") or t.get("gpus") or ([t["gpu_id"]] if t.get("gpu_id") is not None else [])) + } ), None, ) @@ -3468,7 +3521,8 @@ class PlatformStore: 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 - if row["node_id"] in inference_node_ids and not busy: + inference_on_gpu = row["node_id"] in inference_node_ids or row["gpu_index"] in inference_gpu_indexes.get(row["node_id"], set()) + if inference_on_gpu 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) diff --git a/backend/app/modules/compute_gateway/client.py b/backend/app/modules/compute_gateway/client.py index 280d617..4287d5a 100644 --- a/backend/app/modules/compute_gateway/client.py +++ b/backend/app/modules/compute_gateway/client.py @@ -240,6 +240,10 @@ class ComputeNodeClient: async def inference_status(self) -> dict[str, Any]: return await self._request("GET", "/inference/status", timeout=INFERENCE_STATUS_TIMEOUT) + async def gpu_resources(self) -> list[dict[str, Any]]: + """Read live per-GPU metrics from this compute node.""" + return await self.gpus() + async def inference_unload(self) -> dict[str, Any]: return await self._request("POST", "/inference/unload", json_data={}, timeout=INFERENCE_UNLOAD_TIMEOUT) diff --git a/backend/app/modules/compute_gateway/sync.py b/backend/app/modules/compute_gateway/sync.py index 1cac998..e57b775 100644 --- a/backend/app/modules/compute_gateway/sync.py +++ b/backend/app/modules/compute_gateway/sync.py @@ -73,7 +73,8 @@ async def reconcile_inference_loads(store: Any) -> list[dict[str, Any]]: if node_status == "ready": item["status"] = "ready" item.pop("error", None) - store.mark_inference_loaded(node["id"]) + selected_gpus = item.get("gpu_indices") or item.get("gpus") + store.mark_inference_loaded(node["id"], selected_gpus) elif node_status == "error": item["status"] = "error" item["error"] = status.get("error") or "model load failed on compute node" diff --git a/compute/api/main.py b/compute/api/main.py index 6d47c60..2f71703 100644 --- a/compute/api/main.py +++ b/compute/api/main.py @@ -221,6 +221,35 @@ def create_app() -> FastAPI: return fallback_gpu_resources() items: list[dict[str, Any]] = [] + processes_by_uuid: dict[str, list[dict[str, Any]]] = {} + try: + process_result = subprocess.run( + [ + "nvidia-smi", + "--query-compute-apps=gpu_uuid,pid,process_name,used_memory", + "--format=csv,noheader,nounits", + ], + check=True, + capture_output=True, + text=True, + timeout=5, + ) + for process_line in process_result.stdout.splitlines(): + process_parts = [part.strip() for part in process_line.split(",")] + if len(process_parts) < 4: + continue + process_uuid, pid, process_name, used_memory = process_parts[:4] + processes_by_uuid.setdefault(process_uuid, []).append( + { + "pid": int(_safe_float(pid)), + "name": process_name, + "memory_used_gb": round(_safe_float(used_memory) / 1024, 2), + } + ) + except Exception: + # Some driver/runtime combinations do not expose compute-apps; + # utilization and memory metrics remain useful without processes. + pass for line in result.stdout.splitlines(): parts = [part.strip() for part in line.split(",")] if len(parts) < 9: @@ -244,7 +273,7 @@ def create_app() -> FastAPI: "temperature": int(_safe_float(temp)), "power_w": round(_safe_float(power), 1), "power_limit_w": round(_safe_float(power_limit), 1), - "processes": [], + "processes": processes_by_uuid.get(uuid, []), } ) return items @@ -747,6 +776,23 @@ def create_app() -> FastAPI: infer_backend: str (default: "huggingface") infer_dtype: str (default: "auto") """ + requested_gpus = payload.get("gpu_indices") + if requested_gpus is None: + requested_gpus = payload.get("gpus") or [] + try: + requested_gpus = sorted({int(item) for item in requested_gpus}) + except (TypeError, ValueError) as exc: + raise HTTPException(status_code=400, detail=f"invalid GPU selection: {exc}") from exc + if any(item < 0 for item in requested_gpus): + raise HTTPException(status_code=400, detail="GPU index must be non-negative") + if requested_gpus: + known_gpus = {int(item.get("gpu_index", item.get("id", -1))) for item in gpu_resources()} + missing = sorted(set(requested_gpus) - known_gpus) + if missing: + raise HTTPException(status_code=409, detail=f"requested GPU not found: {missing}") + conflict = sorted(set(requested_gpus).intersection(process_manager.locked_gpus())) + if conflict: + raise HTTPException(status_code=409, detail=f"GPU already used by another compute job: {conflict}") session = get_inference_session() result = session.load( model_name_or_path=payload.get("model_name_or_path", ""), @@ -754,6 +800,7 @@ def create_app() -> FastAPI: template=payload.get("template", "qwen"), infer_backend=payload.get("infer_backend", "huggingface"), infer_dtype=payload.get("infer_dtype", "auto"), + gpu_indices=requested_gpus, ) return result diff --git a/compute/engines/llama_factory/inference.py b/compute/engines/llama_factory/inference.py index 72cad82..018ec0b 100644 --- a/compute/engines/llama_factory/inference.py +++ b/compute/engines/llama_factory/inference.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import threading import time import uuid @@ -36,6 +37,7 @@ class InferenceSession: self._generating_args: dict[str, Any] = {} self._model_name: str = "" self._adapter_path: str = "" + self._gpu_indices: list[int] = [] self._loaded_at: float = 0.0 @property @@ -53,6 +55,7 @@ class InferenceSession: "loaded_at": self._loaded_at, "request_id": self._request_id, "error": self._error, + "gpu_indices": list(self._gpu_indices), } def wait_until_loaded(self, timeout: float | None = None) -> dict[str, Any]: @@ -87,8 +90,12 @@ class InferenceSession: template="qwen", infer_backend="huggingface", infer_dtype="auto", + gpu_indices=None, **kwargs, ) -> dict[str, Any]: + requested_gpus = sorted({int(item) for item in (gpu_indices or [])}) + if any(item < 0 for item in requested_gpus): + return {"loaded": False, "status": "error", "error": "GPU index must be non-negative"} with self._state_lock: if self._status == "loading": # A model is already loading — dedupe, reuse the same request id. @@ -97,6 +104,7 @@ class InferenceSession: self._status = "loading" self._error = "" self._request_id = uuid.uuid4().hex[:12] + self._gpu_indices = requested_gpus self._cancel_requested = False self._load_args = { "model_name_or_path": model_name_or_path, @@ -115,13 +123,21 @@ class InferenceSession: def _load_worker(self) -> None: """Build the ChatModel off the state lock so info() never blocks.""" + with self._state_lock: + requested_gpus = list(self._gpu_indices) model = None tokenizer = None generating_args: dict[str, Any] = {} error = "" + previous_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES") try: + # Set visibility before LLaMA-Factory/PyTorch initializes CUDA. + if requested_gpus: + os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(str(item) for item in requested_gpus) if self._teardown_old: self._release_model() + with self._state_lock: + self._gpu_indices = requested_gpus from llamafactory.chat import ChatModel from llamafactory.hparams import get_infer_args @@ -138,6 +154,12 @@ class InferenceSession: generating_args = dict(generating_args) except Exception as exc: # noqa: BLE001 - surface load failure via status error = str(exc) + finally: + if requested_gpus: + if previous_visible_devices is None: + os.environ.pop("CUDA_VISIBLE_DEVICES", None) + else: + os.environ["CUDA_VISIBLE_DEVICES"] = previous_visible_devices with self._state_lock: if error: self._model = None @@ -152,6 +174,7 @@ class InferenceSession: self._model = None self._tokenizer = None self._status = "idle" + self._gpu_indices = [] return self._model = model self._tokenizer = tokenizer @@ -189,6 +212,7 @@ class InferenceSession: self._adapter_path = "" self._loaded_at = 0.0 self._error = "" + self._gpu_indices = [] def unload(self) -> dict[str, Any]: with self._state_lock: @@ -208,6 +232,7 @@ class InferenceSession: self._adapter_path = "" self._loaded_at = 0.0 self._error = "" + self._gpu_indices = [] return {"unloaded": True, "status": "idle"} def chat(self, messages, temperature=0.95, top_p=0.7, max_new_tokens=1024, do_sample=True, **kwargs) -> dict[str, Any]: diff --git a/frontend/src/api/modules/fineTune.ts b/frontend/src/api/modules/fineTune.ts index 374ed49..aefc53e 100644 --- a/frontend/src/api/modules/fineTune.ts +++ b/frontend/src/api/modules/fineTune.ts @@ -42,12 +42,23 @@ export interface FineTunePreflightResult { sync_results?: Array> } +export interface FineTuneGpuStatus { + source: string + items: Array> + selected_gpus: number[] + error?: string +} + /** 训练任务列表 */ export const getFineTuneList = () => get('/fine-tune') /** 训练任务详情 */ export const getFineTune = (id: string | number) => get(`/fine-tune/${id}`) +/** 获取任务所在 Compute 节点的实时 GPU 指标 */ +export const getFineTuneGpuStatus = (id: string | number) => + get(`/fine-tune/${id}/gpu-status`) + /** 任务名查重 */ export const checkFineTuneName = (name: string) => get<{ exists: boolean }>('/fine-tune/check-name', { name }) diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 4c93efa..722b186 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -218,6 +218,8 @@ export interface LoadedModel { port?: number node_id?: string node_name?: string + gpu_indices?: number[] + gpus?: number[] error?: string } @@ -239,6 +241,8 @@ export interface CompareModelRef { gpu_id: number node_id?: string node_name?: string + gpu_indices?: number[] + gpus?: number[] source?: string port?: number } @@ -282,7 +286,9 @@ export interface StartEvalPayload { eval_task_name: string eval_type: EvalType model_id: string | number - gpu_id: string | number + gpu_id: string | number | string[] + gpu_indices?: number[] + gpus?: number[] compute_node_id?: string dataset_id: string | number dimension_id: string | number diff --git a/frontend/src/views/eval/EvalCreateView.vue b/frontend/src/views/eval/EvalCreateView.vue index 1e82048..4973197 100644 --- a/frontend/src/views/eval/EvalCreateView.vue +++ b/frontend/src/views/eval/EvalCreateView.vue @@ -39,7 +39,7 @@ const createdDimensionId = ref('') const taskForm = ref({ eval_task_name: '', model_id: '', - gpu_id: '', + gpu_id: [], data_source: 'dataset', dataset_id: '', leaderboard: false, @@ -145,12 +145,24 @@ async function handleSubmit() { const dimensionId = await resolveDimensionId() // GPU 选择为「节点:GPU序号」复合值，解析出节点与 GPU 序号， // 多算力节点时必须把节点信息传给后端，否则会派发到错误的算力节点 - const [gpuNodeId, gpuIndex] = String(taskForm.value.gpu_id).split(':') + const selectedGpuKeys = Array.isArray(taskForm.value.gpu_id) + ? taskForm.value.gpu_id + : [String(taskForm.value.gpu_id)] + const gpuSelections = selectedGpuKeys + .map((key) => { + const [nodeId, gpuIndex] = String(key).split(':') + return { nodeId, gpuIndex: Number(gpuIndex) } + }) + .filter((item) => item.nodeId && Number.isInteger(item.gpuIndex) && item.gpuIndex >= 0) + const gpuNodeId = gpuSelections[0]?.nodeId || '' + const gpuIndices = gpuSelections.map((item) => item.gpuIndex) const evalResult: any = await startEval({ eval_task_name: taskForm.value.eval_task_name, eval_type: 'custom', model_id: taskForm.value.model_id, - gpu_id: Number(gpuIndex) || 0, + gpu_id: gpuIndices[0] ?? 0, + gpu_indices: gpuIndices, + gpus: gpuIndices, compute_node_id: gpuNodeId || '', dataset_id: taskForm.value.data_source === 'dataset' ? taskForm.value.dataset_id : '', dimension_id: dimensionId, diff --git a/frontend/src/views/eval/create/EvalTaskSetupStep.vue b/frontend/src/views/eval/create/EvalTaskSetupStep.vue index 8b595c5..cf49fb7 100644 --- a/frontend/src/views/eval/create/EvalTaskSetupStep.vue +++ b/frontend/src/views/eval/create/EvalTaskSetupStep.vue @@ -6,7 +6,7 @@ import type { DatasetItem, GpuInfo, TrainedModel } from '@/types' export interface EvalTaskSetupDraft { eval_task_name: string model_id: string | number - gpu_id: string | number + gpu_id: string | number | string[] data_source: 'dataset' | 'inference' dataset_id: string | number leaderboard: boolean @@ -23,6 +23,13 @@ defineProps<{ const form = defineModel({ required: true }) const formRef = ref() +function handleGpuChange(value: string | number | string[]) { + const keys = Array.isArray(value) ? value.map(String) : [String(value || '')] + const nodeId = keys[0]?.split(':', 1)[0] + if (!nodeId || !Array.isArray(form.value.gpu_id)) return + form.value.gpu_id = keys.filter((key) => key.split(':', 1)[0] === nodeId) +} + const rules: FormRules = { eval_task_name: [ { required: true, message: '请输入任务名称', trigger: 'blur' }, @@ -101,7 +108,16 @@ defineExpose({ validate }) - + (items: T[], id /** GPU 选择为「节点:GPU序号」复合值，解析并展示为可读标签 */ const gpuLabel = computed(() => { - const key = String(props.task.gpu_id || '') + const key = Array.isArray(props.task.gpu_id) ? String(props.task.gpu_id[0] || '') : String(props.task.gpu_id || '') const gpu = props.gpus.find((g) => `${g.node_id || ''}:${g.id ?? 0}` === key) if (gpu) return `${gpu.node_name || gpu.node_code || '算力节点'} / GPU ${gpu.id ?? 0}` const [nodeId, idx] = key.split(':') diff --git a/frontend/src/views/inference/InferenceCreateView.vue b/frontend/src/views/inference/InferenceCreateView.vue index 4224f17..c1b174e 100644 --- a/frontend/src/views/inference/InferenceCreateView.vue +++ b/frontend/src/views/inference/InferenceCreateView.vue @@ -4,8 +4,7 @@ import { useRouter } from 'vue-router' import { ElMessage, type FormInstance, type FormRules } from 'element-plus' import PageCard from '@/components/PageCard.vue' import { getModelList, getTrainedModels } from '@/api/modules/model' -import { getSystemInfo } from '@/api/modules/system' -import { getComputeNodes, type ComputeNode } from '@/api/modules/compute' +import { getComputeGpus, getComputeNodes, type ComputeNode } from '@/api/modules/compute' import { createCompare, loadCompare } from '@/api/modules/compare' import type { ModelItem, TrainedModel, GpuInfo } from '@/types' @@ -83,8 +82,8 @@ const form = reactive({ description: '', /** 选中的模型 key（单选） */ model_key: '', - /** 使用的 GPU */ - gpu_key: '', + /** 使用的 GPU（同一节点内可多选） */ + gpu_keys: [] as string[], }) const rules: FormRules = { @@ -94,12 +93,26 @@ const rules: FormRules = { /** 当前选中的模型对象 */ const selectedModel = computed(() => modelMap.value[form.model_key]) -const selectedGpu = computed(() => idleGpus.value.find((g) => `${g.node_id || ''}:${g.id ?? 0}` === form.gpu_key)) +const selectedGpus = computed(() => idleGpus.value.filter((gpu) => form.gpu_keys.includes(gpuKey(gpu)))) + +function gpuKey(gpu: GpuInfo) { + return `${gpu.node_id || ''}:${gpu.id ?? 0}` +} + +function handleGpuChange(keys: string[]) { + const nodeId = keys[0]?.split(':', 1)[0] + if (!nodeId) return + const filtered = keys.filter((key) => key.split(':', 1)[0] === nodeId) + if (filtered.length !== keys.length) { + ElMessage.info('一次推理只能使用同一算力节点内的 GPU，已忽略其它节点的选择') + } + form.gpu_keys = filtered +} watch(selectedModel, (model) => { if (!model?.compute_node_id) return const gpu = idleGpus.value.find((item) => item.node_id === model.compute_node_id) - if (gpu) form.gpu_key = `${gpu.node_id || ''}:${gpu.id ?? 0}` + if (gpu) form.gpu_keys = [gpuKey(gpu)] }) async function handleSubmit() { @@ -111,6 +124,10 @@ async function handleSubmit() { ElMessage.warning('请选择模型') return } + if (!selectedGpus.value.length) { + ElMessage.warning('请至少选择一张空闲 GPU') + return + } submitting.value = true startupStatus.value = '正在创建推理任务...' try { @@ -130,9 +147,11 @@ async function handleSubmit() { model_name: m.name, model_path: m.model_path, source: m.source, - gpu_id: selectedGpu.value?.id ?? 0, - node_id: selectedGpu.value?.node_id || m.compute_node_id, - node_name: selectedGpu.value?.node_name || m.compute_node_name, + gpu_id: selectedGpus.value[0]?.id ?? 0, + gpu_indices: selectedGpus.value.map((gpu) => Number(gpu.id ?? 0)), + gpus: selectedGpus.value.map((gpu) => Number(gpu.id ?? 0)), + node_id: selectedGpus.value[0]?.node_id || m.compute_node_id, + node_name: selectedGpus.value[0]?.node_name || m.compute_node_name, }, ], }) @@ -169,17 +188,17 @@ async function loadData() { const [db, trained, sys, nodes] = await Promise.all([ getModelList(), getTrainedModels(), - getSystemInfo(), + getComputeGpus(), getComputeNodes(), ]) dbModels.value = db || [] trainedModels.value = trained?.models || [] - gpus.value = sys?.gpu || [] + gpus.value = (sys || []) as unknown as GpuInfo[] computeNodes.value = nodes || [] // 默认选中第一个空闲 GPU if (idleGpus.value.length > 0) { const firstGpu = idleGpus.value[0] - form.gpu_key = `${firstGpu.node_id || ''}:${firstGpu.id ?? 0}` + form.gpu_keys = [gpuKey(firstGpu)] } } catch { // ignore @@ -228,12 +247,20 @@ onMounted(loadData) - + diff --git a/frontend/src/views/system/TrainingLogView.vue b/frontend/src/views/system/TrainingLogView.vue index 80c946e..35e98f2 100644 --- a/frontend/src/views/system/TrainingLogView.vue +++ b/frontend/src/views/system/TrainingLogView.vue @@ -8,10 +8,9 @@ import TrainingTaskOverview from './training-log/TrainingTaskOverview.vue' import { usePolling } from '@/composables/usePolling' import '@/plugins/echarts-training-log' import { useModelsStore } from '@/stores/models' -import { getFineTune, getFineTuneDiagnostics, getFineTuneLogs, getFineTuneMetrics, type TrainingDiagnostic } from '@/api/modules/fineTune' +import { getFineTune, getFineTuneDiagnostics, getFineTuneGpuStatus, getFineTuneLogs, getFineTuneMetrics, type TrainingDiagnostic } from '@/api/modules/fineTune' import { getTrainingLogFiles, getTrainingLogContent } from '@/api/modules/log' import { getDataset } from '@/api/modules/dataset' -import { getSystemInfo } from '@/api/modules/system' import { TRAIN_TYPE_MAP, TRAIN_METHOD_MAP } from '@/constants' import { buildMetricChartOption, @@ -205,15 +204,20 @@ async function loadDataset(datasetId: string | number) { } } -async function loadGpuStatus() { +async function loadGpuStatus(currentTask: FineTuneTask) { try { - const systemInfo = await getSystemInfo() - gpuPool.value = systemInfo.gpu ?? [] - gpuUpdatedAt.value = new Date() - gpuLoadError.value = '' + const live = await getFineTuneGpuStatus(currentTask.id) + if (live.source === 'compute' && live.items.length) { + gpuPool.value = live.items as unknown as GpuInfo[] + gpuUpdatedAt.value = new Date() + gpuLoadError.value = '' + return + } + gpuPool.value = [] + gpuLoadError.value = live.error || 'Compute 节点暂未返回实时 GPU 指标' } catch { gpuLoadError.value = 'GPU 监控数据暂时不可用' - if (!gpuUpdatedAt.value) gpuPool.value = [] + gpuPool.value = [] } } @@ -306,7 +310,7 @@ async function refreshAll() { const datasetPromise = currentTask.train_dataset_id ? loadDataset(currentTask.train_dataset_id) : Promise.resolve() - await Promise.all([datasetPromise, loadLog(currentTask), loadGpuStatus(), loadDiagnostics(currentTask)]) + await Promise.all([datasetPromise, loadLog(currentTask), loadGpuStatus(currentTask), loadDiagnostics(currentTask)]) await loadMetrics(currentTask) } finally { loading.value = false