From 5b0924f23998eb163b839665f1bfdfb30520a780 Mon Sep 17 00:00:00 2001 From: paperTII <2293564561@qq.com> Date: Mon, 8 Jun 2026 20:55:10 +0800 Subject: [PATCH] Environment pre-check code adaptation --- test/common/envPreCheck/EnvPreCheck.md | 151 +++---- test/common/envPreCheck/EnvPreCheck_ch.md | 68 ++- test/common/envPreCheck/run_env_preCheck.py | 433 +++++++------------- test/common/llmperf/LLMPerf.md | 10 +- test/common/llmperf/LLMPerf_zh.md | 10 +- test/config.yaml | 6 +- 6 files changed, 261 insertions(+), 417 deletions(-) diff --git a/test/common/envPreCheck/EnvPreCheck.md b/test/common/envPreCheck/EnvPreCheck.md index ab0faa172..f2f56a067 100644 --- a/test/common/envPreCheck/EnvPreCheck.md +++ b/test/common/envPreCheck/EnvPreCheck.md @@ -1,52 +1,70 @@ - # 🔍 Environment PreCheck Automation Test Suite | Environment PreCheck Suite +--- + +# 🔍 Environment PreCheck Suite -> 💡 **Core Objective**: Prior to model deployment or training task execution, conduct comprehensive health checks on critical cluster capabilities to proactively identify potential issues in SSH configuration, device status, network connectivity, TLS encryption, model integrity, and storage performance. +> 💡 **Core Objective**: Before model deployment or training tasks begin, perform comprehensive health checks on key cluster capabilities to proactively identify potential issues in SSH configuration, device status, network connectivity, TLS encryption, model integrity, and storage performance. --- ## 📋 Table of Contents -- [🌟 Core Features](#-core-features) -- [🎯 Functional Overview](#-functional-overview) -- [🚀 Quick Start](#-quick-start) -- [📁 Project Structure](#-project-structure) -- [🧪 Test Case Details](#-test-case-details) -- [⚙️ Configuration](#️-configuration) +* [🌟 Key Features](#-key-features) +* [🎯 Overview](#-overview) +* [⚙️ Configuration](#️-configuration) +* [🚀 Quick Start](#-quick-start) +* [🧪 Test Case Details](#-test-case-details) --- -## 🌟 Core Features +## 🌟 Key Features -| Feature | Description | -|------|------| -| 🎯 **Intelligent Platform Detection** | Automatically detect NPU (Ascend) / GPU environments and execute corresponding test logic | -| 🔧 **Modular Architecture** | Support flexible combination of test cases by stage, platform, and feature dimensions | -| 🛡️ **Fail-Fast Mechanism** | Immediately terminate upon critical check failures to avoid invalid resource consumption | -| 📊 **Comprehensive Coverage** | 6 major dimensions, 12+ detailed checks to ensure environmental consistency | +| Feature | Description | +| ------------------------------------- | ------------------------------------------------------------------------------------------- | +| 🎯 **Intelligent Platform Detection** | Automatically detects NPU (Ascend) / GPU environments and executes corresponding test logic | +| 🔧 **Modular Architecture** | Supports flexible combination of test cases by stage, platform, and feature | +| 🛡️ **Failure Fast Mechanism** | Critical check failures immediately terminate execution to avoid wasted resources | +| 📊 **Comprehensive Coverage** | Covers 6 major dimensions and 12+ checks to ensure environment consistency | --- -## 🎯 Functional Overview +## 🎯 Overview -This test suite automatically executes the following health checks prior to task initiation: +This test suite automatically performs the following health checks before task execution: ### 🔐 Infrastructure Layer -- **SSH Passwordless Login Verification**: Ensure bidirectional passwordless access between Master ↔ Worker is functional -- **Device Health Inspection**: NPU/GPU online status, driver version, and temperature monitoring +* **SSH Passwordless Login Verification**: Ensures bidirectional password-free access between Master ↔ Worker +* **Device Health Inspection**: NPU/GPU status, driver versions, temperature monitoring ### 🌐 Network Communication Layer -- **Inter-Node Connectivity Testing**: HCCN (NPU) / NVLink & InfiniBand (GPU) link packet loss detection -- **TLS Encryption Status**: Verify Ascend device TLS switch configuration complies with security baselines +* **Node Connectivity Testing**: Packet loss detection for HCCN (NPU) / NVLink & InfiniBand (GPU) +* **TLS Encryption Status**: Verifies whether TLS settings between Ascend devices meet security baselines ### 💾 Data Integrity Layer -- **Model Weight Vault**: File list integrity scan → MD5/SHA256 hash verification → weight format validity validation +* **Model Weights Vault**: File list integrity scan → MD5/SHA256 hash verification → weight format validation ### ⚡ Performance Baseline Layer -- **Storage Bandwidth Stress Test**: Compare measured Embedding/Fetch operation bandwidth against expected thresholds (< 85% triggers warning) +* **Storage Bandwidth Benchmarking**: Compare measured Dump/Load bandwidth against expected thresholds (< 85% triggers warning) + +--- + +## ⚙️ Configuration (`config.yaml`) + +| Parameter | Type | Description | Example | +| --------------------------- | ------ | ---------------------------------------------- | ---------------------------------- | +| `master_ip` | string | Master node SSH IP | `192.168.1.10` | +| `worker_ip` | list | Worker node IP list (empty = single-node test) | `["192.168.1.11", "192.168.1.12"]` | +| `ascend_rt_visible_devices` | string | Visible NPU device indices | `"0,1,2,3,4,5,6,7"` | +| `node_num` | int | Total number of nodes (master + workers) | `2` | +| `model_path` | string | Root directory of model weights | `/data/models/llama-7b` | +| `hf_model_name` | string | HuggingFace model identifier | `meta-llama/Llama-2-7b` | +| `middle_page` | string | Model intermediate page / organization name | `model_storage` | +| `expected_embed_bandwidth` | float | Expected Dump bandwidth (GB/s) | `12.0` | +| `expected_fetch_bandwidth` | float | Expected Load bandwidth (GB/s) | `8.0` | +| `storage_backends` | list | Mounted storage paths in current environment | `["/mnt/nfs"]` | --- @@ -57,31 +75,30 @@ This test suite automatically executes the following health checks prior to task ```bash tests/ ├── common/envPreCheck/ -│ ├── run_env_preCheck.py # Core detection engine -│ └── utils/ # Auxiliary utilities +│ ├── run_env_preCheck.py # Core check engine ├── suites/E2E/ │ └── test_environment_precheck.py # Test entry point -└── config.yaml # PreCheck threshold configuration file +└── config.yaml # Precheck threshold configuration ``` -### 🎮 Execution Methods +### 🎮 How to Run ```bash -# Enter test directory +# Navigate to test directory cd tests/ -# 1️⃣ Execute full precheck (stage 2) -pytest --stage=2 - -# 2️⃣ Execute by hardware platform +# 1️⃣ Run by hardware platform pytest --platform=npu # Ascend NPU environment pytest --platform=gpu # NVIDIA GPU environment -# 3️⃣ Execute by individual feature +# 2️⃣ Run by specific feature pytest --feature=test_ssh_login pytest --feature=test_check_bandwidth -# 4️⃣ Run specific file directly +# 3️⃣ Run full precheck (stage 2) +pytest --stage=2 + +# 4️⃣ Run a specific test file directly pytest suites/E2E/test_environment_precheck.py -v ``` @@ -95,8 +112,8 @@ pytest suites/E2E/test_environment_precheck.py -v test_ssh_login() ``` -- **Verification Content**: Master → Worker bidirectional passwordless login -- **Failure Strategy**: ❌ **Immediate termination** of all subsequent tests (blocking issue) +* **Validation**: Bidirectional passwordless login between Master → Worker +* **Failure Strategy**: ❌ **Immediately aborts** all subsequent tests (blocking issue) ### 🖥️ Device Status Check @@ -108,21 +125,21 @@ test_hccn_check_device_status() test_nvidia_check_device_status() ``` -- **Check Items**: Device online status, driver loading, memory health, temperature thresholds +* **Checks**: Device availability, driver status, memory health, temperature thresholds ### 🌐 Inter-Node Network Quality ```python -test_check_hccn_ping() # NPU: HCCN links -test_check_nvidia_ping() # GPU: NCCL networks +test_check_hccn_ping() # NPU: HCCN link +test_check_nvidia_ping() # GPU: NCCL network ``` -**Generate full-link topology report**: +**Generates full topology report**: ``` ✅ local_card_0 → local_card_1 [0.02ms, 0% loss] ✅ local_card_0 → remote_ip:192.168.1.10 [0.15ms, 0% loss] -⚠️ remote_card_1 → local_ip:192.168.1.5 [2.34ms, 3% loss] ← Abnormal link +⚠️ remote_card_1 → local_ip:192.168.1.5 [2.34ms, 3% loss] ← abnormal link ``` ### 🔒 TLS Security Configuration @@ -131,8 +148,8 @@ test_check_nvidia_ping() # GPU: NCCL networks test_check_tls() ``` -- **Check Target**: `tls_switch` status of each card -- **Pass Criteria**: All device TLS switches are consistent and comply with security policies (usually 0 or 1) +* **Check Target**: `tls_switch` status on each device +* **Pass Criteria**: All devices have consistent TLS settings and comply with security policy (typically 0 or 1) ### 📦 Model Weight Integrity @@ -142,54 +159,22 @@ test_check_model_weights() **Three-layer protection system**: -1. **File Tree Scanning**: Confirm existence of all `.bin`, `.safetensors`, `.json` files -2. **Hash Verification**: Compare against pre-computed checksums to prevent transmission corruption -3. **Format Validity**: Rapid loading validation using `torch.load` / `safetensors` +1. **File Tree Scan**: Ensure all `.bin`, `.safetensors`, `.json` files exist +2. **Hash Verification**: Compare precomputed checksums to prevent corruption +3. **Format Validation**: Quick load validation using `torch.load` / `safetensors` -### ⚡ Storage Bandwidth Benchmark Test +### ⚡ Storage Bandwidth Benchmark ```python test_check_bandwidth() ``` -- **Test Scenario**: Large-scale Embedding reads / Checkpoint Fetch writes -- **Decision Logic**: +* **Test Scenario**: Large-scale embedding read / checkpoint fetch write +* **Evaluation Logic**: + ```python if actual_bandwidth < expected_threshold * 0.85: - raise PerformanceWarning("Insufficient storage bandwidth may impact training efficiency") + raise PerformanceWarning("Insufficient storage bandwidth, may impact training efficiency") ``` --- - -## ⚙️ Configuration (`config.yaml`) - -| Configuration Item | Type | Description | Example Value | -|--------|------|------|--------| -| `master_ip` | string | SSH login IP of the cluster master node | `192.168.1.10` | -| `worker_ip` | list | IP addresses of cluster worker nodes | `["192.168.1.11", "192.168.1.12"]` | -| `ascend_rt_visible_devices` | string | Visible device IDs for Ascend/NPU | `"0,1,2,3,4,5,6,7"` | -| `node_num` | int | Total number of cluster nodes | `2` | -| `model_path` | string | Root directory of model weights | `/data/models/llama-7b` | -| `hf_model_name` | string | HuggingFace model identifier | `meta-llama/Llama-2-7b` | -| `middle_page` | string | Intermediate page/organization name corresponding to the model | `model_storage` | -| `expected_embed_bandwidth` | float | Expected embedding bandwidth (GB/s) | `12.0` | -| `expected_fetch_bandwidth` | float | Expected fetch bandwidth (GB/s) | `8.0` | -| `kvCache_block_number` | int | Number of KV Cache pre-allocated blocks | `4096` | -| `storage_backends` | list | Storage backend mount paths | `["/data", "/mnt/nfs"]` | - ---- - -## 🎨 Output Example - -```diff -🚀 Launching Environment PreCheck Suite (Platform: NPU) -━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ -✅ [PASS] SSH Passwordless Login (2/2 nodes) -✅ [PASS] NPU Device Status (8/8 cards online) -✅ [PASS] HCCN Link Connectivity (56/56 links) -⚠️ [WARN] TLS Configuration (card_3: tls_switch=1, expected=0) -✅ [PASS] Model Weight Integrity (hash verified) -❌ [FAIL] Storage Bandwidth Check (6.5 GB/s < 12.0 GB/s * 0.85) -━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ -🔴 PreCheck failed, please fix high-priority issues before launching training tasks -``` \ No newline at end of file diff --git a/test/common/envPreCheck/EnvPreCheck_ch.md b/test/common/envPreCheck/EnvPreCheck_ch.md index ca85167e6..f648c628b 100644 --- a/test/common/envPreCheck/EnvPreCheck_ch.md +++ b/test/common/envPreCheck/EnvPreCheck_ch.md @@ -1,4 +1,3 @@ -我来帮你重新润色这份技术文档，让它更加专业且富有视觉层次： --- @@ -13,10 +12,9 @@ - [🌟 核心特性](#-核心特性) - [🎯 功能概述](#-功能概述) +- [⚙️ 配置说明](#️-配置说明) - [🚀 快速开始](#-快速开始) -- [📁 项目结构](#-项目结构) - [🧪 测试用例详解](#-测试用例详解) -- [⚙️ 配置说明](#️-配置说明) --- @@ -51,7 +49,23 @@ ### ⚡ 性能基线层 -- **存储带宽压测**：Embedding/Fetch 操作实测带宽对比预期阈值（< 85% 触发告警） +- **存储带宽压测**：Dump/Load 操作实测带宽对比预期阈值（< 85% 触发告警） + +--- +## ⚙️ 配置说明（`config.yaml`） + +| 配置项 | 类型 | 说明 | 示例值 | +|--------|------|------|--------| +| `master_ip` | string | Master 节点 SSH IP | `192.168.1.10` | +| `worker_ip` | list | Worker 节点 IP 列表（若不填则表示单节点自测） | `["192.168.1.11", "192.168.1.12"]` | +| `ascend_rt_visible_devices` | string | NPU 可见设备序号 | `"0,1,2,3,4,5,6,7"` | +| `node_num` | int | master和worker总节点数 | `2` | +| `model_path` | string | 模型权重根目录 | `/data/models/llama-7b` | +| `hf_model_name` | string | HuggingFace 模型标识 | `meta-llama/Llama-2-7b` | +| `middle_page` | string | 模型中间页/组织名称 | `model_storage` | +| `expected_embed_bandwidth` | float | 预期 Dump 带宽 (GB/s) | `12.0` | +| `expected_fetch_bandwidth` | float | 预期 Load 带宽 (GB/s) | `8.0` | +| `storage_backends` | list | 当前容器所在环境的挂载点路径 | `["/mnt/nfs"]` | --- @@ -63,7 +77,6 @@ tests/ ├── common/envPreCheck/ │ ├── run_env_preCheck.py # 核心检测引擎 -│ └── utils/ # 辅助工具集 ├── suites/E2E/ │ └── test_environment_precheck.py # 测试入口 └── config.yaml # 预检阈值配置文件 @@ -75,17 +88,17 @@ tests/ # 进入测试目录 cd tests/ -# 1️⃣ 执行完整预检（阶段 2） -pytest --stage=2 - -# 2️⃣ 按硬件平台执行 +# 1️⃣ 按硬件平台执行 pytest --platform=npu # Ascend NPU 环境 pytest --platform=gpu # NVIDIA GPU 环境 -# 3️⃣ 按特性单独执行 +# 2️⃣ 按特性单独执行 pytest --feature=test_ssh_login pytest --feature=test_check_bandwidth +# 3️⃣ 执行完整预检（阶段 2） +pytest --stage=2 + # 4️⃣ 直接运行特定文件 pytest suites/E2E/test_environment_precheck.py -v ``` @@ -164,39 +177,4 @@ test_check_bandwidth() raise PerformanceWarning("存储带宽不足，可能影响训练效率") ``` ---- - -## ⚙️ 配置说明（`config.yaml`） - -| 配置项 | 类型 | 说明 | 示例值 | -|--------|------|------|--------| -| `master_ip` | string | Master 节点 SSH IP | `192.168.1.10` | -| `worker_ip` | list | Worker 节点 IP 列表 | `["192.168.1.11", "192.168.1.12"]` | -| `ascend_rt_visible_devices` | string | NPU 可见设备序号 | `"0,1,2,3,4,5,6,7"` | -| `node_num` | int | 集群总节点数 | `2` | -| `model_path` | string | 模型权重根目录 | `/data/models/llama-7b` | -| `hf_model_name` | string | HuggingFace 模型标识 | `meta-llama/Llama-2-7b` | -| `middle_page` | string | 中间页/组织名称 | `model_storage` | -| `expected_embed_bandwidth` | float | 预期 Embedding 带宽 (GB/s) | `12.0` | -| `expected_fetch_bandwidth` | float | 预期 Fetch 带宽 (GB/s) | `8.0` | -| `kvCache_block_number` | int | KV Cache 预分配块数 | `4096` | -| `storage_backends` | list | 存储后端挂载路径 | `["/data", "/mnt/nfs"]` | - ---- - -## 🎨 输出示例 - -```diff -🚀 启动环境预检套件 (Platform: NPU) -━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ -✅ [PASS] SSH 免密登录 (2/2 nodes) -✅ [PASS] NPU 设备状态 (8/8 cards online) -✅ [PASS] HCCN 链路连通性 (56/56 links) -⚠️ [WARN] TLS 配置 (card_3: tls_switch=1, expected=0) -✅ [PASS] 模型权重完整性 (hash verified) -❌ [FAIL] 存储带宽检测 (6.5 GB/s < 12.0 GB/s * 0.85) -━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ -🔴 预检未通过，请修复高优问题后再启动训练任务 -``` - --- \ No newline at end of file diff --git a/test/common/envPreCheck/run_env_preCheck.py b/test/common/envPreCheck/run_env_preCheck.py index daae41032..ce277b7e4 100644 --- a/test/common/envPreCheck/run_env_preCheck.py +++ b/test/common/envPreCheck/run_env_preCheck.py @@ -1,3 +1,5 @@ +import mmap +import multiprocessing import os import re import secrets @@ -9,8 +11,11 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Tuple +import numpy as np import yaml +from ucm.store.factory_v1 import UcmConnectorFactoryV1, UcmKVStoreBaseV1 + CODE_ROOT = Path(__file__).resolve().parent Custom_SSH_DIR = (CODE_ROOT / "ssh_keys").resolve() Custom_SSH_DIR.mkdir(parents=True, exist_ok=True) @@ -31,9 +36,12 @@ HF_MODEL_NAME = config.get("Env_preCheck", {}).get("hf_model_name", "") MIDDLE_PAGE = config.get("Env_preCheck", {}).get("middle_page", "") - KVCACHE_BLOCK_NUMBER = config.get("Env_preCheck", {}).get( - "kvCache_block_number", "" - ) + WORKER_NUMBER = config.get("Env_preCheck", {}).get("worker_number", 1) + SHARD_SIZE = config.get("Env_preCheck", {}).get("shard_size", 1 * 1024 * 1024) + SHARD_NUMBER = config.get("Env_preCheck", {}).get("shard_number", 1) + BLOCK_NUMBER = config.get("Env_preCheck", {}).get("block_number", 64) + DUMP_EPOCH_NUMBER = config.get("Env_preCheck", {}).get("dump_epoch_number", 32) + LOAD_EPOCH_NUMBER = config.get("Env_preCheck", {}).get("load_epoch_number", 32) STORAGE_BACKENDS = config.get("Env_preCheck", {}).get("storage_backends", "") @@ -476,7 +484,6 @@ def run_hccn_device_status_check(): run_check "for i in $(echo \"$ASCEND_RT_VISIBLE_DEVICES\" | tr ',' ' ' | xargs); do hccn_tool -i \$i -link -g; done" "physical_link_status" run_check "for i in $(echo \"$ASCEND_RT_VISIBLE_DEVICES\" | tr ',' ' ' | xargs); do hccn_tool -i \$i -net_health -g; done" "network_health_status" run_check "for i in $(echo \"$ASCEND_RT_VISIBLE_DEVICES\" | tr ',' ' ' | xargs); do hccn_tool -i \$i -ip -g; done" "gpu_ip_config" -run_check "for i in $(echo \"$ASCEND_RT_VISIBLE_DEVICES\" | tr ',' ' ' | xargs); do hccn_tool -i \$i -gateway -g; done" "gateway_config" """ print("\n----------------------------------------\n") print(f"[UC] Starting to check the Ascend card status, node IP: {MASTER_IP}") @@ -490,6 +497,12 @@ def run_hccn_device_status_check(): # Parsing Output result_dict = parse_hccn_output(output) + overall_status = all( + v.get("status", False) for v in result_dict.values() if isinstance(v, dict) + ) + + result_dict["device_status_check"] = {"status": overall_status} + return result_dict @@ -690,33 +703,46 @@ def remote_local_hccn_cards_ping_test( :param dst_ip: destination IP :return: dict -> { "card_X_to_card_Y": {"status": True/False, "output": "..."}, "global_status": {...} } """ + HCCN_TOOL = "/usr/local/Ascend/driver/tools/hccn_tool" + if src_type == "local": - print(f"[TEST] LOCAL card {src_card} → REMOTE {dst_ip}") + print(f"[TEST] LOCAL card {src_card} to REMOTE {dst_ip}") pair_key = f"local_card_{src_card} to remote_{dst_ip.replace('.', '_')}" - cmd = f"hccn_tool -i {src_card} -ping -g address {dst_ip}" + + cmd = ( + f"ssh -q -o LogLevel=ERROR " + f"-o StrictHostKeyChecking=no " + f"-o UserKnownHostsFile=/dev/null " + f'-i "{str(LOCAL_SSH_KEY)}" ' + f"root@{MASTER_IP} {HCCN_TOOL} -i {src_card} -ping -g address {dst_ip}" + ) else: - print(f"[TEST] REMOTE {src_ip} card {src_card} → LOCAL {dst_ip}") + print(f"[TEST] REMOTE {src_ip} card {src_card} to LOCAL {dst_ip}") pair_key = f"remote_card_{src_card} to local_{dst_ip.replace('.', '_')}" - cmd = f'ssh -i "{str(LOCAL_SSH_KEY)}" root@{src_ip} hccn_tool -i {src_card} -ping -g address {dst_ip}' + + cmd = ( + f"ssh -q -o LogLevel=ERROR " + f"-o StrictHostKeyChecking=no " + f"-o UserKnownHostsFile=/dev/null " + f'-i "{str(LOCAL_SSH_KEY)}" ' + f"root@{src_ip} {HCCN_TOOL} -i {src_card} -ping -g address {dst_ip}" + ) result_dict = {pair_key: {"status": False, "output": ""}} try: completed = subprocess.run(cmd, shell=True, capture_output=True, text=True) - if completed.returncode == 0 and src_type == "local": - print(f"[RESULT] LOCAL card {src_card} → REMOTE {dst_ip}: Success") - elif completed.returncode == 0 and src_type == "remote": - print(f"[RESULT] REMOTE {src_ip} card {src_card} → LOCAL {dst_ip}: Success") - elif src_type == "local": - print(f"[RESULT] LOCAL card {src_card} → REMOTE {dst_ip}: Failed") - elif src_type == "remote": - print(f"[RESULT] REMOTE {src_ip} card {src_card} → LOCAL {dst_ip}: Failed") + + if completed.returncode == 0: + print(f"[RESULT] {pair_key}: SUCCESS") + else: + print(f"[RESULT] {pair_key}: FAILED") result_dict[pair_key]["status"] = completed.returncode == 0 except Exception as e: + print(f"[ERROR] Ping test failed: {e}", file=sys.stderr) result_dict[pair_key]["status"] = False - # Calculate global status (only one ping here, caller can aggregate multiple pings) result_dict["global_status"] = { "status": result_dict[pair_key]["status"], "output": "HCCN local/remote ping global status", @@ -1119,282 +1145,133 @@ def run_check_model_weight(): return result_dict -class StdoutInterceptor: - """ - Intercepts all stdout and stderr output from both C++ and Python code, - and prevents it from being printed directly to the console. +def create_worker(device_id: int) -> UcmKVStoreBaseV1: + module_path = "ucm.store.pipeline.connector" + class_name = "UcmPipelineStore" + config = {} + config["store_pipeline"] = "Posix" + config["posix_io_engine"] = "aio" + config["storage_backends"] = STORAGE_BACKENDS + config["tensor_size"] = SHARD_SIZE + config["shard_size"] = SHARD_SIZE + config["device_id"] = device_id + return UcmConnectorFactoryV1.create_connector(class_name, config, module_path) + + +def make_array(size, alignment=262144, dtype=np.uint8) -> tuple[np.ndarray, mmap.mmap]: + itemsize = np.dtype(dtype).itemsize + total_bytes = size * itemsize + mm = mmap.mmap(-1, total_bytes + alignment) + raw_array = np.frombuffer(mm, dtype=np.uint8, count=total_bytes + alignment) + raw_ptr = raw_array.__array_interface__["data"][0] + aligned_addr = (raw_ptr + alignment - 1) & ~(alignment - 1) + offset = aligned_addr - raw_ptr + array = raw_array[offset : offset + total_bytes].view(dtype=dtype) + return array, mm + + +def dump(epoch, device_id, worker, block_ids, block_ptr, dump_bw_list): + total_size = SHARD_SIZE * SHARD_NUMBER * BLOCK_NUMBER + costs = [] + for i in range(SHARD_NUMBER): + idxes = [i for _ in range(BLOCK_NUMBER)] + ptrs = [[ptr + i * SHARD_SIZE] for ptr in block_ptr] + tp = time.perf_counter() + task = worker.dump_data(block_ids, idxes, ptrs) + worker.wait(task) + costs.append(time.perf_counter() - tp) + total_cost = np.sum(costs) + bw = total_size / total_cost / 1e9 + + dump_bw_list.append(bw) + + print( + f"epoch={epoch:03}, worker={device_id:02}, " + f"dump=[{SHARD_SIZE} x {BLOCK_NUMBER} x {SHARD_NUMBER}], " + f"avg_cost={np.average(costs) * 1e3:.3f}ms, " + f"p99_cost={np.percentile(costs, 99) * 1e3:.3f}ms, " + f"total_cost={total_cost * 1e3:.3f}ms, " + f"bw={bw:.3f}GB/s." + ) - This is useful for capturing logs programmatically for filtering or analysis. - """ - def __enter__(self): - # Save original stdout and stderr file descriptors - self.original_stdout = os.dup(1) - self.original_stderr = os.dup(2) +def load(epoch, device_id, worker, block_ids, block_ptr, load_bw_list): + total_size = SHARD_SIZE * SHARD_NUMBER * BLOCK_NUMBER + costs = [] + for i in range(SHARD_NUMBER): + idxes = [i for _ in range(BLOCK_NUMBER)] + ptrs = [[ptr + i * SHARD_SIZE] for ptr in block_ptr] + tp = time.perf_counter() + task = worker.load_data(block_ids, idxes, ptrs) + worker.wait(task) + costs.append(time.perf_counter() - tp) + total_cost = np.sum(costs) + bw = total_size / total_cost / 1e9 + + load_bw_list.append(bw) + + print( + f"epoch={epoch:03}, worker={device_id:02}, " + f"load=[{SHARD_SIZE} x {BLOCK_NUMBER} x {SHARD_NUMBER}], " + f"avg_cost={np.average(costs) * 1e3:.3f}ms, " + f"p99_cost={np.percentile(costs, 99) * 1e3:.3f}ms, " + f"total_cost={total_cost * 1e3:.3f}ms, " + f"bw={bw:.3f}GB/s." + ) - # Create a pipe to capture output - self.pipe_out_r, self.pipe_out_w = os.pipe() - # Redirect stdout and stderr to the pipe - os.dup2(self.pipe_out_w, 1) - os.dup2(self.pipe_out_w, 2) - os.close(self.pipe_out_w) +def worker_loop(device_id, barrier, dump_bw_list, load_bw_list): + store = create_worker(device_id) + block_ids = [secrets.token_bytes(16) for _ in range(BLOCK_NUMBER)] - self.logs = [] - self._stop_thread = False + mmap_handles = [] + block_data = [] + for _ in range(BLOCK_NUMBER): + arr, mm = make_array(SHARD_SIZE * SHARD_NUMBER) + block_data.append(arr) + mmap_handles.append(mm) - # Start a background thread to read from the pipe continuously - self.thread = threading.Thread(target=self._read_pipe) - self.thread.daemon = True - self.thread.start() - return self + block_ptr = [block.ctypes.data for block in block_data] + barrier.wait() - def _read_pipe(self): - # Continuously read from the pipe until stopped - while not self._stop_thread: - try: - chunk = os.read(self.pipe_out_r, 4096) - if chunk: - text = chunk.decode() - self.logs.append(text) - # Do not print to terminal; logs are kept internally - else: - time.sleep(0.01) - except OSError: - break + for epoch in range(DUMP_EPOCH_NUMBER): + dump(epoch, device_id, store, block_ids, block_ptr, dump_bw_list) + barrier.wait() - def __exit__(self, exc_type, exc_val, exc_tb): - # Stop background thread and restore stdout/stderr - self._stop_thread = True - time.sleep(0.05) + for epoch in range(LOAD_EPOCH_NUMBER): + load(epoch, device_id, store, block_ids, block_ptr, load_bw_list) + barrier.wait() + + for mm in mmap_handles: try: - os.close(self.pipe_out_r) - except OSError: + mm.close() + except Exception: pass - os.dup2(self.original_stdout, 1) - os.dup2(self.original_stderr, 2) - os.close(self.original_stdout) - os.close(self.original_stderr) - - def read(self): - """Return all captured logs as a single string.""" - return "".join(self.logs) - - -def setup_uc(block_size): - """ - Initialize UC (Unified Cache) with a given block size. - - Args: - block_size (int): Total block size in bytes for UC setup. - - Raises: - RuntimeError: if ucmstore.Setup returns a non-zero value. - """ - import ucmstore - - param = ucmstore.SetupParam(STORAGE_BACKENDS, block_size, True) - ret = ucmstore.Setup(param) - if ret != 0: - raise RuntimeError(f"ucmstore.Setup failed: ret={ret}") - - -def filter_task_logs(logs): - """ - Filter UC output logs to extract only lines containing Task information, - including task_id and bandwidth. - - Args: - logs (str): Raw UC logs. - - Returns: - str: Filtered log lines, suitable for printing. - """ - filtered_lines = [] - for line in logs.splitlines(): - m = re.search(r"(Task\(\d+,[^\)]*\).*?bw=[\d\.]+GB/s)", line) - if m: - filtered_lines.append(m.group(1)) - return "\n".join(filtered_lines) - - -def embed(hashes, block_layer_size, block_layer): - """ - Execute UC embedding (writing KVCache blocks) operation and measure bandwidth. - - Args: - hashes (list[str]): List of block hashes to embed. - block_layer_size (int): Size of each block layer in bytes. - block_layer (int): Number of layers per block. - - Returns: - float | None: Average bandwidth in GB/s, or None if no valid bw found. - - Raises: - RuntimeError: If any UC operation fails. - """ - import ucmstore - - with StdoutInterceptor() as cap: - # Allocate blocks in UC - ret = ucmstore.AllocBatch(hashes) - if sum(ret) != 0: - raise RuntimeError(f"ucmstore.AllocBatch failed: sum(ret)={sum(ret)} != 0") - - block_number = len(hashes) - buffers = ucmstore.MakeHostBuffers(block_layer_size, block_layer * block_number) - if len(buffers) == 0: - raise RuntimeError("ucmstore.MakeHostBuffers failed: no buffers allocated") - - # Prepare data for DumpFromHost - data_id, data_off, data_addr, data_len = [], [], [], [] - for block_idx in range(block_number): - offset = 0 - for layer_idx in range(block_layer): - data_id.append(hashes[block_idx]) - data_off.append(offset) - data_addr.append(buffers[block_idx * block_layer + layer_idx]) - data_len.append(block_layer_size) - offset += block_layer_size - - # Dump data to UC - task_id = ucmstore.DumpFromHost(data_id, data_off, data_addr, data_len) - if task_id <= 0: - raise RuntimeError( - f"ucmstore.DumpFromHost failed: invalid task_id={task_id}" - ) - - # Wait for completion - ret = ucmstore.Wait(task_id) - if ret != 0: - raise RuntimeError( - f"ucmstore.Wait failed for embed task_id={task_id}, ret={ret}" - ) - - # Release host buffers and commit - ucmstore.ReleaseHostBuffers(buffers) - ucmstore.CommitBatch(hashes, True) - - logs = cap.read() - print(filter_task_logs(logs)) - - # Extract average bandwidth - bw_list = [float(x) for x in re.findall(r"bw=([\d\.]+)GB/s", logs)] - avg_bw = sum(bw_list) / len(bw_list) if bw_list else None - return avg_bw -def fetch(hashes, block_layer_size, block_layer): - """ - Execute UC fetching (reading KVCache blocks) operation and measure bandwidth. - - Args: - hashes (list[str]): List of block hashes to fetch. - block_layer_size (int): Size of each block layer in bytes. - block_layer (int): Number of layers per block. - - Returns: - float | None: Average bandwidth in GB/s, or None if no valid bw found. - - Raises: - RuntimeError: If any UC operation fails. - """ - import ucmstore - - with StdoutInterceptor() as cap: - block_number = len(hashes) - results = ucmstore.LookupBatch(hashes) - if not all(results): - raise RuntimeError("ucmstore.LookupBatch failed: some blocks not found") - - buffers = ucmstore.MakeHostBuffers(block_layer_size, block_layer * block_number) - if len(buffers) == 0: - raise RuntimeError("ucmstore.MakeHostBuffers failed: no buffers allocated") - - # Prepare data for LoadToHost - data_id, data_off, data_addr, data_len = [], [], [], [] - for block_idx in range(block_number): - offset = 0 - for layer_idx in range(block_layer): - data_id.append(hashes[block_idx]) - data_off.append(offset) - data_addr.append(buffers[block_idx * block_layer + layer_idx]) - data_len.append(block_layer_size) - offset += block_layer_size - - # Load data from UC - task_id = ucmstore.LoadToHost(data_id, data_off, data_addr, data_len) - if task_id <= 0: - raise RuntimeError("ucmstore.LoadToHost failed: invalid task_id") - - # Wait for completion - ret = ucmstore.Wait(task_id) - if ret != 0: - raise RuntimeError( - f"ucmstore.Wait failed for fetch task_id={task_id}, ret={ret}" - ) +def run_bandwidth_check(): + manager = multiprocessing.Manager() + dump_bw_list = manager.list() + load_bw_list = manager.list() - # Release buffers - ucmstore.ReleaseHostBuffers(buffers) + barrier = multiprocessing.Barrier(WORKER_NUMBER) + workers = [] - logs = cap.read() - print(filter_task_logs(logs)) + for i in range(WORKER_NUMBER): + p = multiprocessing.Process( + target=worker_loop, args=(i, barrier, dump_bw_list, load_bw_list) + ) + workers.append(p) + p.start() - # Extract average bandwidth - bw_list = [float(x) for x in re.findall(r"bw=([\d\.]+)GB/s", logs)] - avg_bw = sum(bw_list) / len(bw_list) if bw_list else None - return avg_bw + for w in workers: + w.join() + avg_dump = np.mean(dump_bw_list) if len(dump_bw_list) > 0 else 0.0 + avg_load = np.mean(load_bw_list) if len(load_bw_list) > 0 else 0.0 -# ========= Bandwidth Check ========= -def run_bandwidth_check(): - """ - Run UC embedding and fetching operations on KVCache blocks, - measure bandwidth for each batch, and calculate overall average. - - Returns: - dict: Summary of average bandwidth for 'embed' and 'fetch' in GB/s. - """ - # UC block and layer configuration - block_dim = 576 - block_len = 128 - block_elem_size = 2 - block_layer = 61 - block_layer_size = block_dim * block_len * block_elem_size - block_size = block_layer_size * block_layer - batch_size = 256 - - setup_uc(block_size) - hashes = [secrets.token_hex(16) for _ in range(KVCACHE_BLOCK_NUMBER)] - total_batches = (KVCACHE_BLOCK_NUMBER + batch_size - 1) // batch_size - - bw_summary = {"embed": [], "fetch": []} - - print("\n----------------------------------------") - print("[UC] Start embed batch and fetch batch. Processing KVCache blocks...") - - # Embed batches - for batch in range(total_batches): - start = batch_size * batch - end = min(start + batch_size, KVCACHE_BLOCK_NUMBER) - avg_bw = embed(hashes[start:end], block_layer_size, block_layer) - if avg_bw: - bw_summary["embed"].append(avg_bw) - - print("[UC] Start fetch batch. Processing KVCache blocks...") - # Fetch batches - for batch in range(total_batches): - start = batch_size * batch - end = min(start + batch_size, KVCACHE_BLOCK_NUMBER) - avg_bw = fetch(hashes[start:end], block_layer_size, block_layer) - if avg_bw: - bw_summary["fetch"].append(avg_bw) - - # Calculate overall average bandwidth - for key in bw_summary: - if bw_summary[key]: - bw_summary[key] = sum(bw_summary[key]) / len(bw_summary[key]) - else: - bw_summary[key] = None + print(f"\n==== FINAL AVERAGE BANDWIDTH ====") + print(f" Dump BW (avg): {avg_dump:.3f} GB/s") + print(f" Load BW (avg): {avg_load:.3f} GB/s") - return bw_summary + return {"dump": round(avg_dump, 3), "load": round(avg_load, 3)} diff --git a/test/common/llmperf/LLMPerf.md b/test/common/llmperf/LLMPerf.md index 35822da1d..a6c98499c 100644 --- a/test/common/llmperf/LLMPerf.md +++ b/test/common/llmperf/LLMPerf.md @@ -69,16 +69,16 @@ graph TD ### 2️⃣ Load Parameter Configuration (`test_uc_performance.py`) -Defines **performance test load matrix**, supporting Cartesian product combinations of multiple parameter sets: +Defines **performance test load matrix**, The test cases are the same for each list at the corresponding positions: | Parameter | Type | Description | Example | |------|------|------|------| | `mean_input_tokens` | list[int] | Average input length distribution | `[512, 2048, 4096]` | -| `mean_output_tokens` | list[int] | Average output length distribution | `[128, 512]` | -| `concurrent_requests` | list[int] | Concurrent request count gradient | `[1, 4, 8, 16]` | -| `max_num_completed_requests` | list[int] | Maximum completed requests per round | `[100, 50]` | +| `mean_output_tokens` | list[int] | Average output length distribution | `[128, 512, 1024]` | +| `concurrent_requests` | list[int] | Concurrent request count gradient | `[1, 4, 8]` | +| `max_num_completed_requests` | list[int] | Maximum completed requests per round | `[100, 50, 50]` | | `hit_rate` | list[int] | Cache hit rate (%) | `[0, 50, 90]` | -| `random_seed` | list[int] | Random seed | `[42, 0]` | +| `random_seed` | list[int] | Random seed | `[42, 0, 0]` | #### 🎲 Random Seed Strategy diff --git a/test/common/llmperf/LLMPerf_zh.md b/test/common/llmperf/LLMPerf_zh.md index 3766d4dc7..82f6edea5 100644 --- a/test/common/llmperf/LLMPerf_zh.md +++ b/test/common/llmperf/LLMPerf_zh.md @@ -69,16 +69,16 @@ graph TD ### 2️⃣ 负载参数配置 (`test_uc_performance.py`) -定义**性能测试负载矩阵**，支持多组参数笛卡尔积组合： +定义**性能测试负载矩阵**，测试用例为每组list相同对应位置： | 参数 | 类型 | 说明 | 示例 | |------|------|------|------| | `mean_input_tokens` | list[int] | 平均输入长度分布 | `[512, 2048, 4096]` | -| `mean_output_tokens` | list[int] | 平均输出长度分布 | `[128, 512]` | -| `concurrent_requests` | list[int] | 并发请求数梯度 | `[1, 4, 8, 16]` | -| `max_num_completed_requests` | list[int] | 单轮最大完成数 | `[100, 50]` | +| `mean_output_tokens` | list[int] | 平均输出长度分布 | `[128, 512, 1024]` | +| `concurrent_requests` | list[int] | 并发请求数梯度 | `[1, 4, 8]` | +| `max_num_completed_requests` | list[int] | 单轮最大完成数 | `[100, 50, 50]` | | `hit_rate` | list[int] | 缓存命中率(%) | `[0, 50, 90]` | -| `random_seed` | list[int] | 随机种子 | `[42, 0]` | +| `random_seed` | list[int] | 随机种子 | `[42, 0, 0]` | #### 🎲 Random Seed 策略 diff --git a/test/config.yaml b/test/config.yaml index b9017fa29..9f37964cd 100644 --- a/test/config.yaml +++ b/test/config.yaml @@ -57,5 +57,9 @@ Env_preCheck: middle_page: "" expected_embed_bandwidth: 10 expected_fetch_bandwidth: 10 - kvCache_block_number: 1024 + worker_number: 1 + shard_size: 1 * 1024 * 1024 + block_number: 64 + dump_epoch_number: 32 + load_epoch_number: 32 storage_backends: [""] \ No newline at end of file