一、CANN 资源模型：理解 NPU 的硬件基础

在讨论调度策略之前，必须先理解昇腾 NPU 的硬件拓扑。每颗昇腾芯片内部包含多个 AI Core（计算核心）和一个/多个 CPU Core（控制核心），片上内存（L1/L2 Cache）和高带宽内存（HBM）通过 NoC（片上网络）互联。多颗芯片之间通过 HCCS（Hyperscale Cluster Connection Service）高速互联，形成机内多卡集群。

在 CANN 的视角下，一颗 NPU 被抽象为一个 Device，每个 Device 有独立的：

• AI Core：执行算子计算的并行计算单元
• 内存空间：片上 L2 Cache + 外部 HBM
• 任务队列：接收并排队来自用户的计算任务
• Stream：任务执行的逻辑通道，支持流水线并行

理解这些抽象至关重要——调度器的每一个决策（分配哪个 Device、使用多少内存、排在哪个队列）都直接影响任务的执行效率和隔离性。

1.1 Device 与 Context 的关系

在 CANN 编程模型中，一个进程通常通过 aclrtSetDevice 绑定到某颗 NPU。一旦绑定，该进程的所有内存分配和计算任务都定向到这颗 Device。多个进程绑定同一颗 Device 时，它们共享该 Device 的硬件资源，但 CANN 通过 Context 机制为每个进程维护独立的资源视图。

// 设置当前进程使用的 NPU 设备
aclError ret = aclrtSetDevice(device_id);
assert(ret == ACL_SUCCESS);

// 创建独立的 Context，隔离本进程的资源
aclrtContext context;
ret = aclrtCreateContext(&context, device_id);
assert(ret == ACL_SUCCESS);

// 切换到该 Context，后续操作都在此 Context 下执行
ret = aclrtSetCurrentContext(context);
assert(ret == ACL_SUCCESS);

为什么要创建 Context？ 因为同一个 Device 上可能运行多个进程。如果不创建 Context，进程间会共享同一个默认 Context，导致内存分配冲突、Stream 竞争等问题。显式创建 Context 相当于为每个进程开辟一个独立的"工作间"，互不干扰。

1.2 Stream：任务执行的逻辑通道

Stream 是 CANN 中任务调度的核心抽象。每个 Stream 内部的任务严格按提交顺序执行（FIFO），但多个 Stream 之间可以并行执行。这类似于 CPU 编程中的多线程——每个 Stream 是一条独立的执行流水线。

// 创建一个 Stream 用于异步任务执行
aclrtStream stream;
ret = aclrtCreateStream(&stream);
assert(ret == ACL_SUCCESS);

// 在 Stream 上提交一个矩阵乘法任务
// 矩阵乘法是 CANN 中最常见的计算密集型操作
ret = aclblasGemmEx(
    ACLblasRowMajor,      // 行优先存储
    ACLblasNoTrans,       // A 矩阵不转置
    ACLblasNoTrans,       // B 矩阵不转置
    M, N, K,              // 矩阵维度
    1.0f,                 // alpha 系数
    inputA,               // A 矩阵数据指针
    ACLblasFp16,          // A 矩阵数据类型
    K,                    // A 矩阵 leading dimension
    inputB,               // B 矩阵数据指针
    ACLblasFp16,          // B 矩阵数据类型
    N,                    // B 矩阵 leading dimension
    0.0f,                 // beta 系数
    outputC,              // C 矩阵输出指针
    ACLblasFp16,          // C 矩阵数据类型
    N,                    // C 矩阵 leading dimension
    ACLblasFp16Compute,   // 计算精度
    stream                // 指定 Stream
);
assert(ret == ACL_SUCCESS);

// 同步等待该 Stream 上所有任务完成
ret = aclrtSynchronizeStream(stream);
assert(ret == ACL_SUCCESS);

多 Stream 并行的实际意义：假设一个训练 Step 包含前向计算和梯度更新两个阶段。可以将前向计算提交到 Stream A，梯度更新提交到 Stream B。如果硬件支持，这两个任务可以流水线执行，提升整体吞吐。

1.3 内存分配与管理

NPU 的 HBM 是最宝贵的资源之一。CANN 提供了多种内存管理策略：

// 分配 NPU 设备内存（HBM 上）
void* device_ptr = nullptr;
size_t alloc_size = 1024 * 1024 * 1024;  // 1GB
aclError ret = aclrtMalloc(&device_ptr, alloc_size, ACL_MEM_MALLOC_HBM_FIRST);
assert(ret == ACL_SUCCESS);

// 分配 Host 内存（DMA 可访问，用于 Host-Device 传输）
void* host_ptr = nullptr;
ret = aclrtMallocHost(&host_ptr, alloc_size);
assert(ret == ACL_SUCCESS);

// Host → Device 数据传输（通过 DMA 引擎）
ret = aclrtMemcpy(device_ptr, alloc_size, host_ptr, alloc_size, ACL_MEMCPY_HOSTToDevice);
assert(ret == ACL_SUCCESS);

// 释放内存
ret = aclrtFree(device_ptr);
assert(ret == ACL_SUCCESS);
ret = aclrtFreeHost(host_ptr);
assert(ret == ACL_SUCCESS);

内存分配策略的选择：ACL_MEM_MALLOC_HBM_FIRST 优先从 HBM 分配，HBM 不足时回退到 Host 内存。对于训练场景，模型参数和激活值通常很大，必须确保 HBM 优先；对于推理场景，模型较小，可以灵活选择。

---

二、任务队列与优先级调度

CANN 的任务调度采用多级队列架构。每个 Device 内部维护了多个任务队列，不同优先级的任务被路由到不同队列。调度器按优先级从高到低轮询队列，确保高优先级任务（如在线推理请求）能及时获得执行资源。

2.1 优先级队列的创建与配置

// 创建一个高优先级 Stream（用于在线推理）
aclrtStream high_priority_stream;
aclrtStreamAttr attr;
aclrtInitStreamAttr(&attr, ACL_STREAM_ATTR_PRIORITY);
aclrtSetStreamAttrPriority(&attr, 0);  // 0 = 最高优先级
aclrtCreateStreamWithAttr(&high_priority_stream, &attr);

// 创建一个普通优先级 Stream（用于离线训练）
aclrtStream normal_stream;
aclrtInitStreamAttr(&attr, ACL_STREAM_ATTR_PRIORITY);
aclrtSetStreamAttrPriority(&attr, 5);  // 5 = 中等优先级
aclrtCreateStreamWithAttr(&normal_stream, &attr);

// 创建一个低优先级 Stream（用于后台数据预处理）
aclrtStream low_priority_stream;
aclrtInitStreamAttr(&attr, ACL_STREAM_ATTR_PRIORITY);
aclrtSetStreamAttrPriority(&attr, 10);  // 10 = 低优先级
aclrtCreateStreamWithAttr(&low_priority_stream, &attr);

2.2 调度器的工作原理

CANN 调度器在底层维护了以下状态：

• 就绪队列：已提交但尚未开始执行的任务
• 执行队列：正在 NPU 上执行的任务
• 等待队列：因资源不足（如显存不够）而暂时挂起的任务

调度决策的核心逻辑：

1. 检查高优先级队列是否有就绪任务
   → 有 → 检查 NPU 资源是否充足
     → 充足 → 取出任务，提交到 NPU 执行
     → 不足 → 放入等待队列，标记等待原因
   → 无 → 转到步骤 2

2. 检查中优先级队列是否有就绪任务
   → 同上逻辑

3. 检查低优先级队列
   → 同上逻辑

4. 检查等待队列
   → 有任务的等待条件已满足（如显存释放）→ 重新放入对应优先级的就绪队列

2.3 实际调度场景示例

考虑一个典型的多租户场景：

• 租户 A：在线推理服务，要求延迟 < 50ms，优先级高
• 租户 B：模型微调训练，优先级中
• 租户 C：数据预处理，优先级低

# 伪代码：基于优先级的调度策略
class PriorityScheduler:
    def __init__(self, npu_count=8):
        self.npu_count = npu_count
        self.queues = {
            'high': [],      # 在线推理
            'normal': [],    # 训练任务
            'low': []        # 数据预处理
        }
        self.active_tasks = {}  # device_id -> task
    
    def submit_task(self, task):
        """提交任务到对应优先级队列"""
        priority = task.get('priority', 'normal')
        self.queues[priority].append(task)
        print(f"[调度] 任务 {task['id']} 进入 {priority} 队列")
    
    def schedule(self):
        """调度决策：从高到低扫描队列"""
        for priority in ['high', 'normal', 'low']:
            if not self.queues[priority]:
                continue
            
            # 找到空闲的 NPU
            free_device = self._find_free_device()
            if free_device is None:
                break  # 所有 NPU 都在忙，等待
            
            task = self.queues[priority].pop(0)
            self._assign_task(task, free_device)
    
    def _find_free_device(self):
        for i in range(self.npu_count):
            if i not in self.active_tasks:
                return i
        return None
    
    def _assign_task(self, task, device_id):
        """将任务绑定到指定 NPU"""
        self.active_tasks[device_id] = task
        print(f"[调度] 任务 {task['id']} → NPU {device_id}")

# 使用示例
scheduler = PriorityScheduler(npu_count=8)

# 提交不同类型的任务
scheduler.submit_task({'id': 'infer-001', 'priority': 'high', 'model': 'resnet50'})
scheduler.submit_task({'id': 'train-001', 'priority': 'normal', 'model': 'bert-base'})
scheduler.submit_task({'id': 'preproc-001', 'priority': 'low', 'data': 'imagenet'})

# 执行调度
scheduler.schedule()

---

三、显存隔离：防止租户间资源侵占

在多租户环境中，最危险的问题之一是显存泄漏——某个租户的程序因 Bug 未释放显存，导致其他租户无法分配内存。CANN 提供了多层隔离机制来应对这一问题。

3.1 基于 Context 的内存隔离

每个 Context 拥有独立的内存分配器。一个 Context 中的 aclrtMalloc 分配的内存，对另一个 Context 不可见。这提供了进程级的内存隔离。

// 进程 A 的 Context
aclrtContext ctx_a;
aclrtCreateContext(&ctx_a, device_id);
aclrtSetCurrentContext(ctx_a);

// 进程 A 分配 2GB 显存
void* ptr_a;
aclrtMalloc(&ptr_a, 2ULL * 1024 * 1024 * 1024, ACL_MEM_MALLOC_HBM_FIRST);

// 切换到进程 B 的 Context
aclrtContext ctx_b;
aclrtCreateContext(&ctx_b, device_id);
aclrtSetCurrentContext(ctx_b);

// 进程 B 看不到 ptr_a，只能看到自己 Context 下分配的内存
// 进程 B 分配自己的显存
void* ptr_b;
aclrtMalloc(&ptr_b, 1ULL * 1024 * 1024 * 1024, ACL_MEM_MALLOC_HBM_FIRST);

3.2 显存配额限制

在生产环境中，仅靠 Context 隔离是不够的。还需要对每个租户设置显存使用上限：

// 伪代码：显存配额管理
typedef struct {
    uint32_t tenant_id;
    size_t quota_bytes;      // 配额上限
    size_t used_bytes;       // 已使用量
    aclrtContext context;    // 对应的 Context
} TenantQuota;

// 检查是否超过配额
aclError check_quota(TenantQuota* quota, size_t request_size) {
    if (quota->used_bytes + request_size > quota->quota_bytes) {
        printf("[拒绝] 租户 %u 显存不足：请求 %zu MB，已用 %zu MB，配额 %zu MB\n",
               quota->tenant_id,
               request_size / (1024*1024),
               quota->used_bytes / (1024*1024),
               quota->quota_bytes / (1024*1024));
        return ACL_ERROR_FAILURE;
    }
    return ACL_SUCCESS;
}

// 带配额检查的内存分配
aclError tenant_malloc(TenantQuota* quota, void** ptr, size_t size) {
    aclError ret = check_quota(quota, size);
    if (ret != ACL_SUCCESS) return ret;
    
    aclrtSetCurrentContext(quota->context);
    ret = aclrtMalloc(ptr, size, ACL_MEM_MALLOC_HBM_FIRST);
    if (ret == ACL_SUCCESS) {
        quota->used_bytes += size;
    }
    return ret;
}

3.3 显存水位监控

实时监控每个租户的显存使用情况，防止 OOM（Out of Memory）导致整个设备崩溃：

// 查询当前 Device 的显存使用情况
aclrtMemInfo mem_info;
aclError ret = aclrtGetMemInfo(&mem_info);
if (ret == ACL_SUCCESS) {
    printf("[显存监控] 总量: %zu MB, 已用: %zu MB, 可用: %zu MB\n",
           mem_info.total / (1024*1024),
           mem_info.used / (1024*1024),
           mem_info.free / (1024*1024));
    
    // 显存使用率超过 85% 时触发告警
    double usage_rate = (double)mem_info.used / mem_info.total;
    if (usage_rate > 0.85) {
        printf("[告警] 显存使用率 %.1f%%，建议释放非必要缓存\n", usage_rate * 100);
    }
}

---

四、多租户配额管理

配额管理是多租户平台的核心能力。需要从多个维度为每个租户设置资源使用上限。

4.1 配额维度设计

一个完善的配额体系应包含以下维度：

维度	说明	示例值
NPU 数量	可同时使用的 NPU 卡数	最多 4 张
显存上限	单卡或总显存使用上限	单卡 32GB
计算时间	单次任务的最大执行时间	最长 24 小时
并发任务数	同时在执行的任务数量	最多 8 个
优先级范围	允许使用的优先级等级	高/中/低
网络带宽	跨机通信的带宽限制	10 Gbps

4.2 配额管理器的实现

import json
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional

@dataclass
class TenantQuota:
    """租户配额定义"""
    tenant_id: str
    max_npus: int = 4                    # 最多使用的 NPU 数量
    max_memory_per_device: int = 32      # 单卡显存上限 (GB)
    max_task_duration: int = 86400       # 单任务最大执行时间 (秒)
    max_concurrent_tasks: int = 8        # 最大并发任务数
    allowed_priorities: List[str] = field(default_factory=lambda: ['high', 'normal', 'low'])
    max_bandwidth: int = 10              # 网络带宽限制 (Gbps)
    
    # 运行时状态
    used_npus: int = 0
    active_tasks: int = 0
    task_history: List[dict] = field(default_factory=list)

@dataclass
class Task:
    """提交的任务"""
    task_id: str
    tenant_id: str
    priority: str
    required_npus: int
    required_memory: int  # GB
    estimated_duration: int  # 秒

class QuotaManager:
    """多租户配额管理器"""
    
    def __init__(self, total_npus: int = 8):
        self.total_npus = total_npus
        self.quotas: Dict[str, TenantQuota] = {}
        self.device_allocation: Dict[int, str] = {}  # device_id -> tenant_id
    
    def register_tenant(self, tenant_id: str, quota: TenantQuota):
        """注册租户配额"""
        self.quotas[tenant_id] = quota
        print(f"[配额管理] 租户 {tenant_id} 注册成功，配额: {quota.max_npus} NPUs, "
              f"{quota.max_memory_per_device}GB/卡")
    
    def check_and_allocate(self, task: Task) -> Optional[int]:
        """检查配额并分配资源"""
        quota = self.quotas.get(task.tenant_id)
        if not quota:
            print(f"[拒绝] 租户 {task.tenant_id} 未注册")
            return None
        
        # 检查优先级权限
        if task.priority not in quota.allowed_priorities:
            print(f"[拒绝] 租户 {task.tenant_id} 无权使用 {task.priority} 优先级")
            return None
        
        # 检查并发任务数
        if quota.active_tasks >= quota.max_concurrent_tasks:
            print(f"[拒绝] 租户 {task.tenant_id} 已达最大并发数 {quota.max_concurrent_tasks}")
            return None
        
        # 检查 NPU 数量配额
        if quota.used_npus + task.required_npus > quota.max_npus:
            print(f"[拒绝] 租户 {task.tenant_id} NPU 配额不足: "
                  f"需 {task.required_npus}，已用 {quota.used_npus}，配额 {quota.max_npus}")
            return None
        
        # 找到可用的 NPU
        available_devices = []
        for i in range(self.total_npus):
            if i not in self.device_allocation:
                available_devices.append(i)
        
        if len(available_devices) < task.required_npus:
            print(f"[等待] 全局 NPU 资源不足，需 {task.required_npus} 张，"
                  f"可用 {len(available_devices)} 张")
            return None
        
        # 分配 NPU
        allocated = available_devices[:task.required_npus]
        for dev_id in allocated:
            self.device_allocation[dev_id] = task.tenant_id
        
        quota.used_npus += task.required_npus
        quota.active_tasks += 1
        quota.task_history.append({
            'task_id': task.task_id,
            'start_time': time.time(),
            'devices': allocated
        })
        
        print(f"[分配] 任务 {task.task_id} 获得 NPU {allocated}，"
              f"租户 {task.tenant_id} 当前使用 {quota.used_npus}/{quota.max_npus}")
        
        return allocated[0]  # 返回主设备 ID
    
    def release_resources(self, task: Task):
        """释放任务占用的资源"""
        quota = self.quotas.get(task.tenant_id)
        if not quota:
            return
        
        # 查找该任务占用的设备
        devices_to_release = []
        for dev_id, tenant in list(self.device_allocation.items()):
            if tenant == task.tenant_id:
                devices_to_release.append(dev_id)
        
        for dev_id in devices_to_release[:task.required_npus]:
            del self.device_allocation[dev_id]
        
        quota.used_npus -= task.required_npus
        quota.active_tasks -= 1
        
        print(f"[释放] 任务 {task.task_id} 释放 NPU {devices_to_release[:task.required_npus]}，"
              f"租户 {task.tenant_id} 当前使用 {quota.used_npus}/{quota.max_npus}")

# 使用示例
manager = QuotaManager(total_npus=8)

# 注册两个租户
manager.register_tenant('team-a', TenantQuota(
    tenant_id='team-a',
    max_npus=4,
    max_memory_per_device=32,
    max_concurrent_tasks=4,
    allowed_priorities=['high', 'normal', 'low']
))

manager.register_tenant('team-b', TenantQuota(
    tenant_id='team-b',
    max_npus=2,
    max_memory_per_device=16,
    max_concurrent_tasks=2,
    allowed_priorities=['normal', 'low']  # team-b 不允许使用高优先级
))

# 提交任务
task1 = Task(task_id='train-001', tenant_id='team-a', priority='high',
             required_npus=2, required_memory=16, estimated_duration=3600)
task2 = Task(task_id='infer-001', tenant_id='team-b', priority='normal',
             required_npus=1, required_memory=8, estimated_duration=600)

manager.check_and_allocate(task1)
manager.check_and_allocate(task2)

4.3 配额动态调整

在生产环境中，配额不是一成不变的。需要根据时段、负载等因素动态调整：

class DynamicQuotaAdjuster:
    """基于时段的配额动态调整"""
    
    def __init__(self, quota_manager: QuotaManager):
        self.manager = quota_manager
        self.schedules = {}
    
    def add_schedule(self, tenant_id: str, time_range: str, 
                     override_config: dict):
        """添加时段配额规则
        time_range: '09:00-18:00' (工作时间)
        override_config: 覆盖的配额参数
        """
        self.schedules.setdefault(tenant_id, []).append({
            'time_range': time_range,
            'config': override_config
        })
    
    def apply_schedule(self):
        """根据当前时间应用配额调整"""
        from datetime import datetime
        now = datetime.now()
        current_time = now.strftime('%H:%M')
        
        for tenant_id, rules in self.schedules.items():
            for rule in rules:
                start, end = rule['time_range'].split('-')
                if start <= current_time <= end:
                    quota = self.manager.quotas.get(tenant_id)
                    if quota:
                        for key, value in rule['config'].items():
                            if hasattr(quota, key):
                                old_val = getattr(quota, key)
                                setattr(quota, key, value)
                                print(f"[动态配额] 租户 {tenant_id}: "
                                      f"{key} {old_val} → {value}")

# 示例：工作时间给在线推理团队更多 NPU
adjuster = DynamicQuotaAdjuster(manager)
adjuster.add_schedule('team-a', '09:00-18:00', {'max_npus': 6})
adjuster.add_schedule('team-a', '22:00-06:00', {'max_npus': 2})

---

五、设备热插拔与故障恢复

NPU 硬件故障在大规模集群中不可避免。CANN 提供了设备热插拔和故障恢复机制，确保单卡故障不会导致整个训练任务失败。

5.1 设备状态监控

// 检查设备是否正常
aclrtDeviceInfo device_info;
aclError ret = aclrtGetDeviceInfo(device_id, &device_info);
if (ret != ACL_SUCCESS) {
    printf("[故障] 设备 %d 状态异常，错误码: %d\n", device_id, ret);
    // 触发故障转移
    trigger_failover(device_id);
} else {
    printf("[正常] 设备 %d: %s, AI Core 数: %d, 显存: %zu MB\n",
           device_id,
           device_info.name,
           device_info.aiCoreNum,
           device_info.memSize / (1024*1024));
}

5.2 故障转移流程

当检测到设备故障时，调度器需要执行以下步骤：

1. 暂停该设备上的所有任务：调用 aclrtSynchronizeDevice 确保所有正在执行的任务完成或终止
2. 将任务迁移到健康设备：重新分配资源，将未完成的任务提交到其他设备
3. 释放故障设备的资源：清理 Context、Stream、内存等
4. 记录故障信息：用于后续分析和告警

// 故障转移的伪代码流程
void handle_device_failure(int failed_device) {
    printf("[故障处理] 设备 %d 发生故障，开始转移\n", failed_device);
    
    // 1. 同步设备，确保所有任务完成
    aclrtSynchronizeDevice(failed_device);
    
    // 2. 获取该设备上的所有 Context
    // （实际实现需要维护 Context 注册表）
    List<aclrtContext> contexts = get_contexts_on_device(failed_device);
    
    for (aclrtContext ctx : contexts) {
        // 3. 将 Context 迁移到健康设备
        int healthy_device = find_healthy_device();
        if (healthy_device >= 0) {
            aclrtContext new_ctx;
            aclrtCreateContext(&new_ctx, healthy_device);
            
            // 重新提交未完成的任务到新 Context
            resubmit_pending_tasks(ctx, new_ctx);
            
            // 清理旧 Context
            aclrtDestroyContext(ctx);
        }
    }
    
    // 4. 从调度器中移除故障设备
    remove_device_from_scheduler(failed_device);
    
    // 5. 记录故障日志
    log_failure(failed_device, time(NULL));
    
    // 6. 如果配置了告警，发送通知
    send_alert("设备故障", failed_device);
}

5.3 自动恢复机制

设备故障恢复后，需要自动将其重新加入集群：

class DeviceHealthChecker:
    """设备健康检查与自动恢复"""
    
    def __init__(self, check_interval: int = 30):
        self.check_interval = check_interval
        self.device_status = {}  # device_id -> status
        self.recovery_callbacks = []
    
    def check_device(self, device_id: int) -> bool:
        """检查单个设备是否健康"""
        try:
            # 尝试在设备上执行一个简单的计算
            import acl
            acl.aclInit(None)
            context, _ = acl.aclrtCreateContext(device_id)
            acl.aclrtSetCurrentContext(context)
            
            # 分配一小块内存并写入数据
            test_size = 1024
            test_ptr = acl.aclrtMalloc(test_size, acl.ACL_MEM_MALLOC_HBM_FIRST)
            if test_ptr:
                acl.aclrtFree(test_ptr)
                acl.aclrtDestroyContext(context)
                return True
            else:
                acl.aclrtDestroyContext(context)
                return False
        except Exception as e:
            print(f"[健康检查] 设备 {device_id} 异常: {e}")
            return False
    
    def run_recovery_loop(self):
        """持续监控并自动恢复"""
        while True:
            for device_id in range(8):  # 假设 8 张卡
                is_healthy = self.check_device(device_id)
                was_healthy = self.device_status.get(device_id, True)
                
                if is_healthy and not was_healthy:
                    # 设备恢复健康，自动加入
                    print(f"[恢复] 设备 {device_id} 恢复健康，重新加入集群")
                    self.device_status[device_id] = True
                    for callback in self.recovery_callbacks:
                        callback(device_id, 'recovered')
                
                elif not is_healthy and was_healthy:
                    # 设备故障，触发转移
                    print(f"[故障] 设备 {device_id} 检测到故障")
                    self.device_status[device_id] = False
                    for callback in self.recovery_callbacks:
                        callback(device_id, 'failed')
            
            time.sleep(self.check_interval)

---

六、常见问题与最佳实践

Q1：如何避免训练任务饿死在线推理任务？

核心策略：使用优先级队列 + 资源预留。为在线推理任务预留一定数量的 NPU（如总数量的 30%），确保即使训练任务占满了其余 NPU，在线推理仍有可用资源。

// 预留配置
typedef struct {
    int total_npus;           // 总 NPU 数量
    int reserved_for_infer;   // 为推理预留的数量
    int available_for_train;  // 训练可用的数量
} NpuReservation;

NpuReservation reservation = {
    .total_npus = 8,
    .reserved_for_infer = 3,   // 预留 3 张给推理
    .available_for_train = 5   // 训练最多用 5 张
};

Q2：多租户间如何实现显存完全隔离？

推荐方案：使用 Docker 容器 + CANN 容器化部署。每个租户运行在独立容器中，通过 cgroup 限制显存使用上限。CANN 的容器化方案（Ascend Container）支持将 NPU 设备映射到容器内，同时通过 cgroup v2 限制内存使用。

# 使用 Docker 启动一个限制显存的容器
docker run -it \
    --device /dev/davinci0 \
    --device /dev/davinci1 \
    --memory=32g \
    --memory-swap=32g \
    -v /usr/local/Ascend:/usr/local/Ascend \
    -v /usr/local/bin/npu-smi:/usr/local/bin/npu-smi \
    ascend-containers/cann:latest \
    bash

# 在容器内验证 NPU 可用
npu-smi info

Q3：如何监控多租户的资源使用情况？

推荐工具链：

• npu-smi：实时查看 NPU 利用率、显存使用、温度等
• CANN Profiling：分析任务的执行时间、Stream 利用率
• 自定义监控脚本：周期性采集指标，推送到 Prometheus/Grafana

# 查看所有 NPU 的实时状态
npu-smi info

# 查看指定 NPU 的详细信息
npu-smi info -t board -i 0

# 查看 NPU 上的进程信息
npu-smi info -t proc

Q4：任务调度延迟过高怎么优化？

优化方向：

1. 减少 Context 切换开销：尽量让同一租户的任务复用 Context，避免频繁创建/销毁
2. 批量提交任务：将多个小任务打包成一个 Batch 提交，减少调度器的决策次数
3. 调整 Stream 数量：每个租户分配固定数量的 Stream，避免 Stream 竞争
4. 使用算子融合：通过 CANN 的 AutoFusion 能力，将多个小算子融合成一个大算子，减少调度开销

// 批量提交任务的示例
// 将多个矩阵乘法打包成一个 Batch
int batch_size = 16;
aclrtStream stream;
aclrtCreateStream(&stream);

for (int i = 0; i < batch_size; i++) {
    // 提交第 i 个矩阵乘法到同一个 Stream
    aclblasGemmEx(/* params for batch i */, stream);
}

// 一次同步，等待整个 Batch 完成
aclrtSynchronizeStream(stream);

---

七、总结

多租户 NPU 资源管理是一个系统工程，涉及硬件理解、调度算法、内存管理、故障处理等多个层面。关键要点：

1. 理解硬件：NPU 的 Device/Context/Stream 抽象是调度的基础
2. 优先级调度：通过多级队列确保关键任务的低延迟
3. 显存隔离：Context 级隔离 + 配额限制，防止租户间干扰
4. 动态配额：根据时段和负载灵活调整资源分配
5. 故障恢复：自动检测 + 任务迁移 + 设备恢复，保障高可用