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 "aligned_reduce_scatter_double_ring.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 32 : AlignedReduceScatterDoubleRing::AlignedReduceScatterDoubleRing(const HcclDispatcher dispatcher)
16 32 : : AlgTemplateBase(dispatcher)
17 32 : {}
18 :
19 32 : AlignedReduceScatterDoubleRing::~AlignedReduceScatterDoubleRing() {}
20 :
21 32 : HcclResult AlignedReduceScatterDoubleRing::Prepare(
22 : DeviceMem& inputMem, DeviceMem& outputMem, DeviceMem& scratchMem, const u64 count, const HcclDataType dataType,
23 : const Stream& stream, const std::vector<std::vector<Slice>>& multRingsSlices, const HcclReduceOp reductionOp,
24 : const u32 root, const u64 baseOffset, const bool disableDMAReduce, const u64 reduceAttrBitMap,
25 : const HcomCollOpInfo* opInfo, const u32 userRank, std::vector<Stream>& subStreams,
26 : const std::vector<std::shared_ptr<LocalNotify>>& mainSignals,
27 : const std::vector<std::shared_ptr<LocalNotify>>& subSignals, const std::vector<std::vector<u32>>& ringsOrders,
28 : const std::vector<std::vector<Slice>>& userMemInputSlicesOfDoubleRing)
29 : {
30 32 : reduceAttr_ = reduceAttrBitMap;
31 32 : opInfo_ = opInfo;
32 32 : userRank_ = userRank;
33 32 : subStreams_ = subStreams;
34 32 : mainSignals_ = mainSignals;
35 32 : subSignals_ = subSignals;
36 32 : ringsOrders_ = ringsOrders;
37 32 : userMemInputSlicesOfDoubleRing_ = userMemInputSlicesOfDoubleRing;
38 32 : return AlgTemplateBase::Prepare(
39 : inputMem, outputMem, scratchMem, count, dataType, stream, multRingsSlices, reductionOp, root, baseOffset,
40 32 : disableDMAReduce);
41 : }
42 :
43 : // reduce scatter ring direct算法的函数入口
44 0 : HcclResult AlignedReduceScatterDoubleRing::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
45 : {
46 : // 基本的检查
47 0 : CHK_RET(CheckParameters(rank, rankSize, links));
48 :
49 : // 判断rank_size == 1的情况，并拷贝
50 0 : if (rankSize == 1) {
51 0 : CHK_RET(OneRankMemcpy());
52 0 : return HCCL_SUCCESS;
53 : }
54 : // 收集本地mem信息
55 0 : CHK_RET(InitSenderReducer());
56 :
57 : // 收集邻居信息
58 0 : CHK_RET(GetInitializedNeighborLinks(rank, rankSize, links));
59 :
60 : // 填充slice_
61 0 : CHK_RET(SetSlices(rank, rankSize));
62 :
63 : // 运行reduce-scatter, ring算法
64 0 : CHK_RET(RunReduceScatter(rank, rankSize));
65 :
66 0 : CHK_RET(LaunchTaskExtend(dispatcher_, stream_, subStreams_));
67 :
68 0 : HCCL_INFO("AlignedReduceScatterDoubleRing finished: rank[%u] end", rank);
69 0 : return HCCL_SUCCESS;
70 : }
71 :
72 : HcclResult
73 0 : AlignedReduceScatterDoubleRing::CheckParameters(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
74 : {
75 0 : CHK_PTR_NULL(opInfo_);
76 0 : CHK_RET(CheckConcurrentDirectParameters(rank, rankSize, links));
77 : // 判断subStreams数量是否正确
78 0 : CHK_PRT_RET(
79 : subStreams_.size() < 1,
80 : HCCL_ERROR("[AlignedReduceScatterDoubleRing] subStreams size[%u] is less than 1", subStreams_.size()),
81 : HCCL_E_PARA);
82 0 : for (auto& s : subStreams_) {
83 0 : CHK_PTR_NULL(s.ptr());
84 : }
85 : // 判断mainSignals数量是否正确
86 0 : CHK_PRT_RET(
87 : mainSignals_.size() < 1,
88 : HCCL_ERROR("[AlignedReduceScatterDoubleRing] mainSignals size[%u] is less than 1", mainSignals_.size()),
89 : HCCL_E_PARA);
90 : // 判断subSignals数量是否正确
91 0 : CHK_PRT_RET(
92 : subSignals_.size() < 1,
93 : HCCL_ERROR("[AlignedReduceScatterDoubleRing] subSignals size[%u] is less than 1", subSignals_.size()),
94 : HCCL_E_PARA);
95 : // 判断ringsOrder数量是否正确
96 0 : for (u32 ringIndex = 0; ringIndex < ringsOrders_.size(); ringIndex++) {
97 0 : CHK_PRT_RET(
98 : ringsOrders_[ringIndex].size() != rankSize,
99 : HCCL_ERROR(
100 : "[AlignedReduceScatterDoubleRing] ringsOrders[%u] size[%u] is not equal to rank size[%u]", ringIndex,
101 : ringsOrders_[ringIndex].size(), rankSize),
102 : HCCL_E_PARA);
103 : }
104 : // 判断userMemInputSlices数量是否正确
105 0 : for (u32 ringIndex = 0; ringIndex < userMemInputSlicesOfDoubleRing_.size(); ringIndex++) {
106 0 : CHK_PRT_RET(
107 : userMemInputSlicesOfDoubleRing_[ringIndex].size() % rankSize != 0,
108 : HCCL_ERROR(
109 : "[AlignedReduceScatterDoubleRing] userMemInputSlicesOfDoubleRing[%u] size[%u] can not divided by "
110 : "size[%u]",
111 : ringIndex, userMemInputSlicesOfDoubleRing_[ringIndex].size(), rankSize),
112 : HCCL_E_PARA);
113 : }
114 0 : u32 mainSliceSize = multRingsSlices_[ALIGNED_MAIN_RING_INDEX].size() / rankSize;
115 0 : u32 subSliceSize = multRingsSlices_[ALIGNED_SUB_RING_INDEX].size() / rankSize;
116 0 : CHK_PRT_RET(
117 : mainSliceSize != subSliceSize,
118 : HCCL_ERROR(
119 : "[AlignedReduceScatterDoubleRing] mainSliceSize[%u] is not equal to subSliceSize[%u].", mainSliceSize,
120 : subSliceSize),
121 : HCCL_E_PARA);
122 0 : HCCL_INFO("AlignedReduceScatterDoubleRing finished to CheckParameters");
123 0 : return HCCL_SUCCESS;
124 : }
125 :
126 0 : HcclResult AlignedReduceScatterDoubleRing::OneRankMemcpy()
127 : {
128 0 : CHK_RET(MainRecordSub()); // 主流通知从流开始通信
129 0 : CHK_RET(SubWaitMain()); // 从流等待主流通知
130 0 : for (u32 ringIndex = 0; ringIndex < multRingsSlices_.size(); ringIndex++) {
131 0 : for (u32 sliceIdx = 0; sliceIdx < multRingsSlices_[ringIndex].size(); sliceIdx++) {
132 0 : const Slice& srcSlice = userMemInputSlicesOfDoubleRing_[ringIndex][sliceIdx];
133 0 : const Slice& dstSlice = multRingsSlices_[ringIndex][sliceIdx];
134 0 : DeviceMem src = DeviceMem::create(static_cast<u8*>(opInfo_->inputAddr) + srcSlice.offset, srcSlice.size);
135 0 : DeviceMem dst;
136 0 : if (opInfo_->outputAddr != nullptr) {
137 : // opInfo_->outputAddr != nullptr指示要将输出发送至user output
138 0 : u64 stepOffset = multRingsSlices_[ringIndex][ringsOrders_[ringIndex][0]].offset;
139 0 : HCCL_DEBUG(
140 : "Memcpy operation: stream[main], rank[%u] starts to rcv offset[%llu], size[%llu] at userMemOut_",
141 : userRank_, stepOffset, dstSlice.size);
142 0 : dst = DeviceMem::create(static_cast<u8*>(opInfo_->outputAddr) + stepOffset, dstSlice.size);
143 : } else {
144 : // opInfo_->outputAddr == nullptr指示要将输出发送至CCL buffer
145 0 : HCCL_DEBUG(
146 : "Memcpy operation: stream[main], rank[%u] starts to rcv offset[%llu], size[%llu] at outputMem_",
147 : userRank_, dstSlice.offset, dstSlice.size);
148 0 : dst = outputMem_.range(dstSlice.offset, dstSlice.size);
149 : }
150 0 : if (ringIndex == 1) {
151 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
152 : } else {
153 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, subStreams_[0]));
154 : }
155 0 : }
156 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRing][OneRankMemcpy] ringIndex[%u] Memcpy success", ringIndex);
157 : }
158 0 : CHK_RET(SubRecordMain()); // 从流通知主流通信完成
159 0 : CHK_RET(MainWaitSub()); // 主流等待从流通知
160 0 : return HCCL_SUCCESS;
161 : }
162 :
163 0 : HcclResult AlignedReduceScatterDoubleRing::InitSenderReducer()
164 : {
165 : // 创建reducer & sender
166 0 : senderInfo_.reset(new (std::nothrow) Sender(dataType_, reductionOp_, reduceAttr_));
167 0 : CHK_SMART_PTR_NULL(senderInfo_);
168 :
169 0 : reducerInfo_.reset(new (std::nothrow) Reducer(dataType_, reductionOp_, reduceAttr_));
170 0 : CHK_SMART_PTR_NULL(reducerInfo_);
171 0 : HCCL_INFO("AlignedReduceScatterDoubleRing finished to InitSenderReducer");
172 0 : return HCCL_SUCCESS;
173 : }
174 :
175 0 : HcclResult AlignedReduceScatterDoubleRing::GetInitializedNeighborLinks(
176 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
177 : {
178 : // 收集左邻居信息
179 0 : leftLink_ = links[(rank + rankSize - 1) % rankSize];
180 0 : CHK_SMART_PTR_NULL(leftLink_);
181 :
182 : // 收集右邻居信息
183 0 : rightLink_ = links[(rank + 1) % rankSize];
184 0 : CHK_SMART_PTR_NULL(rightLink_);
185 0 : HCCL_INFO("AlignedReduceScatterDoubleRing finished to GetInitializedNeighborLinks");
186 0 : return HCCL_SUCCESS;
187 : }
188 :
189 0 : HcclResult AlignedReduceScatterDoubleRing::SetSlices(const u32 rank, const u32 rankSize)
190 : {
191 0 : for (u32 ringIndex = 0; ringIndex < multRingsSlices_.size(); ringIndex++) {
192 0 : if (multRingsSlices_[ringIndex].size() == 0) {
193 0 : multRingsSlices_[ringIndex].resize(rankSize);
194 :
195 : // 生成std::vector<Slice> multRingsSlices_[ringIndex]
196 0 : u64 sliceSize = count_ * SIZE_TABLE[dataType_];
197 :
198 0 : for (u32 i = 0; i < rankSize; i++) {
199 0 : multRingsSlices_[ringIndex][i].size = sliceSize;
200 : // 用于DMA消减过程中，消除src与dst不对位的风险
201 0 : multRingsSlices_[ringIndex][i].offset = RoundUpWithDivisor(i * sliceSize, HCCL_MIN_SLICE_ALIGN);
202 :
203 0 : HCCL_DEBUG(
204 : "multRingsSlices_[%u], rank[%u], slices[%u].offset=[%llu], slices[%u].size=[%llu]", ringIndex, rank,
205 : i, multRingsSlices_[ringIndex][i].offset, i, multRingsSlices_[ringIndex][i].size);
206 : }
207 : }
208 0 : for (u32 i = 0; i < multRingsSlices_[ringIndex].size(); i++) {
209 0 : HCCL_DEBUG(
210 : "[AlignedReduceScatterDoubleRing][SetSlices] multRingsSlices_[%u], rank[%u], "
211 : "slices[%u].offset=[%llu], slices[%u].size=[%llu]",
212 : ringIndex, rank, i, multRingsSlices_[ringIndex][i].offset, i, multRingsSlices_[ringIndex][i].size);
213 : }
214 : // 最后一步搬到userMemOut_的offset, 不同的ring环offset不一样
215 : u64 toUserMemOffset;
216 0 : if (ringIndex == 0) {
217 0 : toUserMemOffset = multRingsSlices_[ringIndex][ringsOrders_[ringIndex][0]].offset;
218 : } else {
219 0 : const auto& prevRingSlice = multRingsSlices_[ringIndex - 1][ringsOrders_[ringIndex - 1][rank]];
220 0 : const auto& slice = multRingsSlices_[ringIndex][ringsOrders_[ringIndex][rank]];
221 0 : toUserMemOffset = slice.offset - prevRingSlice.offset;
222 : }
223 0 : HCCL_DEBUG(
224 : "[AlignedReduceScatterDoubleRing][SetSlices] rank[%u], ring[%u], toUserMemOffset[%u]", rank, ringIndex,
225 : toUserMemOffset);
226 0 : lastStepOffsets_.emplace_back(toUserMemOffset);
227 : }
228 0 : HCCL_INFO("AlignedReduceScatterDoubleRing finished to SetSlices");
229 0 : return HCCL_SUCCESS;
230 : }
231 :
232 0 : HcclResult AlignedReduceScatterDoubleRing::PrepareInitSlices(
233 : const u32 rankSize, u64 ringIndex, u32 discontinuousSliceSize, u32 discontinuousSliceIdx, u32 initSlice0Idx,
234 : u32 initSlice1Idx, DeviceMem& dstInit, DeviceMem& srcInit, DeviceMem& dstSubInit, DeviceMem& srcSubInit)
235 : {
236 : // 第-1步，片内将部分数据从userIn搬到cclIn
237 : const Slice& srcInitSlice0
238 0 : = userMemInputSlicesOfDoubleRing_[ringIndex][initSlice0Idx * discontinuousSliceSize + discontinuousSliceIdx];
239 0 : srcInit = DeviceMem::create(static_cast<u8*>(opInfo_->inputAddr) + srcInitSlice0.offset, srcInitSlice0.size);
240 : const Slice& dstInitSlice0
241 0 : = multRingsSlices_[ringIndex][initSlice0Idx * discontinuousSliceSize + discontinuousSliceIdx];
242 0 : dstInit = inputMem_.range(dstInitSlice0.offset, dstInitSlice0.size);
243 :
244 : const Slice& srcInitSlice1
245 0 : = userMemInputSlicesOfDoubleRing_[ringIndex][initSlice1Idx * discontinuousSliceSize + discontinuousSliceIdx];
246 0 : srcSubInit = DeviceMem::create(static_cast<u8*>(opInfo_->inputAddr) + srcInitSlice1.offset, srcInitSlice1.size);
247 : const Slice& dstInitSlice1
248 0 : = multRingsSlices_[ringIndex][initSlice1Idx * discontinuousSliceSize + discontinuousSliceIdx];
249 0 : dstSubInit = inputMem_.range(dstInitSlice1.offset, dstInitSlice1.size);
250 : // 第-1步并发
251 0 : if (rankSize == TWO_RANK_SIZE && opInfo_->outputAddr != nullptr) {
252 0 : HCCL_DEBUG(
253 : "Memcpy operation: step[-1] stream[main] src rank[%u] starts to copy(rcv) offset[%llu], size[%llu] on "
254 : "userMemInput to offset[%llu], size[%llu] on userMemOut_",
255 : userRank_, srcInitSlice1.offset, srcInitSlice1.size, lastStepOffsets_[ringIndex], dstInitSlice1.size);
256 0 : dstInit = DeviceMem::create(
257 0 : static_cast<u8*>(opInfo_->outputAddr) + lastStepOffsets_[ringIndex], dstInitSlice1.size);
258 : } else {
259 0 : HCCL_DEBUG(
260 : "Memcpy operation: step[-1] stream[main] src rank[%u] starts to copy(rcv) offset[%llu], size[%llu] on "
261 : "userMemInput to offset[%llu], size[%llu] on CCL",
262 : userRank_, srcInitSlice1.offset, srcInitSlice1.size, dstInitSlice1.offset, dstInitSlice1.size);
263 : }
264 0 : HCCL_DEBUG(
265 : "Memcpy operation: step[-1] stream[sub] src rank[%u] starts to copy(rcv) offset[%llu], "
266 : "size[%llu] on userMemInput to offset[%llu], size[%llu] on CCL",
267 : userRank_, srcInitSlice0.offset, srcInitSlice0.size, dstInitSlice0.offset, dstInitSlice0.size);
268 0 : return HCCL_SUCCESS;
269 : }
270 :
271 : HcclResult
272 0 : AlignedReduceScatterDoubleRing::MemcpyInitSlicesOnMainStreams(u64 ringIndex, DeviceMem& dstInit, DeviceMem& srcInit)
273 : {
274 0 : if (ringIndex == 1) {
275 0 : CHK_RET(MainWaitSub());
276 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
277 0 : CHK_RET(MainRecordSub());
278 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstInit, srcInit, stream_));
279 : } else {
280 0 : CHK_RET(LocalNotify::Post(subStreams_[0], dispatcher_, mainSignals_[0], profilerInput_.stage));
281 0 : CHK_RET(LocalNotify::Wait(subStreams_[0], dispatcher_, subSignals_[0], profilerInput_.stage));
282 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstInit, srcInit, subStreams_[0]));
283 : }
284 0 : return HCCL_SUCCESS;
285 : }
286 :
287 0 : HcclResult AlignedReduceScatterDoubleRing::MemcpyInitSlices(
288 : u64 ringIndex, DeviceMem& dstInit, DeviceMem& srcInit, DeviceMem& dstSubInit, DeviceMem& srcSubInit)
289 : {
290 0 : CHK_RET(MemcpyInitSlicesOnMainStreams(ringIndex, dstInit, srcInit));
291 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB && (!disableDMAReduce_)) {
292 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRing][MemcpyInitSlices] no graph mode");
293 0 : CHK_RET(LocalNotify::Post(
294 : subStreams_[ringIndex + 1], dispatcher_, mainSignals_[ringIndex + 1], profilerInput_.stage));
295 0 : CHK_RET(LocalNotify::Wait(
296 : subStreams_[ringIndex + 1], dispatcher_, subSignals_[ringIndex + 1], profilerInput_.stage));
297 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstSubInit, srcSubInit, subStreams_[ringIndex + 1]));
298 : } else {
299 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRing][MemcpyInitSlices] graph mode");
300 0 : CHK_RET(MemcpyInitSlicesOnMainStreams(ringIndex, dstSubInit, srcSubInit));
301 : }
302 0 : return HCCL_SUCCESS;
303 : }
304 :
305 0 : HcclResult AlignedReduceScatterDoubleRing::RunInitStep(const u32 rank, const u32 rankSize)
306 : {
307 : // 主环初始indexes
308 0 : u32 initSlice0Idx = (rankSize - rank - 1 + rankSize) % rankSize;
309 0 : u32 initSlice1Idx = (rankSize - rank - DMA_REDUCE_TWO_OFFSET + rankSize) % rankSize;
310 : // 从环初始indexes
311 0 : u32 subInitSlice0Idx = (rank + rankSize - 1) % rankSize;
312 0 : u32 subInitSlice1Idx = (rank + rankSize - DMA_REDUCE_TWO_OFFSET) % rankSize;
313 0 : u32 discontinuousSliceSize = multRingsSlices_[ALIGNED_SUB_RING_INDEX].size() / rankSize;
314 0 : DeviceMem dstInit;
315 0 : DeviceMem srcInit;
316 0 : DeviceMem dstSubInit;
317 0 : DeviceMem srcSubInit;
318 0 : DeviceMem subDstInit;
319 0 : DeviceMem subSrcInit;
320 0 : DeviceMem subDstSubInit;
321 0 : DeviceMem subSrcSubInit;
322 0 : for (u32 discontinuousSliceIdx = 0; discontinuousSliceIdx < discontinuousSliceSize; discontinuousSliceIdx++) {
323 0 : CHK_RET(PrepareInitSlices(
324 : rankSize, ALIGNED_SUB_RING_INDEX, discontinuousSliceSize, discontinuousSliceIdx, subInitSlice0Idx,
325 : subInitSlice1Idx, subDstInit, subSrcInit, subDstSubInit, subSrcSubInit));
326 0 : CHK_RET(PrepareInitSlices(
327 : rankSize, ALIGNED_MAIN_RING_INDEX, discontinuousSliceSize, discontinuousSliceIdx, initSlice0Idx,
328 : initSlice1Idx, dstInit, srcInit, dstSubInit, srcSubInit));
329 0 : HCCL_DEBUG("Memcpy operation: step[-1] starts on ring[%u]", ALIGNED_SUB_RING_INDEX);
330 0 : CHK_RET(MemcpyInitSlices(ALIGNED_SUB_RING_INDEX, subDstInit, subSrcInit, subDstSubInit, subSrcSubInit));
331 0 : HCCL_DEBUG("Memcpy operation: step[-1] starts on ring[%u]", ALIGNED_MAIN_RING_INDEX);
332 0 : CHK_RET(MemcpyInitSlices(ALIGNED_MAIN_RING_INDEX, dstInit, srcInit, dstSubInit, srcSubInit));
333 : }
334 0 : return HCCL_SUCCESS;
335 0 : }
336 :
337 : HcclResult
338 0 : AlignedReduceScatterDoubleRing::PrepareRunMainStream(u32 ringIndex, Stream& stream, LINK& preLink, LINK& nextLink)
339 : {
340 0 : HCCL_DEBUG("AlignedReduceScatterDoubleRing PrepareRunMainStream start");
341 0 : if (ringIndex == 1) {
342 0 : stream = stream_;
343 0 : preLink = rightLink_;
344 0 : nextLink = leftLink_;
345 : } else {
346 0 : stream = subStreams_[0];
347 0 : preLink = leftLink_;
348 0 : nextLink = rightLink_;
349 : }
350 0 : HCCL_DEBUG("AlignedReduceScatterDoubleRing PrepareRunMainStream end");
351 0 : return HCCL_SUCCESS;
352 : }
353 :
354 0 : HcclResult AlignedReduceScatterDoubleRing::PreSync(const u32 ringIndex)
355 : {
356 0 : if (ringIndex == 1) {
357 0 : CHK_RET(MainWaitSub());
358 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
359 0 : CHK_RET(MainRecordSub());
360 : } else {
361 0 : CHK_RET(LocalNotify::Post(subStreams_[0], dispatcher_, mainSignals_[0], profilerInput_.stage));
362 0 : CHK_RET(LocalNotify::Wait(subStreams_[0], dispatcher_, subSignals_[0], profilerInput_.stage));
363 : }
364 0 : return HCCL_SUCCESS;
365 : }
366 :
367 0 : HcclResult AlignedReduceScatterDoubleRing::PrepareDeviceMems(
368 : const u32 step, const u32 ringIndex, const u32 rankSize, const u32 txSliceIdx, const u32 rxSliceIdx,
369 : const u32 subSliceIdx, std::vector<SenderMemoryInfo>& txReduceMems, std::vector<ReducerMemoryInfo>& rxReduceMems,
370 : std::vector<DeviceMem>& localSrcMems, std::vector<DeviceMem>& localDstMems)
371 : {
372 0 : u32 sliceSize = multRingsSlices_[ringIndex].size() / rankSize;
373 0 : for (u32 sliceIdx = 0; sliceIdx < sliceSize; sliceIdx++) {
374 0 : const Slice& rxSlice = multRingsSlices_[ringIndex][rxSliceIdx * sliceSize + sliceIdx];
375 0 : const Slice& cclSlice = multRingsSlices_[ringIndex][subSliceIdx * sliceSize + sliceIdx];
376 0 : const Slice& txSlice = multRingsSlices_[ringIndex][txSliceIdx * sliceSize + sliceIdx];
377 0 : const Slice& subSlice = userMemInputSlicesOfDoubleRing_[ringIndex][subSliceIdx * sliceSize + sliceIdx];
378 : // PrepareReduceDeviceMems
379 : // Ack
380 0 : DeviceMem dst;
381 0 : if (step == rankSize - DMA_REDUCE_TWO_OFFSET && opInfo_->outputAddr != nullptr) {
382 0 : HCCL_DEBUG(
383 : "Reduce operation: step[%u] stream[main], dst rank[%u] starts to rcv offset[%llu], size[%llu] "
384 : "at userMemOut_",
385 : step, userRank_, lastStepOffsets_[ringIndex], rxSlice.size);
386 0 : dst = DeviceMem::create(static_cast<u8*>(opInfo_->outputAddr) + lastStepOffsets_[ringIndex], rxSlice.size);
387 : } else {
388 0 : HCCL_DEBUG(
389 : "Reduce operation: step[%u] stream[main], dst rank[%u] starts to rcv offset[%llu], size[%llu] "
390 : "at inputMem_",
391 : step, userRank_, rxSlice.offset, rxSlice.size);
392 0 : dst = inputMem_.range(rxSlice.offset, rxSlice.size);
393 : }
394 : // 在inline reduce场景, 需要利用scratchMem_暂存
395 0 : DeviceMem srcMemTemp = scratchMem_.range(rxSlice.offset, rxSlice.size);
396 0 : DeviceMem srcMem = inputMem_.range(txSlice.offset, txSlice.size);
397 0 : HCCL_DEBUG(
398 : "Reduce operation: step[%u] stream[main], receiver starts to rcv offset[%llu], size[%llu]", step,
399 : txSlice.offset, txSlice.size);
400 0 : rxReduceMems.emplace_back(ReducerMemoryInfo{baseOffset_ + rxSlice.offset, dst, dst, srcMemTemp});
401 0 : txReduceMems.emplace_back(SenderMemoryInfo{baseOffset_ + txSlice.offset, srcMem});
402 :
403 : // PrepareLocalCopyDeviceMems
404 0 : DeviceMem localSrt;
405 0 : DeviceMem localDst;
406 0 : if (step == rankSize - DMA_REDUCE_TWO_OFFSET) {
407 : // do nothing
408 0 : } else if (step == rankSize - DMA_REDUCE_THREE_OFFSET && opInfo_->outputAddr != nullptr) {
409 0 : HCCL_DEBUG(
410 : "Memcpy operation: step[%u] subStream[%u], src rank[%u] sends offset[%llu], size[%llu], "
411 : "dst rank[%u] starts to rcv offset[%llu], size[%llu], "
412 : "from userMemIn_ to userMemOut_",
413 : step, ringIndex + 1, userRank_, subSlice.offset, subSlice.size, userRank_, lastStepOffsets_[ringIndex],
414 : subSlice.size);
415 0 : localSrt = DeviceMem::create(static_cast<u8*>(opInfo_->inputAddr) + subSlice.offset, subSlice.size);
416 : localDst
417 0 : = DeviceMem::create(static_cast<u8*>(opInfo_->outputAddr) + lastStepOffsets_[ringIndex], subSlice.size);
418 : } else {
419 0 : HCCL_DEBUG(
420 : "Memcpy operation: step[%u] subStream[%u], src rank[%u] sends offset[%llu], size[%llu], "
421 : "dst rank[%u] starts to rcv offset[%llu], size[%llu], "
422 : "from userMemIn_ to inputMem_",
423 : step, ringIndex + 1, userRank_, subSlice.offset, subSlice.size, userRank_, cclSlice.offset,
424 : cclSlice.size);
425 0 : localSrt = DeviceMem::create(static_cast<u8*>(opInfo_->inputAddr) + subSlice.offset, subSlice.size);
426 0 : localDst = inputMem_.range(cclSlice.offset, cclSlice.size);
427 : }
428 0 : localSrcMems.emplace_back(localSrt);
429 0 : localDstMems.emplace_back(localDst);
430 0 : }
431 0 : return HCCL_SUCCESS;
432 : }
433 :
434 : HcclResult
435 0 : AlignedReduceScatterDoubleRing::RxAsyncMemcpy(const u32 ringIndex, RxMemoryInfo& mem, Stream& stream, const LINK& link)
436 : {
437 : // PreSync
438 0 : CHK_RET(PreSync(ringIndex));
439 0 : CHK_PTR_NULL(mem.dst);
440 0 : void* srcMemPtr = nullptr;
441 0 : CHK_RET(link->GetRemoteMem(mem.srcMemType, &srcMemPtr));
442 :
443 0 : DeviceMem srcDevMem(static_cast<s8*>(srcMemPtr) + mem.srcOffset, mem.len);
444 0 : DeviceMem dstDevMem(static_cast<s8*>(mem.dst), mem.len);
445 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstDevMem, srcDevMem, stream, link->GetRemoteRank(), link->GetLinkType()));
446 0 : return HCCL_SUCCESS;
447 0 : }
448 :
449 0 : HcclResult AlignedReduceScatterDoubleRing::ReducerRun(
450 : const u32 ringIndex, const HcclDispatcher dispatcher, const LINK& link, ReducerMemoryInfo& reduceMem,
451 : Stream& stream)
452 : {
453 0 : CHK_PTR_NULL(stream.ptr());
454 0 : bool isSpInlineReduce = link->IsSpInlineReduce();
455 0 : HcclResult ret = HCCL_SUCCESS;
456 0 : if (isSpInlineReduce && static_cast<bool>((INLINE_REDUCE_BITMASK & reduceAttr_))) {
457 0 : void* remoteMem = nullptr;
458 0 : CHK_RET(link->GetRemoteMem(UserMemType::INPUT_MEM, &remoteMem));
459 0 : const u64 dataBytes = reduceMem.remoteRcvTemp.size();
460 0 : CHK_RET(PreSync(ringIndex));
461 0 : CHK_RET(HcclReduceAsync(
462 : dispatcher, static_cast<s8*>(remoteMem) + reduceMem.remoteMemOffset, dataBytes / SIZE_TABLE[dataType_],
463 : dataType_, reductionOp_, stream, reduceMem.localsrc.ptr(), link->GetRemoteRank(), link->GetLinkType(),
464 : INLINE_REDUCE_BIT));
465 :
466 0 : if (reduceMem.localsrc != reduceMem.localdst) {
467 0 : ret = HcclD2DMemcpyAsync(dispatcher, reduceMem.localdst, reduceMem.localsrc, stream);
468 0 : CHK_PRT_RET(
469 : ret != HCCL_SUCCESS,
470 : HCCL_ERROR(
471 : "[AlignedReduceScatterDoubleRing][Run]memcpy_async localSrc[%p] localDst[%p] failed",
472 : reduceMem.localsrc.ptr(), reduceMem.localdst.ptr()),
473 : ret);
474 : }
475 0 : } else {
476 0 : RxMemoryInfo rxMem = RxMemoryInfo{
477 0 : UserMemType::INPUT_MEM, reduceMem.remoteMemOffset, reduceMem.remoteRcvTemp.ptr(),
478 0 : reduceMem.remoteRcvTemp.size()};
479 :
480 0 : u64 dataCount = reduceMem.localdst.size() / SIZE_TABLE[dataType_];
481 0 : DeviceMem reduceSrc = (reduceMem.localsrc == reduceMem.localdst) ? reduceMem.remoteRcvTemp : reduceMem.localsrc;
482 0 : RxWithReduceMemoryInfo rxWithReduceMem = RxWithReduceMemoryInfo{
483 : UserMemType::INPUT_MEM,
484 0 : reduceMem.remoteMemOffset,
485 0 : reduceMem.remoteRcvTemp.ptr(),
486 0 : reduceMem.remoteRcvTemp.size(),
487 0 : reduceSrc.ptr(),
488 0 : reduceMem.localdst.ptr(),
489 0 : dataCount};
490 0 : CHK_RET(RxAsyncMemcpy(ringIndex, rxMem, stream, link));
491 0 : RxWithReduceMemoryInfo& rxReduceMem = rxWithReduceMem;
492 0 : if (ringIndex == ALIGNED_SUB_RING_INDEX) {
493 0 : CHK_PRT_RET(
494 : stream != subStreams_[0],
495 : HCCL_ERROR("[%s] subStreams_[0] should be used for ringIndex=%d", __func__, ALIGNED_SUB_RING_INDEX),
496 : HCCL_E_INTERNAL);
497 0 : CHK_RET(LocalNotify::Post(subStreams_[0], dispatcher_, mainSignals_[0], profilerInput_.stage));
498 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[0], profilerInput_.stage));
499 : }
500 0 : CHK_RET(HcclReduceAsync(
501 : dispatcher, rxReduceMem.reduceSrc, rxReduceMem.reduceDataCount, dataType_, reductionOp_, stream_,
502 : rxReduceMem.reduceDst, INVALID_VALUE_RANKID, LinkType::LINK_ONCHIP, reduceAttr_));
503 0 : if (ringIndex == ALIGNED_SUB_RING_INDEX) {
504 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[0], profilerInput_.stage));
505 0 : CHK_RET(LocalNotify::Wait(subStreams_[0], dispatcher_, subSignals_[0], profilerInput_.stage));
506 : }
507 0 : }
508 0 : return HCCL_SUCCESS;
509 : }
510 :
511 0 : HcclResult AlignedReduceScatterDoubleRing::LocalMemcpy(
512 : const u32 step, const u32 rankSize, const u32 ringIndex, DeviceMem& localSrcMem, DeviceMem& localDstMem)
513 : {
514 : // 通过校验流数判断是单算子模式还是图模式
515 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB && (!disableDMAReduce_)) {
516 0 : CHK_RET(LocalNotify::Post(
517 : subStreams_[ringIndex + 1], dispatcher_, mainSignals_[ringIndex + 1], profilerInput_.stage));
518 0 : CHK_RET(LocalNotify::Wait(
519 : subStreams_[ringIndex + 1], dispatcher_, subSignals_[ringIndex + 1], profilerInput_.stage));
520 0 : if (localSrcMem != localDstMem && step != rankSize - DMA_REDUCE_TWO_OFFSET) {
521 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, localDstMem, localSrcMem, subStreams_[ringIndex + 1]));
522 : }
523 : } else {
524 : // 图模式
525 0 : CHK_RET(PreSync(ringIndex));
526 0 : if (localSrcMem != localDstMem && step != rankSize - DMA_REDUCE_TWO_OFFSET) {
527 0 : if (ringIndex == 1) {
528 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, localDstMem, localSrcMem, stream_));
529 : } else {
530 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, localDstMem, localSrcMem, subStreams_[0]));
531 : }
532 : }
533 : }
534 0 : return HCCL_SUCCESS;
535 : }
536 :
537 0 : HcclResult AlignedReduceScatterDoubleRing::RunSubStream(
538 : const u32 step, const u32 rankSize, u32 ringIndex, std::vector<DeviceMem>& localSrcMems,
539 : std::vector<DeviceMem>& localDstMems)
540 : {
541 0 : for (u32 sliceIdx = 0; sliceIdx < localSrcMems.size(); sliceIdx++) {
542 0 : CHK_RET(LocalNotify::Post(
543 : subStreams_[ringIndex + 1], dispatcher_, mainSignals_[ringIndex + 1], profilerInput_.stage));
544 0 : CHK_RET(LocalNotify::Wait(
545 : subStreams_[ringIndex + 1], dispatcher_, subSignals_[ringIndex + 1], profilerInput_.stage));
546 0 : if (step != rankSize - DMA_REDUCE_TWO_OFFSET) {
547 0 : CHK_RET(HcclD2DMemcpyAsync(
548 : dispatcher_, localDstMems[sliceIdx], localSrcMems[sliceIdx], subStreams_[ringIndex + 1]));
549 : }
550 : }
551 0 : return HCCL_SUCCESS;
552 : }
553 :
554 0 : HcclResult AlignedReduceScatterDoubleRing::RunAllStreams(
555 : const u32 step, const u32 rankSize, std::vector<SenderMemoryInfo>& mainTxReduceMems,
556 : std::vector<ReducerMemoryInfo>& mainRxReduceMems, std::vector<SenderMemoryInfo>& subTxReduceMems,
557 : std::vector<ReducerMemoryInfo>& subRxReduceMems, std::vector<DeviceMem>& mainLocalSrcMems,
558 : std::vector<DeviceMem>& mainLocalDstMems, std::vector<DeviceMem>& subLocalSrcMems,
559 : std::vector<DeviceMem>& subLocalDstMems)
560 : {
561 : (void)subTxReduceMems;
562 : (void)mainTxReduceMems;
563 0 : Stream mainStream;
564 0 : LINK mainPreLink;
565 0 : LINK mainNextLink;
566 0 : Stream subStream;
567 0 : LINK subPreLink;
568 0 : LINK subNextLink;
569 0 : CHK_RET(PrepareRunMainStream(ALIGNED_MAIN_RING_INDEX, mainStream, mainPreLink, mainNextLink));
570 0 : HCCL_DEBUG(
571 : "Reduce: step[%u] ring[%u], src rank[%u] starts to send slice to dst rank[%u]", step, ALIGNED_MAIN_RING_INDEX,
572 : mainPreLink->GetRemoteRank(), mainNextLink->GetRemoteRank());
573 0 : CHK_RET(PrepareRunMainStream(ALIGNED_SUB_RING_INDEX, subStream, subPreLink, subNextLink));
574 0 : HCCL_DEBUG(
575 : "Reduce: step[%u] ring[%u], src rank[%u] starts to send slice to dst rank[%u]", step, ALIGNED_SUB_RING_INDEX,
576 : subPreLink->GetRemoteRank(), subNextLink->GetRemoteRank());
577 :
578 0 : CHK_RET(mainNextLink->TxAck(mainStream));
579 0 : CHK_RET(mainPreLink->RxAck(mainStream));
580 0 : CHK_RET(subNextLink->TxAck(subStream));
581 0 : CHK_RET(subPreLink->RxAck(subStream));
582 :
583 0 : u32 sliceSize = multRingsSlices_[ALIGNED_MAIN_RING_INDEX].size() / rankSize;
584 0 : for (u32 memIdx = 0; memIdx < sliceSize; memIdx++) {
585 0 : CHK_RET(ReducerRun(ALIGNED_MAIN_RING_INDEX, dispatcher_, mainPreLink, mainRxReduceMems[memIdx], mainStream));
586 0 : CHK_RET(ReducerRun(ALIGNED_SUB_RING_INDEX, dispatcher_, subPreLink, subRxReduceMems[memIdx], subStream));
587 0 : CHK_RET(
588 : LocalMemcpy(step, rankSize, ALIGNED_MAIN_RING_INDEX, mainLocalSrcMems[memIdx], mainLocalDstMems[memIdx]));
589 0 : CHK_RET(LocalMemcpy(step, rankSize, ALIGNED_SUB_RING_INDEX, subLocalSrcMems[memIdx], subLocalDstMems[memIdx]));
590 : }
591 0 : CHK_RET(mainPreLink->TxDataSignal(mainStream));
592 0 : CHK_RET(mainNextLink->RxDataSignal(mainStream));
593 0 : CHK_RET(subPreLink->TxDataSignal(subStream));
594 0 : CHK_RET(subNextLink->RxDataSignal(subStream));
595 0 : return HCCL_SUCCESS;
596 0 : }
597 :
598 0 : HcclResult AlignedReduceScatterDoubleRing::PreRunStreams(
599 : const u32 step, const u32 rankSize, const u32 txSliceIdxMain, const u32 rxSliceIdxMain, const u32 subSliceIdxMain,
600 : const u32 txSliceIdxSub, const u32 rxSliceIdxSub, const u32 subSliceIdxSub,
601 : std::vector<SenderMemoryInfo>& txReduceMemsMain, std::vector<ReducerMemoryInfo>& rxReduceMemsMain,
602 : std::vector<SenderMemoryInfo>& txReduceMemsSub, std::vector<ReducerMemoryInfo>& rxReduceMemsSub,
603 : std::vector<DeviceMem>& localSrcMemsMain, std::vector<DeviceMem>& localDstMemsMain,
604 : std::vector<DeviceMem>& localSrcMemsSub, std::vector<DeviceMem>& localDstMemsSub)
605 : {
606 0 : CHK_RET(PrepareDeviceMems(
607 : step, ALIGNED_MAIN_RING_INDEX, rankSize, txSliceIdxMain, rxSliceIdxMain, subSliceIdxMain, txReduceMemsMain,
608 : rxReduceMemsMain, localSrcMemsMain, localDstMemsMain));
609 0 : CHK_RET(PrepareDeviceMems(
610 : step, ALIGNED_SUB_RING_INDEX, rankSize, txSliceIdxSub, rxSliceIdxSub, subSliceIdxSub, txReduceMemsSub,
611 : rxReduceMemsSub, localSrcMemsSub, localDstMemsSub));
612 0 : return HCCL_SUCCESS;
613 : }
614 :
615 0 : HcclResult AlignedReduceScatterDoubleRing::RunReduceScatter(const u32 rank, const u32 rankSize)
616 : {
617 0 : HCCL_INFO("AlignedReduceScatterDoubleRing starts, the input param rank[%u]", rank);
618 : // 空拷贝用于后续操作附着
619 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
620 : // 主环主流通知从环主流开始通信
621 0 : CHK_RET(MainRecordSub());
622 : // 从环主流等待主环主流通知
623 0 : CHK_RET(SubWaitMain());
624 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
625 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
626 0 : CHK_RET(ExecEmptyTasks());
627 0 : CHK_RET(RunInitStep(rank, rankSize));
628 : }
629 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
630 0 : CHK_RET(ExecEmptyTasks());
631 :
632 : // 例如rank[0,1,2,3]中，rank0的rxSliceIdx = 2，txSliceIdx = 3, subSliceIdx = 1
633 : // 从环初始indexes
634 0 : u32 txSliceIdxSub = (rank + rankSize - 1) % rankSize;
635 0 : u32 rxSliceIdxSub = (rank + rankSize - DMA_REDUCE_TWO_OFFSET) % rankSize;
636 0 : u32 subSliceIdxSub = (rank + rankSize - DMA_REDUCE_THREE_OFFSET) % rankSize;
637 0 : HCCL_DEBUG(
638 : "[RunReduceScatter]txSliceIdxSub is [%u], rxSliceIdxSub is [%u], subSliceIdxSub is [%u]", txSliceIdxSub,
639 : rxSliceIdxSub, subSliceIdxSub);
640 : // 主环初始indexes
641 0 : u32 txSliceIdxMain = (rankSize - rank - 1 + rankSize) % rankSize;
642 0 : u32 rxSliceIdxMain = (rankSize - rank - DMA_REDUCE_TWO_OFFSET + rankSize) % rankSize;
643 0 : u32 subSliceIdxMain = (rankSize - rank - DMA_REDUCE_THREE_OFFSET + rankSize) % rankSize;
644 :
645 0 : for (u32 step = 0; step < rankSize - 1; step++) {
646 : // 并发
647 0 : std::vector<SenderMemoryInfo> txReduceMemsMain;
648 0 : std::vector<ReducerMemoryInfo> rxReduceMemsMain;
649 0 : std::vector<SenderMemoryInfo> txReduceMemsSub;
650 0 : std::vector<ReducerMemoryInfo> rxReduceMemsSub;
651 0 : std::vector<DeviceMem> localDstMemsMain;
652 0 : std::vector<DeviceMem> localSrcMemsMain;
653 0 : std::vector<DeviceMem> localSrcMemsSub;
654 0 : std::vector<DeviceMem> localDstMemsSub;
655 0 : CHK_RET(PreRunStreams(
656 : step, rankSize, txSliceIdxMain, rxSliceIdxMain, subSliceIdxMain, txSliceIdxSub, rxSliceIdxSub,
657 : subSliceIdxSub, txReduceMemsMain, rxReduceMemsMain, txReduceMemsSub, rxReduceMemsSub, localSrcMemsMain,
658 : localDstMemsMain, localSrcMemsSub, localDstMemsSub));
659 0 : CHK_RET(RunAllStreams(
660 : step, rankSize, txReduceMemsMain, rxReduceMemsMain, txReduceMemsSub, rxReduceMemsSub, localSrcMemsMain,
661 : localDstMemsMain, localSrcMemsSub, localDstMemsSub));
662 : // 更新索引
663 0 : txSliceIdxSub = (txSliceIdxSub + rankSize - 1) % rankSize;
664 0 : rxSliceIdxSub = (rxSliceIdxSub + rankSize - 1) % rankSize;
665 0 : subSliceIdxSub = (subSliceIdxSub + rankSize - 1) % rankSize;
666 0 : txSliceIdxMain = (txSliceIdxMain + rankSize - 1) % rankSize;
667 0 : rxSliceIdxMain = (rxSliceIdxMain + rankSize - 1) % rankSize;
668 0 : subSliceIdxMain = (subSliceIdxMain + rankSize - 1) % rankSize;
669 0 : }
670 : // 从环主流通知主环主流通信完成
671 0 : CHK_RET(SubRecordMain());
672 : // 主环主流等待从环主流通知
673 0 : CHK_RET(MainWaitSub());
674 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
675 0 : CHK_RET(ExecEmptyTasks());
676 0 : HCCL_INFO("AlignedReduceScatterDoubleRing finished to RunReduceScatter");
677 0 : return HCCL_SUCCESS;
678 : }
679 :
680 0 : HcclResult AlignedReduceScatterDoubleRing::GetActiveSubstreamNum(u32& activeSubstreamNum)
681 : {
682 0 : activeSubstreamNum = subStreams_.size();
683 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB && disableDMAReduce_) {
684 0 : if (subStreams_.size() <= 2) {
685 0 : HCCL_ERROR(
686 : "[AlignedReduceScatterDoubleRing][GetActiveSubstreamNum]subStreams_.size()[%zu] <= 2",
687 : subStreams_.size());
688 0 : return HCCL_E_PARA;
689 : }
690 0 : activeSubstreamNum = subStreams_.size() - 2;
691 : }
692 0 : return HCCL_SUCCESS;
693 : }
694 :
695 0 : HcclResult AlignedReduceScatterDoubleRing::ExecEmptyTasks()
696 : {
697 0 : u32 activeSubstreamNum = 0;
698 0 : CHK_RET(GetActiveSubstreamNum(activeSubstreamNum));
699 0 : for (u32 signalIndex = 0; signalIndex < activeSubstreamNum; signalIndex++) {
700 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, subStreams_[signalIndex], dispatcher_));
701 : }
702 0 : return HCCL_SUCCESS;
703 : }
704 :
705 : // 主流通知从流干活
706 0 : HcclResult AlignedReduceScatterDoubleRing::MainRecordSub()
707 : {
708 0 : u32 activeSubstreamNum = 0;
709 0 : CHK_RET(GetActiveSubstreamNum(activeSubstreamNum));
710 0 : for (u32 signalIndex = 0; signalIndex < activeSubstreamNum; signalIndex++) {
711 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[signalIndex], profilerInput_.stage));
712 : }
713 0 : return HCCL_SUCCESS;
714 : }
715 : // 从流等待主流
716 0 : HcclResult AlignedReduceScatterDoubleRing::SubWaitMain()
717 : {
718 0 : u32 activeSubstreamNum = 0;
719 0 : CHK_RET(GetActiveSubstreamNum(activeSubstreamNum));
720 0 : for (u32 streamIndex = 0; streamIndex < activeSubstreamNum; streamIndex++) {
721 0 : CHK_RET(
722 : LocalNotify::Wait(subStreams_[streamIndex], dispatcher_, subSignals_[streamIndex], profilerInput_.stage));
723 : }
724 0 : return HCCL_SUCCESS;
725 : }
726 : // 主流等待从流
727 0 : HcclResult AlignedReduceScatterDoubleRing::MainWaitSub()
728 : {
729 0 : u32 activeSubstreamNum = 0;
730 0 : CHK_RET(GetActiveSubstreamNum(activeSubstreamNum));
731 0 : for (u32 signalIndex = 0; signalIndex < activeSubstreamNum; signalIndex++) {
732 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[signalIndex], profilerInput_.stage));
733 : }
734 0 : return HCCL_SUCCESS;
735 : }
736 : // 从流告诉主流活干完了
737 0 : HcclResult AlignedReduceScatterDoubleRing::SubRecordMain()
738 : {
739 0 : u32 activeSubstreamNum = 0;
740 0 : CHK_RET(GetActiveSubstreamNum(activeSubstreamNum));
741 0 : for (u32 streamIndex = 0; streamIndex < activeSubstreamNum; streamIndex++) {
742 0 : CHK_RET(
743 : LocalNotify::Post(subStreams_[streamIndex], dispatcher_, mainSignals_[streamIndex], profilerInput_.stage));
744 : }
745 0 : return HCCL_SUCCESS;
746 : }
747 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_REDUCESCATTER_DB_RING, AlignedReduceScatterDoubleRing);
748 : } // namespace hccl
|
