Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "alg_template_register.h"
12 : #include "reduce_ring.h"
13 :
14 : namespace hccl {
15 1 : ReduceRing::ReduceRing(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher)
16 : {
17 1 : }
18 :
19 2 : ReduceRing::~ReduceRing()
20 : {
21 2 : }
22 :
23 1 : HcclResult ReduceRing::Prepare(u64 reduceAttrBitMap, HcomCollOpInfo *opInfo)
24 : {
25 1 : reduceAttr_ = reduceAttrBitMap;
26 1 : return HCCL_SUCCESS;
27 : }
28 :
29 : // reduce算法的入口函数
30 0 : HcclResult ReduceRing::RunAsync(const u32 rank, const u32 rankSize,
31 : const std::vector<std::shared_ptr<Transport> > &links)
32 : {
33 0 : CHK_SMART_PTR_NULL(dispatcher_);
34 0 : CHK_PTR_NULL(stream_.ptr());
35 0 : bool bRetNull = (!outputMem_ || !inputMem_);
36 0 : CHK_PRT_RET(bRetNull, HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] inputmem or outputmem is null", rank),
37 : HCCL_E_PARA);
38 :
39 0 : HcclResult ret = HCCL_SUCCESS;
40 0 : HCCL_INFO("ReduceRing run: rank[%u] totalrank[%u] root[%u] inputmem[%p] output[%p] count[%llu]", \
41 : rank, rankSize, root_, inputMem_.ptr(), outputMem_.ptr(), count_);
42 :
43 : // 如果ranksize为1, inline reduce和普通跨片reduce操作一致，从input->output
44 0 : if (rankSize == 1) {
45 0 : if (inputMem_ != outputMem_) {
46 0 : ret = HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
47 : }
48 0 : return ret;
49 : }
50 :
51 0 : HCCL_DEBUG("[ReduceRing][RunAsync]rankSize is %u", rankSize);
52 : // 创建reducer & sender
53 0 : senderInfo_.reset(new (std::nothrow) Sender(dataType_, reductionOp_, reduceAttr_));
54 0 : CHK_SMART_PTR_NULL(senderInfo_);
55 :
56 0 : reducerInfo_.reset(new (std::nothrow) Reducer(dataType_, reductionOp_, reduceAttr_));
57 0 : CHK_SMART_PTR_NULL(reducerInfo_);
58 :
59 : // 获取ring algorithm所需的通信连接
60 0 : u32 ringPrevRank = (rank + rankSize - 1) % rankSize;
61 0 : u32 ringNextRank = (rank + 1) % rankSize;
62 :
63 0 : if (links.size() < rankSize) {
64 0 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] Link size is less than rank size", rank);
65 0 : return HCCL_E_INTERNAL;
66 : }
67 :
68 0 : linkLeft_ = links[ringPrevRank];
69 0 : CHK_SMART_PTR_NULL(linkLeft_);
70 :
71 0 : linkRight_ = links[ringNextRank];
72 0 : CHK_SMART_PTR_NULL(linkRight_);
73 :
74 0 : scratch_ = DeviceMem::create(inputMem_.ptr(), inputMem_.size());
75 :
76 0 : u32 dataSize = DataUnitSize(dataType_);
77 0 : if (dataSize == 0) {
78 0 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] unit data size is zero", rank);
79 0 : return HCCL_E_INTERNAL;
80 : }
81 :
82 : // 计算以chunk_size_为最大处理单元时，能够处理的最大数据个数
83 : // 每轮需要操作的数据个数
84 0 : CHK_RET(linkLeft_->TxAck(stream_));
85 0 : CHK_RET(linkRight_->RxAck(stream_));
86 0 : u64 length = count_ * dataSize;
87 :
88 0 : if (rank == root_) {
89 : // root节点只接收数据
90 0 : DeviceMem localSrc = scratch_.range(0, length);
91 0 : DeviceMem dst = outputMem_.range(0, length);
92 0 : HCCL_DEBUG("rank [%u] recv data offset[%llu] size[%llu] reduce", rank, 0, length);
93 :
94 : // 需要从前一节点接收数据,替换reducer接口
95 0 : ret = reducerInfo_->run(dispatcher_, linkLeft_, baseOffset_, localSrc, dst, dst,
96 0 : stream_, DstMemType::RESULT_OUTPUT_MEM);
97 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
98 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] reduce data offset[%llu] size[%llu]", rank, 0, length), ret);
99 :
100 : // 给前一节点发送同步
101 0 : CHK_RET(linkLeft_->TxAck(stream_));
102 0 : CHK_RET(linkLeft_->RxWaitDone(stream_));
103 0 : } else if (ringPrevRank == root_) {
104 : // 本rank的前一节点是root节点，本rank数据拷贝到下一rank，不做reduce操作
105 : // 需要向下一节点拷贝的数据
106 0 : DeviceMem localSrc = scratch_.range(0, length);
107 :
108 : // 数据拷贝和向下一节点发送
109 0 : HCCL_DEBUG("rank [%u] send offset[%llu] size[%llu]", rank, 0, length);
110 :
111 0 : ret = senderInfo_->run(linkRight_, baseOffset_, localSrc, stream_);
112 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
113 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] send scratch offset[%llu] size[%llu] "\
114 : "failed", rank, baseOffset_, length), ret);
115 :
116 : // 等待后一节点同步信号
117 0 : CHK_RET(linkRight_->RxAck(stream_));
118 0 : CHK_RET(linkRight_->TxWaitDone(stream_));
119 0 : } else {
120 : // 其余节点，先接收数据，和自身数据进行reduce操作，结果放入tx中，发送至下一节点
121 : // 剩余需要处理的数据大于满chunk size时，以chunksize为处理单位，否则直接处理剩余数据
122 :
123 : // 接收到的数据和scratch数据运算后，放入output
124 0 : DeviceMem localSrc = scratch_.range(0, length);
125 0 : DeviceMem dst = outputMem_.range(0, length);
126 :
127 : // 用reduce接口封装
128 0 : HCCL_DEBUG("rank[%u] recv data reduce offset[%llu] size[%llu]", rank, 0, length);
129 :
130 0 : ret = reducerInfo_->run(dispatcher_, linkLeft_, baseOffset_, localSrc, localSrc, dst,
131 0 : stream_, DstMemType::RESULT_INPUT_MEM);
132 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
133 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] reducer offset[%llu] size[%llu] failed", rank, 0, length), ret);
134 :
135 : // 给前一节点发送同步
136 0 : CHK_RET(linkLeft_->TxAck(stream_));
137 0 : CHK_RET(linkLeft_->RxWaitDone(stream_));
138 :
139 : // tx数据向下一个节点发送
140 : // 需要再封装接口，只把数据发到tx_mem,send_only
141 0 : HCCL_DEBUG("rank[%u] send localSrc offset[%llu] size[%llu]", rank, 0, length);
142 :
143 0 : ret = senderInfo_->run(linkRight_, baseOffset_, localSrc, stream_);
144 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
145 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] sender offset[%llu] failed", rank, baseOffset_), ret);
146 :
147 : // 等待后一节点同步信号
148 0 : CHK_RET(linkRight_->RxAck(stream_));
149 0 : CHK_RET(linkRight_->TxWaitDone(stream_));
150 0 : }
151 0 : CHK_RET(linkRight_->TxDataSignal(stream_));
152 0 : CHK_RET(linkLeft_->RxDataSignal(stream_));
153 0 : HCCL_INFO("ReduceRing finished: rank[%u]", rank);
154 0 : return HCCL_SUCCESS;
155 : }
156 0 : HcclResult ReduceRing::GetNslbAdjInfo(const u32 rank, const u32 rankSize,
157 : const std::vector<LINK> &links, AdjInfo& nslbAdjInfo)
158 : {
159 0 : if (rankSize == 1) {
160 0 : return HCCL_E_NOT_SUPPORT;
161 : }
162 0 : u32 ringNextRank = (rank + 1) % rankSize;
163 0 : LINK nslbNext = links[ringNextRank];
164 :
165 0 : NslbDpAdjInfo adjInfoStep = {0};
166 0 : nslbAdjInfo.dstRankNum = 1;
167 0 : adjInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
168 0 : adjInfoStep.phaseId = 1;
169 0 : adjInfoStep.rev = 0;
170 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
171 :
172 0 : return HCCL_SUCCESS;
173 0 : }
174 :
175 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_REDUCE_RING, ReduceRing);
176 : } // namespace hccl
|
