前言

昇腾 NPU 的驱动层是硬件和软件之间的桥梁。你写的每一行 PyTorch 代码,最终都要通过驱动翻译成硬件指令。这篇文章从架构层面讲清楚 CANN driver 的设计、核心组件和故障排查方法。

为什么需要驱动层

传统 GPU vs 昇腾 NPU

层次 传统 GPU 昇腾 NPU
框架层 CUDA/PyTorch PyTorch NPU / CANN
运行时层 CUDA Runtime ACL Runtime
驱动层 nvidia.ko cann.ko
硬件层 NVIDIA GPU Ascend NPU

驱动的核心职责:

  1. 内存管理:GPU/NPU 显存分配和释放
  2. 命令调度:把计算任务发送到硬件队列
  3. 中断处理:硬件事件的通知和响应
  4. 错误处理:硬件异常的上报和处理

昇腾驱动的特殊性

昇腾 NPU 的驱动比 NVIDIA 驱动更复杂,原因在于:

  • 异构架构:Ascend 芯片有 Scalar、Cube、Vector 三种计算单元,需要统一的调度
  • 虚拟化支持:支持多租户、多容器隔离
  • 固件分离:部分逻辑下沉到固件层,驱动和固件协同工作

驱动架构全览

CANN 驱动的分层架构

┌─────────────────────────────────────┐
│         User Space (应用层)          │
│   PyTorch / TensorFlow / MindSpore   │
└─────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────┐
│         ACL Runtime (运行时层)        │
│   图编译 / 算子调度 / 内存管理         │
└─────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────┐
│         HCCL (集合通信层)             │
│   AllReduce / AllGather / Broadcast  │
└─────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────┐
│         CANN Driver (驱动层)          │
│   硬件抽象 / 命令调度 / 内存映射        │
└─────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────┐
│         Firmware (固件层)             │
│   微内核 / 任务调度 / 硬件寄存器        │
└─────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────┐
│         Hardware (硬件层)             │
│   Scalar / Cube / Vector 单元        │
└─────────────────────────────────────┘

驱动的核心组件

# 驱动目录结构
ls -la /usr/local/Ascend/driver/

# driver/        - 主驱动模块
# firmware/      - 固件文件
# tools/         - 诊断工具
# version.info   - 版本信息

# 查看驱动版本
cat /usr/local/Ascend/driver/version.info

# 输出示例:
# Driver Version: 23.0.rc3
# Firmware Version: 1.85.12.b120
# Build Date: 2024-03-15

驱动核心 API

设备管理

# device_mgmt.py
import cann
from cann.driver import Driver

# 获取驱动句柄
driver = Driver.get_instance()
print(f"驱动版本: {driver.version}")
print(f"设备数量: {driver.device_count}")

# 遍历所有设备
for device_id in range(driver.device_count):
    device = driver.get_device(device_id)
    print(f"设备 {device_id}: {device.name}")
    print(f"  显存总量: {device.mem_total / 1024**3:.1f} GB")
    print(f"  显存剩余: {device.mem_free / 1024**3:.1f} GB")
    print(f"  计算单元: {device.compute_units}")

# 输出示例:
# 驱动版本: 23.0.rc3
# 设备数量: 8
# 设备 0: Ascend 910P8
#   显存总量: 32.0 GB
#   显存剩余: 28.5 GB
#   计算单元: 32

内存管理

# mem_mgmt.py
import cann
import numpy as np

# 分配设备内存
mem_size = 1024 * 1024 * 1024  # 1GB
device_mem = cann.driver.malloc(mem_size)
print(f"分配设备内存: {mem_size / 1024**2:.0f} MB")

# 从 CPU 拷贝数据到设备
host_data = np.random.randn(1024).astype(np.float32)
cann.driver.memcpy_host_to_device(device_mem, host_data)

# 从设备拷贝数据到 CPU
output = np.zeros_like(host_data)
cann.driver.memcpy_device_to_host(device_mem, output)

# 释放内存
cann.driver.free(device_mem)

# 内存池管理
mem_pool = cann.driver.MemoryPool(
    block_size=32 * 1024 * 1024,  # 32MB per block
    max_blocks=256
)
mem = mem_pool.allocate(64 * 1024 * 1024)  # 分配 64MB
mem_pool.deallocate(mem)

命令队列

# cmd_queue.py
import cann
from cann.driver import Stream, Event

# 创建命令流(Stream)
stream = cann.driver.Stream(device_id=0)
print(f"创建命令流: {stream.handle}")

# 在 Stream 上提交任务
stream.memcpy_host_to_device(dest, src)
stream.kernel_launch(kernel, grid, block, args)
stream.memcpy_device_to_host(dest, src)

# 同步(等待 Stream 完成)
stream.synchronize()

# 事件(用于跨 Stream 同步)
start_event = stream.record()
# ... 执行一些操作 ...
end_event = stream.record()

# 等待事件完成
end_event.synchronize()

# 计算两个事件之间的时间
elapsed_ms = start_event.elapsed_time(end_event)
print(f"耗时: {elapsed_ms:.2f} ms")

错误处理

# error_handling.py
import cann
from cann.driver import DriverError

try:
    # 尝试执行可能失败的操作
    device = cann.driver.get_device(99)  # 不存在的设备
except DriverError as e:
    print(f"驱动错误: {e.error_code}")
    print(f"错误信息: {e.message}")
    print(f"错误位置: {e.function}:{e.line}")

    # 错误码说明
    error_map = {
        0x0001: "设备不存在",
        0x0002: "内存不足",
        0x0003: "内核启动失败",
        0x0004: "超时",
        0x0005: "非法参数",
    }
    print(f"错误类型: {error_map.get(e.error_code, '未知错误')}")

# 获取最近一次错误
last_error = cann.driver.get_last_error()
if last_error:
    print(f"最近的错误: {last_error}")

驱动的关键机制

1. 内存映射(MMU)

昇腾 NPU 使用统一的虚拟地址空间,驱动负责虚拟地址到物理地址的映射:

# mmu_demo.py
import cann

# 查询虚拟地址的物理地址
va = cann.driver.malloc(1024 * 1024)
pa = cann.driver.va_to_pa(va)
print(f"虚拟地址: 0x{va:x}")
print(f"物理地址: 0x{pa:x}")

# 内存属性查询
attr = cann.driver.query_mem_attr(va)
print(f"内存类型: {attr.type}")     # e.g., "device", "host", "p2p"
print(f"可缓存: {attr.cacheable}")
print(f"对齐: {attr.alignment} bytes")

# 页表操作(高级用法)
# 大页(HugePage)分配,减少 TLB miss
large_mem = cann.driver.malloc_huge(
    size=256 * 1024 * 1024,
    page_size=2 * 1024 * 1024  # 2MB 大页
)
print(f"大页内存: {large_mem.size / 1024**2:.0f} MB")

2. 中断处理

# interrupt_handling.py
import cann
from cann.driver import IrqHandler

class MyIrqHandler(IrqHandler):
    """自定义中断处理"""
    def handle(self, irq_id, data):
        print(f"中断触发: IRQ {irq_id}")

        # 读取中断状态
        status = cann.driver.read_reg(0x1000)
        print(f"中断状态: 0x{status:x}")

        # 处理完成,清除中断
        cann.driver.write_reg(0x1000, 0)
        return True

# 注册中断处理
handler = MyIrqHandler()
irq_id = cann.driver.irq_register(irq=16, handler=handler)
print(f"注册中断 {irq_id}")

# 启用中断
cann.driver.irq_enable(irq_id)

# 禁用中断
cann.driver.irq_disable(irq_id)

# 注销中断
cann.driver.irq_unregister(irq_id)

3. 固件接口

驱动和固件通过 MailBox 机制通信:

# firmware_interface.py
import cann

# 获取固件信息
fw_info = cann.driver.get_firmware_info()
print(f"固件版本: {fw_info.version}")
print(f"固件 Build ID: {fw_info.build_id}")
print(f"支持的微架构: {fw_info.arch}")

# 固件日志读取
fw_log = cann.driver.read_firmware_log(max_lines=100)
for line in fw_log:
    print(line)

# 固件调试接口
# 读取固件内存(仅调试用)
firmware_mem = cann.driver.read_firmware_mem(offset=0x10000, size=256)
print(f"固件内存内容: {firmware_mem.hex()}")

# 发送固件命令
response = cann.driver.send_firmware_command(
    cmd_id=0x1234,
    param=b"\x01\x02\x03\x04",
    timeout_ms=1000
)
print(f"固件响应: {response.hex()}")

性能 profiling

驱动级 profiling

# driver_profiling.py
import cann
from cann.driver import Profiler

# 创建 profiler
profiler = Profiler(device_id=0)

# 开启 profiling
profiler.start(
    collect_kernel=True,
    collect_memcpy=True,
    collect_irq=True,
    sample_interval_us=100
)

# 执行你的代码
model = cann.model.load("resnet50.om")
for i in range(100):
    input_tensor = cann.Tensor.from_numpy(np.random.randn(1, 3, 224, 224).astype(np.float32))
    _ = model.execute(input_tensor)

# 停止 profiling
profiler.stop()

# 获取报告
report = profiler.get_report()

# 1. Kernel 级别
print("=== Kernel 统计 ===")
for item in report.kernels[:10]:
    print(f"{item.name:30s} | 调用: {item.calls:4d} | "
          f"平均: {item.avg_us:7.1f}us | 总计: {item.total_ms:6.2f}ms")

# 输出示例:
# === Kernel 统计 ===
# conv2d_3x3                   | 调用: 5000 | 平均:   123.5us | 总计: 617.5ms
# batch_norm                    | 调用: 5000 | 平均:    45.2us | 总计: 226.0ms
# relu                          | 调用: 5000 | 平均:    12.8us | 总计:  64.0ms
# matmul                        | 调用: 1000 | 平均:   289.3us | 总计: 289.3ms

# 2. 内存拷贝统计
print("\n=== 内存拷贝统计 ===")
for item in report.memcpy:
    print(f"{item.direction:10s} | 大小: {item.size/1024:6.1f}KB | "
          f"平均: {item.avg_us:6.1f}us | 次数: {item.count}")

# 3. 中断统计
print("\n=== 中断统计 ===")
for item in report.irq:
    print(f"IRQ {item.irq_id:2d} | 次数: {item.count:5d} | 总耗时: {item.total_us:8.1f}us")

驱动状态监控

# driver_monitor.py
import cann
import time

def monitor_driver():
    """实时监控驱动状态"""
    driver = cann.driver.Driver.get_instance()

    print("设备ID | 显存使用 | 活跃Stream | 队列深度")
    print("-" * 50)

    for _ in range(10):
        for device_id in range(driver.device_count):
            device = driver.get_device(device_id)

            # 显存
            mem_used = device.mem_used / 1024**3
            mem_total = device.mem_total / 1024**3
            mem_pct = mem_used / mem_total * 100

            # 活跃 Stream
            active_streams = device.get_active_stream_count()

            # 队列深度
            queue_depth = device.get_queue_depth()

            print(f"  {device_id}     | {mem_pct:5.1f}% ({mem_used:.1f}G/{mem_total:.1f}G) | "
                  f"    {active_streams:3d}     |   {queue_depth:3d}")

        time.sleep(1)

# 运行监控
monitor_driver()

故障排查指南

常见驱动错误

错误码 含义 排查方法
1001 设备初始化失败 检查驱动是否正确安装
1002 内存分配失败 减少 batch size 或清理显存
1003 内核启动失败 检查 kernel 参数是否正确
1004 固件超时 重启驱动服务
1005 权限不足 使用 sudo 或检查 cgroup 配置

驱动日志

# 查看驱动日志
# 方式1:dmesg(内核日志)
dmesg | grep -i ascend

# 方式2:ACL 日志
cat /var/log/ascend/acl_log/*
cat /var/log/ascend/driver.log

# 方式3:syslog
journalctl -u ascend-drivers -n 50

# 开启驱动 debug 日志
echo "module cann +p" > /sys/kernel/debug/dynamic_debug/control
echo 1 > /sys/module/ascend/parameters/driver_debug
dmesg -w | grep ascend

驱动重置

# driver_reset.py
import cann
from cann.driver import Driver

# 方式1:软复位(只复位单个设备)
device = Driver.get_instance().get_device(0)
device.reset()

# 方式2:硬复位(复位整个芯片)
Driver.get_instance().reset_device(0, mode="hard")

# 方式3:重启驱动服务
# 需要 sudo 权限
import subprocess
subprocess.run(["systemctl", "restart", "ascend-driver"])

# 方式4:ECCE(错误纠正与恢复)
# 自动从 ECC 错误中恢复
eCCE = cann.driver.ECCErrorCorrector()
eCCE.enable_auto_recovery()

驱动升级

# 备份当前驱动
tar -czvf driver_backup.tar.gz /usr/local/Ascend/driver/

# 升级驱动
bash Ascend-driver-{version}.run --full

# 验证安装
npu-smi info
cat /usr/local/Ascend/driver/version.info

# 重启应用
systemctl restart ascend-manager

总结

CANN 驱动的核心要点:

  1. 分层架构:驱动处于硬件抽象层,承上启下
  2. 内存管理:MMU 负责虚拟地址到物理地址的映射
  3. 命令调度:Stream 和 Event 管理计算任务流
  4. 固件协同:驱动和固件通过 MailBox 通信
  5. Profiling:驱动层提供细粒度的性能数据

理解驱动层,遇到"内核启动失败"、"内存分配错误"这类问题就知道往哪里排查了。

仓库地址:https://atomgit.com/cann/driver

Logo

鲲鹏昇腾开发者社区是面向全社会开放的“联接全球计算开发者,聚合华为+生态”的社区,内容涵盖鲲鹏、昇腾资源,帮助开发者快速获取所需的知识、经验、软件、工具、算力,支撑开发者易学、好用、成功,成为核心开发者。

更多推荐