HCCL 06: 14 个通信算子：从 AllReduce 到 BatchSendRecv

依据官方 API 文档（comm_op_interface/collective_comm.md 与 p2p_comm.md）——HcclAlltoAllVC 与 HcclAlltoAll/HcclAlltoAllV 同列"集合通信"组，其功能说明亦明确写为"集合通信算子 AlltoAllVC"；点对点组仅含 HcclSend/HcclRecv/HcclBatchSendRecv。

每个算子一个模块（ops/all_reduce.py、ops/broadcast.py …），ops/__init__.py 扁平再导出。所有模块结构相同：检查 stream → 调 C 函数 → 检查返回码 → 返回。

算子接口统一以 hccl_comm 为首参、stream 为末参，中间参数与 C 接口逐一对应：

all_reduce(hccl_comm, send_data, recv_data, count, data_type, reduce_op, stream)
broadcast(hccl_comm, buf, count, data_type, root, stream)
send(hccl_comm, send_data, count, data_type, dest_rank, stream)

• send_data/recv_data/buf：device 内存指针（以 ctypes.c_void_p 传入）。注意：pyACL 的 acl.rt.malloc(size, policy) 返回 (dev_ptr, ret) 元组，必须取 [0] 再传入，不要把整个元组传给绑定层：

  dev_buf, ret = acl.rt.malloc(nbytes, 0)   # ACL_MEM_MALLOC_HUGE_ONLY = 0
  # 传 dev_buf，不要传 (dev_buf, ret)
  

• count：元素个数（非字节数）；

• data_type：hccl4py.types.HcclDataType 枚举；

• reduce_op：hccl4py.types.HcclReduceOp（SUM/PROD/MAX/MIN）；

• root/dest_rank/src_rank：uint32；

• stream：pyACL acl.rt.create_stream() 产出的句柄。

V 家族的数组参数

算子名称中的 “V” 代表 “Vector”（向量），指的是这类算子用数组（向量） 来为每个 rank 分别指定数据量，而不是像普通版本那样只用一个标量值。displs 是 displacement（位移/偏移量） 的缩写。

这些 counts/displs 数组是通信任务的“元数据”，HCCL 在任务提交阶段就需要在 CPU/HOST 侧读取并校验它们

V 家族三个算子的数组参数并不相同：

对应的 C 原型（摘自官方文档）：

HcclResult HcclAllGatherV(void *sendBuf, uint64_t sendCount, void *recvBuf, const void *recvCounts, const void *recvDispls, HcclDataType dataType, HcclComm comm, aclrtStream stream);
HcclResult HcclReduceScatterV(void *sendBuf, const void *sendCounts, const void *sendDispls, void *recvBuf, uint64_t recvCount, HcclDataType dataType, HcclReduceOp op, HcclComm comm, aclrtStream stream);
HcclResult HcclAlltoAllV(const void *sendBuf, const void *sendCounts, const void *sdispls, HcclDataType sendType, const void *recvBuf, const void *recvCounts, const void *rdispls, HcclDataType recvType, HcclComm comm, aclrtStream stream);

数组均为 uint64 类型，且必须传 host 指针（C 例程也是传 std::vector.data()）。

all_gather_v 示例（只有接收侧数组）：

recv_counts = np.array([8, 8], dtype=np.uint64) # 一维数组，长度等于 rank 数（2），从 rank 0 接收 8 个元素，1同
recv_displs = np.array([0, 8], dtype=np.uint64)

all_gather_v(comm, send_buf, 8, recv_buf,
             ctypes.c_void_p(recv_counts.ctypes.data),   # host 指针
             ctypes.c_void_p(recv_displs.ctypes.data),   # host 指针
             types.HcclDataType.FP32, stream)

all_to_all_v 示例（收发各一组，共 4 个数组）：

send_counts = np.array([4, 4], dtype=np.uint64)
send_displs = np.array([0, 4], dtype=np.uint64)
recv_counts = np.array([4, 4], dtype=np.uint64)
recv_displs = np.array([0, 4], dtype=np.uint64)

all_to_all_v(comm, send_buf, ctypes.c_void_p(send_counts.ctypes.data),
             ctypes.c_void_p(send_displs.ctypes.data), types.HcclDataType.FP32,
             recv_buf, ctypes.c_void_p(recv_counts.ctypes.data),
             ctypes.c_void_p(recv_displs.ctypes.data), types.HcclDataType.FP32,
             stream)

all_to_all_v 的 send_counts[i] 表示本 rank 发给 rank i 的元素数，recv_counts[i] 表示本 rank 从 rank i 收到的元素数——发送和接收是两组独立数组，不能只传一组。

传 device 指针会导致调用出错（官方明确要求数组参数为 host 侧内存）。

all_to_all_vc 的 count 矩阵

VC (Variable Count)：每个 rank 发往其他各个 rank 的数据量可以不同（可变），并且这种“可变”关系是通过一个 rank × rank 的二维矩阵（即 sendCountMatrix）来全局定义的。

all_to_all_vc 的 send_count_matrix 是 rank×rank 的 uint64 二维数组，行主序连续：

# 2 rank: matrix[i][j] = rank i 发给 rank j 的元素数
matrix = np.array([[4, 4],    # rank0 → rank0: 4, rank0 → rank1: 4
                   [4, 4]], dtype=np.uint64)

all_to_all_vc(comm, send_buf, ctypes.c_void_p(matrix.ctypes.data),
              types.HcclDataType.FP32, recv_buf, types.HcclDataType.FP32, stream)

同样必须 host 指针。注意 C 原型中 sendType 与 recvType 是两个独立参数（HcclAlltoAllVC(sendBuf, sendCountMatrix, sendType, recvBuf, recvType, comm, stream)），上面的示例分别传了发送与接收的数据类型。

batch_send_recv 的结构体数组

batch_send_recv 一次提交一组 send/recv 项，每项是 HcclSendRecvItem 结构体：

from hccl4py._binding import HcclSendRecvItem

HCCL_SEND = 0
HCCL_RECV = 1

items = [
    HcclSendRecvItem(sendRecvType=HCCL_SEND, buf=ctypes.c_void_p(send_dev),
                     count=8, dataType=int(types.HcclDataType.FP32), remoteRank=1),
    HcclSendRecvItem(sendRecvType=HCCL_RECV, buf=ctypes.c_void_p(recv_dev),
                     count=8, dataType=int(types.HcclDataType.FP32), remoteRank=1),
]
arr = (HcclSendRecvItem * len(items))(*items)

batch_send_recv(comm, arr, len(items), stream)

绑定层判断传入的是否已经是 ctypes.Array，不是则构造：

arr = (HcclSendRecvItem * n)(*send_recv_items) if not isinstance(send_recv_items, ctypes.Array) \
    else send_recv_items

异步范围的例外：batch_send_recv 的"异步"仅指同一张卡上的收发任务互不阻塞；卡间收发任务仍然是同步的，与 HcclSend/HcclRecv 一样必须一一对应（本 rank 的每个 send/recv 项都要有对端的对应项，否则会挂住）。不要因为"异步提交"的笼统说法而省略对端的配对任务。

异步语义

除 batch_send_recv 的卡间同步约束（见上节）外，所有算子都是异步提交——接口返回仅表示"任务已提交到 stream"，不代表执行完成。读结果前必须同步：

broadcast(comm, cptr(dev_buf), 8, types.HcclDataType.FP32, root=0, stream=stream)
acl.rt.synchronize_stream(stream)    # 同步后结果可用
# 此时才能从 recv_buf 读数据

同步完成前不能释放 device 缓冲区、不能销毁通信域——C 侧还在 stream 上异步访问这些资源。

stream 防御

每个算子入口调 _check_stream：

def _check_stream(stream):
    if stream is None:
        raise ValueError("HCCL requires a non-null aclrtStream; "
                         "create via acl.rt.create_stream() (pyACL returns (stream, ret), take [0])")
    if isinstance(stream, ctypes.c_void_p) and not stream.value:
        raise ValueError("stream is a NULL c_void_p; pass a real aclrtStream handle")
    return stream

空 stream 传给 C 侧返回 HCCL_E_PARA(1)，但错误信息不够明确。Python 侧提前拦截，给出可行动的提示——pyACL 的 create_stream() 返回元组 (stream, ret)，用户容易直接传元组而非 create_stream()[0]。同理，acl.rt.malloc() 也返回 (dev_ptr, ret) 元组，见本文开头参数说明中的提醒。

样例

14 个算子各有一个可独立运行样例（examples/01_basic/01_broadcast.py ~ 14_batchsendrecv.py），结构统一：

def rank_fn(rank, comm, stream, nranks):
    # 准备数据 → device 缓冲 → 调算子 → 同步 → 读回 → 核对
    ...

if __name__ == "__main__":
    run_2rank(rank_fn)    # fork 2 进程，各占一 chip 说明此时用的是单卡双芯

run_2rank 封装了建域、rootInfo 分发、销毁的全流程，样例只需写 rank_fn 里的算子调用逻辑。

复现提示：部分 V 家族算子在特定产品上有场景约束，测试环境（2 卡）恰好满足。若在你的环境跑不通，请先核查官方文档中各算子的"产品支持情况"与"约束说明"，例如：

• reduce_scatter_v（HcclReduceScatterV）：Atlas A3 仅支持单 Server 场景；Atlas 推理系列仅单 Server 且最多 2 卡（4 NPU）；Atlas A2 仅支持多机对称分布。
• all_gather_v（HcclAllGatherV）：A2/推理系列仅支持指定硬件形态。
• all_to_all_v（HcclAlltoAllV）：Atlas 训练系列有 cluster 组网约束；推理系列仅单 Server。

下一篇：踩坑实录。