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 "reduce_scatter_slim_ring.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 0 : ReduceScatterSlimRing::ReduceScatterSlimRing(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
16 :
17 0 : ReduceScatterSlimRing::~ReduceScatterSlimRing() {}
18 :
19 0 : HcclResult ReduceScatterSlimRing::Prepare(u64 reduceAttrBitMap, HcomCollOpInfo* opInfo)
20 : {
21 : (void)opInfo;
22 0 : reduceAttr_ = reduceAttrBitMap;
23 0 : return HCCL_SUCCESS;
24 : }
25 :
26 0 : HcclResult ReduceScatterSlimRing::RunVectorSourceReducer(
27 : const LINK& link, const std::vector<Slice>& txSlices, const std::vector<Slice>& txSlicetemp)
28 : {
29 : /* 1、对外reduce_scatter，output的大小为每块数据*rank_size。只能发送到对端地址偏移为0开始。
30 : 2、allreduce中使用reduce_scatter，output与Input大小相等，接收和发送偏移相等都为slice.offset */
31 0 : std::vector<SenderMemoryInfo> txMems;
32 0 : for (u32 i = 0; i < txSlices.size(); i++) {
33 0 : DeviceMem srcMem = inputMem_.range(txSlices[i].offset, txSlices[i].size);
34 0 : HCCL_DEBUG(
35 : "send inputmem range[%llu], size[%llu] tx dstmem offset[%llu]", txSlices[i].offset, txSlices[i].size,
36 : txSlicetemp[i].offset);
37 0 : txMems.emplace_back(SenderMemoryInfo{baseOffset_ + txSlicetemp[i].offset, srcMem});
38 0 : }
39 0 : CHK_RET(senderInfo_->run(link, txMems, notifyIdx_, stream_));
40 :
41 0 : return HCCL_SUCCESS;
42 0 : }
43 :
44 0 : HcclResult ReduceScatterSlimRing::RunVectorDestRducer(
45 : const LINK& link, const std::vector<Slice>& rxSlices, const std::vector<Slice>& rxSlicetemp)
46 : {
47 0 : std::vector<ReducerMemoryInfo> rxReduceMems;
48 0 : for (u32 i = 0; i < rxSlices.size(); i++) {
49 0 : DeviceMem dstMem = inputMem_.range(rxSlices[i].offset, rxSlices[i].size);
50 0 : DeviceMem srcMemTemp = scratchMem_.range(rxSlicetemp[i].offset, rxSlicetemp[i].size);
51 0 : HCCL_DEBUG(
52 : "rcv offset[%llu], size[%llu] ,then reduce with "
53 : "offset[%llu] size[%llu] ",
54 : rxSlicetemp[i].offset, rxSlicetemp[i].size, rxSlices[i].offset, rxSlices[i].size);
55 0 : rxReduceMems.emplace_back(ReducerMemoryInfo{baseOffset_ + rxSlices[i].offset, dstMem, dstMem, srcMemTemp});
56 0 : }
57 0 : CHK_RET(reducerInfo_->run(dispatcher_, link, rxReduceMems, notifyIdx_, stream_));
58 :
59 0 : return HCCL_SUCCESS;
60 0 : }
61 :
62 0 : HcclResult ReduceScatterSlimRing::RunVectorFinRducer(
63 : const u32 rank, const LINK& link, const u32 sliceSize, const std::vector<Slice>& inputSlices,
64 : const std::vector<Slice>& outputSlices)
65 : {
66 0 : std::vector<ReducerMemoryInfo> rxReduceMems;
67 0 : for (u32 i = 0; i < sliceSize; i++) {
68 : DeviceMem dstMem
69 0 : = outputMem_.range(outputSlices[rank * sliceSize + i].offset, outputSlices[rank * sliceSize + i].size);
70 : // reduce目的操作
71 : DeviceMem srcMem
72 0 : = inputMem_.range(inputSlices[rank * sliceSize + i].offset, inputSlices[rank * sliceSize + i].size);
73 : DeviceMem scratchMem
74 0 : = scratchMem_.range(outputSlices[rank * sliceSize + i].offset, outputSlices[rank * sliceSize + i].size);
75 0 : rxReduceMems.emplace_back(
76 0 : ReducerMemoryInfo{baseOffset_ + inputSlices[rank * sliceSize + i].offset, srcMem, dstMem, scratchMem});
77 0 : }
78 0 : CHK_RET(reducerInfo_->run(dispatcher_, link, rxReduceMems, notifyIdx_, stream_));
79 :
80 0 : notifyIdx_++;
81 :
82 0 : return HCCL_SUCCESS;
83 0 : }
84 :
85 0 : HcclResult ReduceScatterSlimRing::RunSourceReducer(const LINK& link, const Slice& txSlice, const Slice& txSlicetemp)
86 : {
87 : /* 1、对外reduce_scatter，output的大小为每块数据*rank_size。只能发送到对端地址偏移为0开始。
88 : 2、allreduce中使用reduce_scatter，output与Input大小相等，接收和发送偏移相等都为slice.offset */
89 0 : DeviceMem srcMem = inputMem_.range(txSlice.offset, txSlice.size);
90 0 : HCCL_DEBUG(
91 : " send inputmem range[%llu], size[%llu] tx dstmem offset[%llu]", txSlice.offset, txSlice.size,
92 : txSlicetemp.offset);
93 0 : CHK_RET(senderInfo_->run(link, baseOffset_ + txSlicetemp.offset, srcMem, stream_));
94 :
95 0 : return HCCL_SUCCESS;
96 0 : }
97 :
98 0 : HcclResult ReduceScatterSlimRing::RunDestRducer(const LINK& link, const Slice& rxSlice, const Slice& rxSlicetemp)
99 : {
100 0 : DeviceMem dstMem = inputMem_.range(rxSlice.offset, rxSlice.size);
101 0 : DeviceMem srcMemTemp = scratchMem_.range(rxSlicetemp.offset, rxSlicetemp.size);
102 0 : HCCL_DEBUG(
103 : "rcv offset[%llu], size[%llu] ,then reduce with "
104 : "offset[%llu] size[%llu] ",
105 : rxSlicetemp.offset, rxSlicetemp.size, rxSlice.offset, rxSlice.size);
106 0 : CHK_RET(reducerInfo_->run(dispatcher_, link, baseOffset_ + rxSlice.offset, dstMem, dstMem, srcMemTemp, stream_));
107 :
108 0 : return HCCL_SUCCESS;
109 0 : }
110 :
111 0 : HcclResult ReduceScatterSlimRing::InitSlice(std::vector<Slice>& outputSlices, u32 rank, u32 rankSize, u32 unitSize)
112 : {
113 0 : if (slices_.size() == 0) {
114 0 : slices_.resize(rankSize);
115 0 : outputSlices.resize(rankSize);
116 0 : u64 sliceSize = count_ * unitSize;
117 0 : for (u32 i = 0; i < rankSize; i++) {
118 0 : slices_[i].size = sliceSize;
119 0 : slices_[i].offset = (i * sliceSize);
120 0 : outputSlices[i].size = sliceSize;
121 0 : outputSlices[i].offset = (inputMem_.size() > outputMem_.size()) ? 0 : (i * sliceSize);
122 0 : HCCL_DEBUG(
123 : "rank[%u], slices[%u].offset=[%llu], slices[%u].size=[%llu] outputSlices[%u].offset=[%llu], "
124 : "outputSlices[%u].size=[%llu] ",
125 : rank, i, slices_[i].offset, i, slices_[i].size, i, outputSlices[i].offset, i, outputSlices[i].size);
126 : }
127 : }
128 0 : return HCCL_SUCCESS;
129 : }
130 :
131 0 : HcclResult ReduceScatterSlimRing::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
132 : {
133 : // 判断stream, dispatcher是否为空
134 0 : CHK_SMART_PTR_NULL(dispatcher_);
135 0 : CHK_PTR_NULL(stream_.ptr());
136 0 : if (!outputMem_ || !inputMem_) {
137 0 : HCCL_ERROR("[ReduceScatterSlimRing][RunAsync]rank[%u] run_async inputmem or outputmem is null", rank);
138 0 : return HCCL_E_PTR;
139 : }
140 0 : HCCL_INFO(
141 : "ReduceScatterRing run: rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", rank, rankSize,
142 : inputMem_.ptr(), outputMem_.ptr(), count_);
143 :
144 : // 判断rank_size == 1
145 0 : if (rankSize == 1) {
146 0 : if (inputMem_ != outputMem_) {
147 0 : return HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
148 : }
149 0 : return HCCL_SUCCESS;
150 : }
151 :
152 : // 创建reducer & sender
153 0 : senderInfo_.reset(new (std::nothrow) Sender(dataType_, reductionOp_, reduceAttr_));
154 0 : CHK_SMART_PTR_NULL(senderInfo_);
155 :
156 0 : reducerInfo_.reset(new (std::nothrow) Reducer(dataType_, reductionOp_, reduceAttr_));
157 0 : CHK_SMART_PTR_NULL(reducerInfo_);
158 :
159 0 : if (links.size() < rankSize) {
160 0 : HCCL_ERROR(
161 : "[ReduceScatterSlimRing][RunAsync]rank[%u] link size[%llu] is less than rank size[%u]", rank, links.size(),
162 : rankSize);
163 0 : return HCCL_E_INTERNAL;
164 : }
165 :
166 0 : u32 ringPrevRank = (rank + rankSize - 1) % rankSize;
167 0 : linkLeft_ = links[ringPrevRank];
168 0 : CHK_SMART_PTR_NULL(linkLeft_);
169 :
170 0 : u32 ringNextRank = (rank + 1) % rankSize;
171 0 : linkRight_ = links[ringNextRank];
172 0 : CHK_SMART_PTR_NULL(linkRight_);
173 :
174 0 : u32 unitSize = DataUnitSize(dataType_);
175 0 : if (unitSize == 0) {
176 0 : HCCL_ERROR("[ReduceScatterSlimRing][RunAsync]rank[%u] unit data size is zero", rank);
177 0 : return HCCL_E_INTERNAL;
178 : }
179 :
180 0 : std::vector<Slice> outputSlices(slices_);
181 0 : InitSlice(outputSlices, rank, rankSize, unitSize);
182 : // 运行reduce-scatter, ring算法
183 : // 单环场景下 nicRankList_ 长度默认为 8。
184 : // 多环场景下 nicRankList_ 长度为网口数量。此时若 rankSize != nicRankList_ 则为网口裁剪场景
185 0 : if (rankSize != HCCL_NIC_MAX_NUM || nicRankList_.size() == HCCL_NIC_MAX_NUM) {
186 : // 非网口裁剪场景:
187 0 : CHK_RET(RunReduceScatter(rank, rankSize, slices_, outputSlices));
188 : }
189 :
190 0 : if (barrierSwitchOn_) {
191 : // 执行barrier，保证数据收发完成
192 0 : CHK_RET(ExecuteBarrier(linkRight_, linkLeft_, notifyIdx_));
193 0 : notifyIdx_++;
194 : }
195 :
196 0 : HCCL_INFO("ReduceScatterRing finished: rank[%u]", rank);
197 0 : return HCCL_SUCCESS;
198 0 : }
199 :
200 0 : HcclResult ReduceScatterSlimRing::RunReduceScatter(
201 : const u32 rank, const u32 rankSize, const std::vector<Slice>& inputSlices, const std::vector<Slice>& outputSlices)
202 : {
203 0 : bool bRetSize = (inputSlices.size() < rankSize);
204 0 : CHK_PRT_RET(
205 : bRetSize,
206 : HCCL_ERROR(
207 : "[Run][ReduceScatter]rank[%u] inputslice size[%llu] is less than rank size[%u]", rank, inputSlices.size(),
208 : rankSize),
209 : HCCL_E_INTERNAL);
210 :
211 0 : bRetSize = (outputSlices.size() < rankSize);
212 0 : CHK_PRT_RET(
213 : bRetSize,
214 : HCCL_ERROR(
215 : "[Run][ReduceScatter]rank[%u] outputslice size[%llu] is less than rank size[%u]", rank, outputSlices.size(),
216 : rankSize),
217 : HCCL_E_INTERNAL);
218 :
219 0 : HcclResult ret = HCCL_SUCCESS;
220 :
221 0 : u32 sliceSize = inputSlices.size() / rankSize;
222 :
223 : // 获取rx_slice, 首先向本rank前2个rank处发ack消息
224 0 : u32 rxSliceIndex = (rank + rankSize - 2) % rankSize;
225 :
226 : // reduce源操作, 获取tx_slice，从本rank前一rank开始接收ack
227 0 : u32 txSliceIndex = (rank + rankSize - 1) % rankSize;
228 :
229 0 : std::vector<Slice> txInputSegsSlice;
230 0 : std::vector<Slice> txOutputSegsSlice;
231 0 : for (u32 j = 0; j < sliceSize; j++) {
232 0 : txInputSegsSlice.push_back(inputSlices[txSliceIndex * sliceSize + j]);
233 0 : txOutputSegsSlice.push_back(outputSlices[txSliceIndex * sliceSize + j]);
234 : }
235 0 : ret = RunVectorSourceReducer(linkRight_, txInputSegsSlice, txOutputSegsSlice); // NotifyRecord
236 0 : CHK_PRT_RET(
237 : ret != HCCL_SUCCESS,
238 : HCCL_ERROR("[Run][ReduceScatter]rank[%u] txSliceIndex[%u] Reducer src run failed", rank, txSliceIndex), ret);
239 :
240 : // 本rank既当reduce源, 也当reduce操作的目的
241 0 : for (u32 i = 0; i < (rankSize - 2); i++) { // 中间rank_size - 2次传输
242 : // reduce目的操作
243 0 : std::vector<Slice> rxInputSegsSlice;
244 0 : std::vector<Slice> rxOutputSegsSlice;
245 0 : for (u32 j = 0; j < sliceSize; j++) {
246 0 : rxInputSegsSlice.push_back(inputSlices[rxSliceIndex * sliceSize + j]);
247 0 : rxOutputSegsSlice.push_back(outputSlices[rxSliceIndex * sliceSize + j]);
248 : }
249 0 : ret = RunVectorDestRducer(linkLeft_, rxInputSegsSlice, rxOutputSegsSlice);
250 :
251 0 : CHK_PRT_RET(
252 : ret != HCCL_SUCCESS,
253 : HCCL_ERROR(
254 : "[Run][ReduceScatter]rank[%u] round[%u] rxSlice[%u] Reducer dst run failed", rank, i, rxSliceIndex),
255 : ret);
256 :
257 0 : notifyIdx_++;
258 :
259 : // 获取rx_slice
260 0 : rxSliceIndex = (rxSliceIndex + rankSize - 1) % rankSize;
261 :
262 : // reduce源操作, 获取tx_slice
263 0 : txSliceIndex = (txSliceIndex + rankSize - 1) % rankSize;
264 :
265 0 : std::vector<Slice> txInputSlice;
266 0 : std::vector<Slice> txOutputSlice;
267 0 : for (u32 j = 0; j < sliceSize; j++) {
268 0 : txInputSlice.push_back(inputSlices[txSliceIndex * sliceSize + j]);
269 0 : txOutputSlice.push_back(outputSlices[txSliceIndex * sliceSize + j]);
270 : }
271 0 : ret = RunVectorSourceReducer(linkRight_, txInputSlice, txOutputSlice);
272 0 : CHK_PRT_RET(
273 : ret != HCCL_SUCCESS, HCCL_ERROR("[Run][ReduceScatter]rank[%u] round[%u] Reducer src run failed", rank, i),
274 : ret);
275 0 : }
276 :
277 : /* * 末尾传输, 本rank只当reduce目的, 根据单buffer还是双buffer来决定如何搬移
278 : 当前简化处理, 只考虑单buffer的场景, 双buffer则在run_async中多拷贝一次 */
279 0 : RunVectorFinRducer(rank, linkLeft_, sliceSize, inputSlices, outputSlices);
280 :
281 0 : return HCCL_SUCCESS;
282 0 : }
283 :
284 0 : HcclResult ReduceScatterSlimRing::SetNotifyIdx(u32 notifyIdx)
285 : {
286 0 : notifyIdx_ = notifyIdx;
287 0 : return HCCL_SUCCESS;
288 : }
289 :
290 0 : HcclResult ReduceScatterSlimRing::GetNotifyIdx(u32& notifyIdx)
291 : {
292 0 : notifyIdx = notifyIdx_;
293 0 : return HCCL_SUCCESS;
294 : }
295 :
296 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_REDUCESCATTER_SLIM_RING, ReduceScatterSlimRing);
297 : } // namespace hccl
|
