本文基于CANN开源社区的hccl仓库进行技术解读

前言

大模型训练需要多机多卡并行，但如何让多张NPU协同工作？如何高效地交换数据？

hccl（Huawei Collective Communication Library，华为集合通信库）就是解决这个问题的。它提供了类似NCCL的集合通信接口，让多卡训练变得简单高效。

什么是hccl

hccl是CANN提供的集合通信库，支持：

  • 多机多卡通信
  • 高性能集合通信原语
  • RDMA支持
  • 与主流框架集成

简单说，就是昇腾NPU的"NCCL"。

核心通信原语

1. Broadcast（广播）
// 广播：从一个节点发送数据到所有节点
#include "hccl/hccl.h"

// 初始化hccl
hcclComm_t comm;
hcclCommInitRank(&comm, rank, nranks);

// 广播数据
hcclResult_t ret = hcclBroadcast(
    data,      // 数据指针
    count,     // 数据数量
    datatype,  // 数据类型
    root,      // 根节点rank
    comm,      // 通信域
    stream     // 流
);

// 等待完成
hcclStreamSynchronize(stream);
2. AllReduce（全规约）
// AllReduce：所有节点求和/求平均/求最大/求最小
hcclResult_t ret = hcclAllReduce(
    sendbuf,   // 发送缓冲区
    recvbuf,   // 接收缓冲区
    count,     // 数据数量
    datatype,  // 数据类型
    op,        // 操作：HCCL_SUM, HCCL_AVG, HCCL_MAX, HCCL_MIN
    comm,      // 通信域
    stream     // 流
);
3. AllGather（全收集）
// AllGather：收集所有节点的数据
hcclResult_t ret = hcclAllGather(
    sendbuf,   // 发送缓冲区
    recvbuf,   // 接收缓冲区
    count,     // 每个节点的数据数量
    datatype,  // 数据类型
    comm,      // 通信域
    stream     // 流
);

// recvbuf将包含所有节点的数据
// [rank0_data, rank1_data, rank2_data, ...]
4. ReduceScatter（规约散布）
// ReduceScatter：规约后分发到不同节点
hcclResult_t ret = hcclReduceScatter(
    sendbuf,   // 发送缓冲区
    recvbuf,   // 接收缓冲区
    count,     // 每个节点接收的数据数量
    datatype,  // 数据类型
    op,        // 操作
    comm,      // 通信域
    stream     // 流
);

通信模式

1. Ring AllReduce
// Ring AllReduce模式
// 适合大规模训练，通信效率高

节点0: 1 2 3 4
节点1: 5 6 7 8
节点2: 9 10 11 12
节点3: 13 14 15 16

Reduce-Scatter阶段:
节点0: 1 6 11 16
节点1: 5 10 15 4
节点2: 9 14 3 8
节点3: 13 2 7 12

AllGather阶段:
节点0: 28 28 28 28
节点1: 28 28 28 28
节点2: 28 28 28 28
节点3: 28 28 28 28
2. Tree AllReduce
// Tree AllReduce模式
// 适合小规模训练，延迟低

        节点0
       /    \
    节点1    节点2
     /  \    /  \
  节点3 节点4 节点5 节点6

// 两阶段树形规约
// 第一阶段：规约到根节点
// 第二阶段：从根节点广播
3. 通信模式对比

带宽利用率高

延迟高

延迟低

带宽利用率低

混合模式

Ring AllReduce

适合大规模

不适合小规模

Tree AllReduce

适合小规模

不适合大规模

NCCL2

平衡性能

多机训练

1. 初始化多机通信
// 多机初始化
#include "hccl/hccl.h"

// 获取rank和nranks
int rank = 0;
int nranks = 0;
hcclGetRank(&rank);
hcclGetNranks(&nranks);

// 初始化通信域
hcclComm_t comm;
hcclCommInitRank(&comm, rank, nranks);

// 获取设备信息
int device_id = 0;
hcclGetDeviceId(&device_id);
2. 数据并行训练
// 数据并行训练
void data_parallel_training(
    Model& model,
    DataLoader& dataloader,
    int epochs)
{
    for (int epoch = 0; epoch < epochs; epoch++) {
        for (auto& batch : dataloader) {
            // 前向传播
            auto output = model.forward(batch);

            // 计算loss
            auto loss = compute_loss(output, batch.label);

            // 反向传播
            model.backward(loss);

            // AllReduce梯度
            for (auto& param : model.parameters()) {
                hcclAllReduce(
                    param.grad().data(),
                    param.grad().data(),
                    param.grad().numel(),
                    HCCL_DATA_TYPE_FP32,
                    HCCL_OP_SUM,
                    comm,
                    stream
                );
            }

            // 平均梯度
            for (auto& param : model.parameters()) {
                param.grad().div_(nranks);
            }

            // 更新参数
            model.update();
        }
    }
}
3. 梯度累积
// 梯度累积 + AllReduce
void gradient_accumulation(
    Model& model,
    DataLoader& dataloader,
    int accumulation_steps)
{
    int step = 0;
    for (auto& batch : dataloader) {
        // 前向传播
        auto output = model.forward(batch);

        // 计算loss
        auto loss = compute_loss(output, batch.label);
        loss.div_(accumulation_steps);

        // 反向传播
        model.backward(loss);

        step++;

        // 达到累积步数
        if (step % accumulation_steps == 0) {
            // AllReduce梯度
            for (auto& param : model.parameters()) {
                hcclAllReduce(
                    param.grad().data(),
                    param.grad().data(),
                    param.grad().numel(),
                    HCCL_DATA_TYPE_FP32,
                    HCCL_OP_SUM,
                    comm,
                    stream
                );

                // 平均梯度
                param.grad().div_(nranks);
            }

            // 更新参数
            model.update();

            // 清零梯度
            model.zero_grad();
        }
    }
}

性能优化

1. 通信重叠
// 重叠计算和通信
void overlap_compute_communication(
    Model& model,
    DataLoader& dataloader)
{
    // 创建多个流
    hcclStream_t compute_stream;
    hcclStream_t comm_stream;
    hcclStreamCreate(&compute_stream);
    hcclStreamCreate(&comm_stream);

    for (auto& batch : dataloader) {
        // 在compute_stream上计算
        auto output = model.forward(batch, compute_stream);

        // 在comm_stream上通信
        hcclAllReduce(
            param.grad().data(),
            param.grad().data(),
            param.grad().numel(),
            HCCL_DATA_TYPE_FP32,
            HCCL_OP_SUM,
            comm,
            comm_stream
        );

        // 同步流
        hcclStreamSynchronize(compute_stream);
        hcclStreamSynchronize(comm_stream);
    }
}
2. 混合精度通信
// 使用FP16进行通信
void mixed_precision_comm(
    Model& model,
    DataLoader& dataloader)
{
    for (auto& batch : dataloader) {
        // 前向传播（FP32）
        auto output = model.forward(batch);

        // 反向传播
        model.backward(output);

        // 转换为FP16
        for (auto& param : model.parameters()) {
            auto grad_fp16 = param.grad().to(HCCL_DATA_TYPE_FP16);

            // FP16 AllReduce
            hcclAllReduce(
                grad_fp16.data(),
                grad_fp16.data(),
                grad_fp16.numel(),
                HCCL_DATA_TYPE_FP16,
                HCCL_OP_SUM,
                comm,
                stream
            );

            // 转换回FP32
            param.grad().copy_(grad_fp16);
        }

        // 平均梯度
        for (auto& param : model.parameters()) {
            param.grad().div_(nranks);
        }

        // 更新参数
        model.update();
    }
}

故障恢复

1. 检测节点故障
// 检测节点故障
int check_node_health(int rank) {
    // 发送心跳
    hcclResult_t ret = hcclSend(
        heartbeat_data,
        sizeof(heartbeat_data),
        peer_rank,
        comm,
        stream
    );

    if (ret != HCCL_SUCCESS) {
        printf("Node %d failed\n", rank);
        return -1;
    }

    return 0;
}
2. 故障恢复
// 故障恢复
void recover_from_failure(int failed_rank) {
    // 重新初始化通信域
    hcclCommDestroy(comm);
    hcclCommInitRank(&comm, new_rank, new_nranks);

    // 重新加载checkpoint
    load_checkpoint(restore_path);

    // 恢复训练
    resume_training();
}

与框架集成

1. PyTorch集成
import torch
import torch.distributed as dist

# 初始化hccl
torch.distributed.init_process_group(
    backend='hccl',
    rank=rank,
    world_size=world_size
)

# 数据并行
model = torch.nn.parallel.DistributedDataParallel(
    model,
    device_ids=[local_rank]
)

# 训练
for batch in dataloader:
    output = model(batch)
    loss = criterion(output, labels)
    loss.backward()
    optimizer.step()
2. TensorFlow集成
import tensorflow as tf

# 初始化hccl
strategy = tf.distribute.MultiWorkerMirroredStrategy()

# 在strategy下创建模型
with strategy.scope():
    model = create_model()

# 训练
model.fit(dataset, epochs=epochs)

性能分析

1. 通信性能监控
// 监控通信性能
void monitor_communication_performance() {
    // 记录开始时间
    auto start = std::chrono::high_resolution_clock::now();

    // 执行AllReduce
    hcclAllReduce(...);

    // 记录结束时间
    auto end = std::chrono::high_resolution_clock::now();

    // 计算耗时
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);

    printf("AllReduce time: %ld ms\n", duration.count());

    // 计算带宽
    size_t data_size = count * sizeof(datatype);
    double bandwidth = (double)data_size / duration.count() / 1e6;
    printf("Bandwidth: %.2f GB/s\n", bandwidth);
}
2. 通信热点分析
// 分析通信热点
void analyze_communication_hotspots() {
    // 记录每个通信操作的耗时
    std::map<std::string, double> comm_times;

    // AllReduce
    comm_times["AllReduce"] = measure_allreduce_time();

    // AllGather
    comm_times["AllGather"] = measure_allgather_time();

    // Broadcast
    comm_times["Broadcast"] = measure_broadcast_time();

    // 找出最慢的操作
    auto slowest = std::max_element(
        comm_times.begin(),
        comm_times.end(),
        [](const auto& a, const auto& b) {
            return a.second < b.second;
        }
    );

    printf("Slowest operation: %s (%.2f ms)\n",
           slowest->first.c_str(),
           slowest->second);
}

常见问题

Q1：hccl和NCCL有什么区别？

接口类似，但hccl专门针对昇腾NPU优化，与CANN深度集成。

Q2：如何选择通信模式？

小规模（<8卡）用Tree，大规模（>8卡）用Ring。

Q3：如何提高通信效率？

使用混合精度、通信重叠、梯度累积等优化技术。

总结

hccl是昇腾NPU的集合通信库，主要特点：

  • 提供丰富的集合通信原语
  • 支持多种通信模式
  • 高性能通信
  • 与主流框架集成

对于多机多卡训练，hccl是核心组件。

相关链接

本文基于hccl仓库公开信息撰写，如有错误欢迎指正。

# 3d # 目标检测 # 目标跟踪
Logo
昇腾开源生态专区

昇腾计算产业是基于昇腾系列（HUAWEI Ascend）处理器和基础软件构建的全栈 AI计算基础设施、行业应用及服务，https://devpress.csdn.net/organization/setting/general/146749包括昇腾系列处理器、系列硬件、CANN、AI计算框架、应用使能、开发工具链、管理运维工具、行业应用及服务等全产业链

更多推荐

cover

CANN shmem 共享内存通信模型的进程间同步机制

avatar 昇腾开源生态专区
cover

[特殊字符]提升强化学习效率的开源框架——slime

avatar 昇腾开源生态专区
cover

GLM-5上线魔乐社区，基于昇腾的模型推理+训练部署教程请查收！

avatar 昇腾开源生态专区