muPTI 开发者指南

概述

MUSA Performance Tools Interface (muPTI) 是 MUSA 平台提供的 GPU 性能分析工具接口，用于收集 GPU 硬件性能计数器数据、跟踪 kernel 执行和内存操作。

主要组件

组件	说明
Activity API	记录 GPU 活动（kernel 执行、内存拷贝等）
Callback API	注册回调函数接收运行时事件通知

快速开始

Activity API 使用流程

注册回调: 设置缓冲区回调函数
启用 Activity: 启用需要记录的活动类型
执行: 运行 GPU kernel
收集: 在回调中处理 activity 记录
清理: 调用 muptiFinalize()

完整示例

#include <mupti.h>
#include <malloc.h>
#include <unistd.h>
#include <thread>
#include <atomic>
#include <iostream>

#define CHECK_MUSA_ERROR(call) do { \
    musaError_t err = call; \
    if (err != musaSuccess) { \
        fprintf(stderr, "MUSA error at %s:%d: %s\n", \
                __FILE__, __LINE__, musaGetErrorString(err)); \
        exit(1); \
    } \
} while(0)

#define CHECK_MUPTI_ERROR(call) do { \
    MUptiResult err = call; \
    if (err != MUPTI_SUCCESS) { \
        const char *errStr; \
        muptiGetResultString(err, &errStr); \
        fprintf(stderr, "MUPTI error at %s:%d: %s\n", \
                __FILE__, __LINE__, errStr); \
        exit(1); \
    } \
} while(0)

__global__ void axpy(float *x, float *y, float a) {
    y[threadIdx.x] = a * x[threadIdx.x];
}

// 打印 Activity 记录
void printActivity(MUpti_Activity *record) {
    switch (record->kind) {
    case MUPTI_ACTIVITY_KIND_MEMCPY: {
        MUpti_ActivityMemcpy4 *memcpy = (MUpti_ActivityMemcpy4 *)record;
        printf("MEMCPY size=%llu, device=%u, context=%u, stream=%u\n",
               (unsigned long long)memcpy->bytes, memcpy->deviceId, 
               memcpy->contextId, memcpy->streamId);
        break;
    }
    case MUPTI_ACTIVITY_KIND_MEMSET: {
        MUpti_ActivityMemset3 *memset = (MUpti_ActivityMemset3 *)record;
        printf("MEMSET value=%u, device=%u, context=%u, stream=%u\n",
               memset->value, memset->deviceId, memset->contextId, memset->streamId);
        break;
    }
    case MUPTI_ACTIVITY_KIND_KERNEL:
    case MUPTI_ACTIVITY_KIND_CONCURRENT_KERNEL: {
        MUpti_ActivityKernel6 *kernel = (MUpti_ActivityKernel6 *)record;
        printf("KERNEL \"%s\", grid=[%u,%u,%u], block=[%u,%u,%u]\n",
               kernel->name, kernel->gridX, kernel->gridY, kernel->gridZ,
               kernel->blockX, kernel->blockY, kernel->blockZ);
        break;
    }
    default:
        break;
    }
}

// 缓冲区请求回调
void bufferRequested(uint8_t **buffer, size_t *size, size_t *maxNumRecords) {
    *buffer = static_cast<uint8_t*>(aligned_alloc(8, 256 * 10));
    *size = malloc_usable_size(*buffer);
    *maxNumRecords = 0;
}

// 缓冲区完成回调
void bufferCompleted(MUcontext ctx, uint32_t streamId, uint8_t *buffer, 
                     size_t size, size_t validSize) {
    if (validSize > 0) {
        MUpti_Activity* record = nullptr;
        MUptiResult status;
        do {
            status = muptiActivityGetNextRecord(buffer, validSize, &record);
            if (status == MUPTI_SUCCESS) {
                printActivity(record);
            } else if (status == MUPTI_ERROR_MAX_LIMIT_REACHED) {
                break;
            } else {
                exit(1);
            }
        } while (true);
        
        size_t dropped;
        CHECK_MUPTI_ERROR(muptiActivityGetNumDroppedRecords(ctx, streamId, &dropped));
        if (dropped != 0) {
            printf("Dropped %u activity records\n", (unsigned int) dropped);
        }
    }
    free(buffer);
}

// 初始化 Trace
void initTrace() {
    CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_DRIVER));
    CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_RUNTIME));
    CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_MEMCPY));
    CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_MEMSET));
    CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_KERNEL));
    CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_CONCURRENT_KERNEL));
    CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_SYNCHRONIZATION));
    CHECK_MUPTI_ERROR(muptiActivityRegisterCallbacks(bufferRequested, bufferCompleted));
}

// 清理 Trace
void finiTrace() {
    CHECK_MUPTI_ERROR(muptiActivityFlushAll(1));
    CHECK_MUPTI_ERROR(muptiActivityDisable(MUPTI_ACTIVITY_KIND_DRIVER));
    CHECK_MUPTI_ERROR(muptiActivityDisable(MUPTI_ACTIVITY_KIND_RUNTIME));
    CHECK_MUPTI_ERROR(muptiActivityDisable(MUPTI_ACTIVITY_KIND_MEMCPY));
    CHECK_MUPTI_ERROR(muptiActivityDisable(MUPTI_ACTIVITY_KIND_MEMSET));
    CHECK_MUPTI_ERROR(muptiActivityDisable(MUPTI_ACTIVITY_KIND_KERNEL));
    CHECK_MUPTI_ERROR(muptiActivityDisable(MUPTI_ACTIVITY_KIND_CONCURRENT_KERNEL));
    CHECK_MUPTI_ERROR(muptiActivityDisable(MUPTI_ACTIVITY_KIND_SYNCHRONIZATION));
    CHECK_MUPTI_ERROR(muptiFinalize());
}

// 定期刷新线程
std::atomic<bool> flush_thread_active(true);

void periodic_flush() {
    while(flush_thread_active.load()) {
        muptiActivityFlushAll(0);
        std::this_thread::sleep_for(std::chrono::milliseconds(3));
    }
}

int main(int argc, char *argv[]) {
    // 编译命令: mcc axpy_trace.cu -mtgpu -O2 -lmusart -lmupti -L/usr/local/musa/lib -o axpy_trace
    
    initTrace();
    
    // 启动定期刷新线程
    std::thread flush_thread(periodic_flush);
    
    float a = 2.0f;
    const int kDataLen = 4;
    float host_x[kDataLen] = {1.0f, 2.0f, 3.0f, 4.0f};
    float host_y[kDataLen];
    
    float *device_x;
    float *device_y;
    CHECK_MUSA_ERROR(musaMalloc(&device_x, kDataLen * sizeof(float)));
    CHECK_MUSA_ERROR(musaMalloc(&device_y, kDataLen * sizeof(float)));
    CHECK_MUSA_ERROR(musaMemcpy(device_x, host_x, kDataLen * sizeof(float), musaMemcpyHostToDevice));
    
    musaStream_t streams[10];
    for (int i = 0; i < 10; i++) {
        CHECK_MUSA_ERROR(musaStreamCreate(&(streams[i])));
    }
    
    void* args[] = { &device_x, &device_y, &a };
    
    // Launch kernels
    for (int i = 0; i < 10; i++) {
        CHECK_MUSA_ERROR(musaLaunchKernel((void*)axpy, dim3(1), dim3(kDataLen), args, 0, streams[i]));
    }
    
    CHECK_MUSA_ERROR(musaDeviceSynchronize());
    CHECK_MUSA_ERROR(musaMemcpy(host_y, device_y, kDataLen * sizeof(float), musaMemcpyDeviceToHost));
    
    // 打印结果
    for (int i = 0; i < kDataLen; ++i) {
        std::cout << "y[" << i << "] = " << host_y[i] << "\n";
    }
    
    CHECK_MUSA_ERROR(musaFree(device_x));
    CHECK_MUSA_ERROR(musaFree(device_y));
    CHECK_MUSA_ERROR(musaDeviceReset());
    
    // 停止刷新线程
    flush_thread_active = false;
    flush_thread.join();
    
    finiTrace();
    
    return 0;
}

Activity API

启用 Activity 记录

CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_DRIVER));
CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_RUNTIME));
CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_MEMCPY));
CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_MEMSET));
CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_KERNEL));
CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_CONCURRENT_KERNEL));
CHECK_MUPTI_ERROR(muptiActivityEnable(MUPTI_ACTIVITY_KIND_SYNCHRONIZATION));

注册回调函数

CHECK_MUPTI_ERROR(muptiActivityRegisterCallbacks(bufferRequested, bufferCompleted));

刷新和获取 Activity 记录

muptiActivityFlushAll 参数说明

muptiActivityFlushAll(0)：在程序运行过程中调用，只刷新已完成的 activity 记录
muptiActivityFlushAll(1)：在程序结束后调用，刷新所有 activity 记录，不管是否完成

程序运行时调用 flush (0) 避免刷出未完成的事件，程序结束退出前调用一次 flush (1) 确保刷出所有事件

MUpti_Activity* record = nullptr;
MUptiResult status;

// 刷新 activity 缓冲区
CHECK_MUPTI_ERROR(muptiActivityFlushAll(1));

// 逐条读取 activity 记录
while ((status = muptiActivityGetNextRecord(buffer, validSize, &record)) == MUPTI_SUCCESS) {
    printActivity(record);
}

if (status != MUPTI_ERROR_MAX_LIMIT_REACHED) {
    fprintf(stderr, "Error reading activity records\n");
}

刷新缓冲区

// 场景 1：程序运行过程中定期刷新（只刷新已完成的）
CHECK_MUPTI_ERROR(muptiActivityFlushAll(0));  // 刷新已完成的 activity 记录

// 场景 2：程序结束后刷新（刷新所有）
CHECK_MUPTI_ERROR(muptiActivityFlushAll(1));  // 刷新所有 activity 记录，不管是否完成

Activity 类型

类型	常量
Kernel	`MUPTI_ACTIVITY_KIND_KERNEL`
Concurrent Kernel	`MUPTI_ACTIVITY_KIND_CONCURRENT_KERNEL`
Memcpy	`MUPTI_ACTIVITY_KIND_MEMCPY`
Memset	`MUPTI_ACTIVITY_KIND_MEMSET`
Driver API	`MUPTI_ACTIVITY_KIND_DRIVER`
Runtime API	`MUPTI_ACTIVITY_KIND_RUNTIME`
Synchronization	`MUPTI_ACTIVITY_KIND_SYNCHRONIZATION`

Activity Record 结构

Kernel Activity Record (MUpti_ActivityKernel6):

成员	类型	说明
`kind`	`MUpti_ActivityKind`	Activity 类型
`start`	`uint64_t`	开始时间（ns）
`end`	`uint64_t`	结束时间（ns）
`deviceId`	`uint32_t`	设备 ID
`contextId`	`uint32_t`	上下文 ID
`streamId`	`uint32_t`	流 ID
`gridX/Y/Z`	`uint32_t`	Grid 维度
`blockX/Y/Z`	`uint32_t`	Block 维度
`name`	`const char*`	Kernel 名称
`correlationId`	`uint32_t`	关联 ID
`graphId`	`uint32_t`	图 ID
`graphNodeId`	`uint64_t`	图节点 ID
`queued`	`uint64_t`	Kernel 排队时间（ns），需启用 `muptiActivityEnableLatencyTimestamps()`
`submitted`	`uint64_t`	命令缓冲区提交到 GPU 的时间（ns），需启用 latency timestamps
`staticSharedMemory`	`uint32_t`	静态共享内存
`dynamicSharedMemory`	`uint32_t`	动态共享内存

Memcpy Activity Record (MUpti_ActivityMemcpy4):

成员	类型	说明
`kind`	`MUpti_ActivityKind`	Activity 类型
`start`	`uint64_t`	开始时间（ns）
`end`	`uint64_t`	结束时间（ns）
`deviceId`	`uint32_t`	设备 ID
`contextId`	`uint32_t`	上下文 ID
`streamId`	`uint32_t`	流 ID
`bytes`	`uint64_t`	传输字节数
`copyKind`	`MUpti_ActivityMemcpyKind`	拷贝类型
`srcKind`	`uint8_t`	源类型
`dstKind`	`uint8_t`	目标类型
`correlationId`	`uint32_t`	关联 ID

Memset Activity Record (MUpti_ActivityMemset3):

成员	类型	说明
`kind`	`MUpti_ActivityKind`	Activity 类型
`start`	`uint64_t`	开始时间（ns）
`end`	`uint64_t`	结束时间（ns）
`deviceId`	`uint32_t`	设备 ID
`contextId`	`uint32_t`	上下文 ID
`streamId`	`uint32_t`	流 ID
`value`	`uint32_t`	设置值
`correlationId`	`uint32_t`	关联 ID

最佳实践

性能开销最小化

仅启用需要的功能: 不要同时启用所有 Activity 类型
及时刷新缓冲区: 避免缓冲区溢出导致数据丢失
使用单独线程刷新: 避免阻塞主线程

数据准确性

同步后读取数据: 确保所有 GPU 操作完成
多次测量取平均: 减少偶然误差
预热 GPU: 首次 kernel 执行可能有额外开销

编译和运行

# 编译
mcc axpy_trace.cu -mtgpu -O2 -lmusart -lmupti -L/usr/local/musa/lib -o axpy_trace

# 运行
./axpy_trace

Callback API - 回调函数

Callback API 允许注册回调函数，在特定运行时事件发生时接收通知。

注册回调函数

#include <musa_runtime.h>
#include <mupti.h>
#include <malloc.h>

#define CHECK_MUSA_ERROR(call) do { \
    musaError_t err = call; \
    if (err != musaSuccess) { \
        fprintf(stderr, "MUSA error at %s:%d: %s\n", \
                __FILE__, __LINE__, musaGetErrorString(err)); \
        exit(1); \
    } \
} while(0)

#define CHECK_MUPTI_ERROR(call) do { \
    MUptiResult err = call; \
    if (err != MUPTI_SUCCESS) { \
        const char *errStr; \
        muptiGetResultString(err, &errStr); \
        fprintf(stderr, "MUPTI error at %s:%d: %s\n", \
                __FILE__, __LINE__, errStr); \
        exit(1); \
    } \
} while(0)

// 回调处理函数
void callbackHandler(void* userdata, MUpti_CallbackDomain domain,
    MUpti_CallbackId cbid, const void* cbdata) {
    if (domain == MUPTI_CB_DOMAIN_RUNTIME_API || domain == MUPTI_CB_DOMAIN_DRIVER_API) {
        const MUpti_CallbackData* data = (const MUpti_CallbackData*)cbdata;
        if (data->callbackSite == MUPTI_API_ENTER) {
            printf(">>> CALLBACK ENTER: %s\n", data->functionName);
        } else if (data->callbackSite == MUPTI_API_EXIT) {
            printf(">>> CALLBACK EXIT: %s\n", data->functionName);
        }
    }
}

int main(int argc, char *argv[]) {
    // 创建订阅者
    MUpti_SubscriberHandle subscriber;
    CHECK_MUPTI_ERROR(muptiSubscribe(&subscriber, (MUpti_CallbackFunc)callbackHandler, nullptr));
    
    // 启用驱动 API 回调
    CHECK_MUPTI_ERROR(muptiEnableCallback(1, subscriber, MUPTI_CB_DOMAIN_DRIVER_API, 
        MUPTI_DRIVER_TRACE_CBID_muLaunchKernel));
    
    // 启用运行时 API 回调域
    CHECK_MUPTI_ERROR(muptiEnableDomain(1, subscriber, MUPTI_CB_DOMAIN_RUNTIME_API));
    
    // ... 运行 kernel ...
    
    CHECK_MUPTI_ERROR(muptiFinalize());
    return 0;
}

支持的回调类型

回调域	常量	说明
Driver API	`MUPTI_CB_DOMAIN_DRIVER_API`	驱动层 API 回调
Runtime API	`MUPTI_CB_DOMAIN_RUNTIME_API`	运行时 API 回调

常用回调 ID：

类型	常量	说明
Kernel 启动	`MUPTI_DRIVER_TRACE_CBID_muLaunchKernel`	Kernel 启动时触发
内存分配	`MUPTI_RUNTIME_CBID_musaMalloc`	显存分配时触发
内存释放	`MUPTI_RUNTIME_CBID_musaFree`	显存释放时触发

故障排除

常见问题

问题	可能原因	解决方案
Activity 记录为空	未启用 Activity 或未执行 GPU 操作	调用 `muptiActivityEnable` 并确保执行了 kernel
回调未触发	回调未正确注册	检查 `muptiSubscribe` 和 `muptiEnableCallback` 返回值和回调类型

调试技巧

检查返回值: 始终检查 API 返回值
使用最小测试用例: 用简单 kernel 隔离问题

API 参考

完整 API 参考请参阅 muPTI API Reference。

核心 API 摘要

类别	函数	说明
Activity	`muptiActivityEnable()` / `muptiActivityFlushAll()` / `muptiActivityGetNextRecord()`	Activity 记录管理
Callback	`muptiSubscribe()` / `muptiEnableCallback()` / `muptiEnableDomain()`	回调函数管理

muPTI 开发者指南

概述

主要组件

快速开始

Activity API 使用流程

完整示例

Activity API

启用 Activity 记录

注册回调函数

刷新和获取 Activity 记录

刷新缓冲区

Activity 类型

Activity Record 结构

最佳实践

性能开销最小化

数据准确性

编译和运行

Callback API - 回调函数

注册回调函数

支持的回调类型

故障排除

常见问题

调试技巧

API 参考

核心 API 摘要

相关文档

这篇文章对您有帮助吗？

概述​

主要组件​

快速开始​

Activity API 使用流程​

完整示例​

Activity API​

启用 Activity 记录​

注册回调函数​

刷新和获取 Activity 记录​

刷新缓冲区​

Activity 类型​

Activity Record 结构​

最佳实践​

性能开销最小化​

数据准确性​

编译和运行​

Callback API - 回调函数​

注册回调函数​

支持的回调类型​

故障排除​

常见问题​

调试技巧​

API 参考​

核心 API 摘要​

相关文档​

这篇文章对您有帮助吗？

概述

主要组件

快速开始

Activity API 使用流程

完整示例

Activity API

启用 Activity 记录

注册回调函数

刷新和获取 Activity 记录

刷新缓冲区

Activity 类型

Activity Record 结构

最佳实践

性能开销最小化

数据准确性

编译和运行

Callback API - 回调函数

注册回调函数

支持的回调类型

故障排除

常见问题

调试技巧

API 参考

核心 API 摘要

相关文档